Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Stopword fix PR #1008

Closed
wants to merge 6 commits into from
Closed

Stopword fix PR #1008

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4a1eaa9
Verify blocking (#1007)
sonalgoyal Jan 17, 2025
67cf7d1
telemetry fix
sania-16 Jan 18, 2025
f975379
report generated
Jan 19, 2025
3db27e3
Merge branch 'zinggAI:main' into stopword1002
sania-16 Jan 20, 2025
54fe24e
comment fixes
sania-16 Jan 20, 2025
c530598
comments fix
sania-16 Jan 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion common/client/src/main/java/zingg/common/client/pipe/Pipe.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ public class Pipe<D,R,C> implements Serializable{ // St:StructType, Sv:SaveMode
String name;
String format;
String preprocessors;
Map<String, String> props = new HashMap<String, String>();
protected Map<String, String> props = new HashMap<String, String>();
int id;
protected ZFrame<D, R, C> dataset;
String schema;
Expand Down
7 changes: 7 additions & 0 deletions common/client/src/main/java/zingg/common/client/util/DSUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -253,6 +253,13 @@ public List<FieldDefinition> getFieldDefinitionFiltered(IArguments args, MatchT
.collect(Collectors.toList());
}

public List<FieldDefinition> getFieldDefinitionWithStopwords(IArguments args) {
sania-16 marked this conversation as resolved.
Show resolved Hide resolved
return args.getFieldDefinition()
.stream()
.filter(f -> !(f.getStopWords() == null || f.getStopWords() == ""))
sania-16 marked this conversation as resolved.
Show resolved Hide resolved
.collect(Collectors.toList());
}

public ZFrame<D,R,C> postprocess(ZFrame<D,R,C> actual, ZFrame<D,R,C> orig) {
List<C> cols = new ArrayList<C>();
cols.add(actual.col(ColName.CLUSTER_COLUMN));
Expand Down
5 changes: 3 additions & 2 deletions common/core/src/main/java/zingg/common/core/executor/ZinggBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,14 +73,15 @@ public void setSession(S s) {



public void track( boolean collectMetrics){
public void track(boolean collectMetrics){
Analytics.track(Metric.TOTAL_FIELDS_COUNT, args.getFieldDefinition().size(), collectMetrics);
Analytics.track(Metric.MATCH_FIELDS_COUNT, getDSUtil().getFieldDefinitionFiltered(args, MatchType.DONT_USE).size(),
collectMetrics);
Analytics.track(Metric.DATA_FORMAT, getPipeUtil().getPipesAsString(args.getData()), collectMetrics);
Analytics.track(Metric.OUTPUT_FORMAT, getPipeUtil().getPipesAsString(args.getOutput()), collectMetrics);
Analytics.track(Metric.MODEL_ID, args.getModelId(), collectMetrics);

Analytics.track(Metric.STOPWORDS, getDSUtil().getFieldDefinitionWithStopwords(args).size(),
sania-16 marked this conversation as resolved.
Show resolved Hide resolved
collectMetrics);

}

Expand Down
3 changes: 2 additions & 1 deletion common/core/src/main/java/zingg/common/core/executor/blockingverifier/VerifyBlocking.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,8 @@ public void execute() throws ZinggClientException {
LOG.info("Blocked");

ZFrame<D,R,C> blockCounts = blocked.select(ColName.HASH_COL).groupByCount(ColName.HASH_COL, ColName.HASH_COUNTS_COL).sortDescending(ColName.HASH_COUNTS_COL);

blockCounts = blockCounts.cache();
sania-16 marked this conversation as resolved.
Show resolved Hide resolved

getPipeUtil().write(blockCounts,getVerifyBlockingPipeUtil().getCountsPipe(args));

ZFrame<D,R,C> blockTopRec = blockCounts.select(ColName.HASH_COL,ColName.HASH_COUNTS_COL).limit(noOfBlocks);
Expand Down
1 change: 1 addition & 0 deletions common/core/src/main/java/zingg/common/core/util/Metric.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ public class Metric {
public static final String TRAINING_MATCHES = "trainingDataMatches";
public static final String TRAINING_NONMATCHES = "trainingDataNonmatches";
public static final String DATA_COUNT = "dataCount";
public static final String STOPWORDS = "stopWords";

public static final long timeout = 1200L;
public static final double confidence = 0.95; // default value
Expand Down
16 changes: 14 additions & 2 deletions spark/client/src/main/java/zingg/spark/client/pipe/SparkPipe.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,12 @@
package zingg.spark.client.pipe;

import java.io.IOException;

import org.apache.spark.sql.Column;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;

import org.apache.spark.sql.Dataset;

import zingg.common.client.pipe.FilePipe;
import zingg.common.client.pipe.Pipe;


Expand Down Expand Up @@ -53,5 +53,17 @@ public void setOverwriteMode() {
setMode(SaveMode.Overwrite.toString());
}

public static String massageLocation(String name){
name = name.replaceAll("-", "");
name = name.replaceAll("@","");
name = name.replaceAll(",","");
name = name.replaceAll(":","");
return name;
}

public void setLocation(String fileName){
this.props.put(FilePipe.LOCATION, massageLocation(fileName));
}


}
3 changes: 0 additions & 3 deletions spark/client/src/main/java/zingg/spark/client/util/SparkModelHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,8 +151,5 @@ public Pipe<Dataset<Row>, Row, Column> getStopWordsPipe(String fileName) {
p.setOverwriteMode();
return p;
}




}
3 changes: 0 additions & 3 deletions spark/client/src/main/java/zingg/spark/client/util/SparkPipeUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,11 @@
import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;

import zingg.common.client.ZFrame;
//import zingg.common.client.pipe.InMemoryPipe;
import zingg.common.client.pipe.Pipe;
import zingg.common.client.util.DFReader;
import zingg.common.client.util.DFWriter;
import zingg.common.client.util.IModelHelper;
import zingg.common.client.util.PipeUtil;
import zingg.spark.client.SparkFrame;
import org.apache.spark.sql.SparkSession;
Expand Down
34 changes: 34 additions & 0 deletions spark/client/src/test/java/zingg/spark/client/TestSparkPipe.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
package zingg.spark.client;

import static org.junit.jupiter.api.Assertions.assertEquals;

import java.util.HashMap;
import java.util.Map;

import org.junit.jupiter.api.Test;

import zingg.spark.client.pipe.SparkPipe;

public class TestSparkPipe {

@Test
public void testMassageLocation(){
String input = "zin:gg/spar,k/client/blo@cks/-1234";
String expected = "zingg/spark/client/blocks/1234";
assertEquals(expected, SparkPipe.massageLocation(input));
}

@Test
public void testSetLocation(){
String input = "zin:gg/spar,k/client/blo@cks/-1234";
String exp = "zingg/spark/client/blocks/1234";
Map<String, String> expProps = new HashMap<String, String>();
expProps.put("location", exp);
SparkPipe p = new SparkPipe();
p.setLocation(input);
Map<String, String> props = new HashMap<String, String>();
props = p.getProps();
assertEquals(props.get("location"), expProps.get("location"));
}

}
4 changes: 1 addition & 3 deletions spark/core/src/main/java/zingg/spark/core/executor/SparkVerifyBlockingPipes.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,11 @@
package zingg.spark.core.executor;

import org.apache.spark.internal.config.R;
import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import zingg.common.client.IArguments;
import zingg.common.client.pipe.FilePipe;
import zingg.common.client.pipe.Pipe;
import zingg.common.client.util.IModelHelper;
import zingg.common.client.util.PipeUtilBase;
Expand All @@ -24,7 +22,7 @@ public SparkVerifyBlockingPipes(PipeUtilBase<SparkSession,Dataset<Row>,Row,Colum
public Pipe<Dataset<Row>,Row,Column> getPipeForVerifyBlockingLocation(IArguments args, String type){
SparkPipe p = new SparkPipe();
p.setFormat(Pipe.FORMAT_PARQUET);
p.setProp(FilePipe.LOCATION, getName(args,timestamp,type));
p.setLocation(getName(args,timestamp,type));
p.setOverwriteMode();
return p;
}
Expand Down
3 changes: 0 additions & 3 deletions spark/core/src/test/java/zingg/spark/core/util/TestUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,10 @@
package zingg.spark.core.util;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.util.ArrayList;
import java.util.List;

import org.apache.spark.sql.types.DataType;
import org.junit.jupiter.api.Test;

import zingg.common.client.util.*;
Expand Down
Loading