mapping fix for labeller

* added cache

* changes

* added select to make original order

* added cache

* Case normalize (zinggAI#1027)

* added Case normalizer preprocessor

* removed toLowerCase() in sim call()

* fixed junits

* added junits for case normalizer

* added for spark driver memory in spark session builder

* added log

* added log

* logged memory in GB

* abstracted out stopWord files names

* added logging

* added exception

* made join on both id and cluster

* test not needed as we are already case normalizing at start

Author: Nitish1814

common/client/src/main/java/zingg/common/client/util/DSUtil.java

@@ -271,17 +271,28 @@ public ZFrame<D,R,C> postprocess(ZFrame<D,R,C> actual, ZFrame<D,R,C> orig) {
 	public ZFrame<D,R,C> postProcessLabel(ZFrame<D,R,C> updatedLabelledRecords, ZFrame<D,R,C> unmarkedRecords) {
 		List<C> cols = new ArrayList<C>();
 		cols.add(updatedLabelledRecords.col(ColName.ID_COL));
+		cols.add(updatedLabelledRecords.col(ColName.CLUSTER_COLUMN));
 
 		String[] unmarkedRecordColumns = unmarkedRecords.columns();
 
 		//drop isMatch column from unMarked records
 		//and replace with updated isMatch column
 		cols.add(updatedLabelledRecords.col(ColName.MATCH_FLAG_COL));
-		ZFrame<D,R,C> zFieldsFromUpdatedLabelledRecords = updatedLabelledRecords.select(cols);
+		ZFrame<D,R,C> zFieldsFromUpdatedLabelledRecords = updatedLabelledRecords.select(cols).
+				withColumnRenamed(ColName.ID_COL, ColName.COL_PREFIX + ColName.ID_COL).
+				withColumnRenamed(ColName.CLUSTER_COLUMN, ColName.COL_PREFIX + ColName.CLUSTER_COLUMN);
+
 		unmarkedRecords = unmarkedRecords.drop(ColName.MATCH_FLAG_COL);
 
+		/*
+			join on z_id and z_cluster
+		 */
+		C joinCondition1 = unmarkedRecords.equalTo(unmarkedRecords.col(ColName.ID_COL), zFieldsFromUpdatedLabelledRecords.col(ColName.COL_PREFIX + ColName.ID_COL));
+		C joinCondition2 = unmarkedRecords.equalTo(unmarkedRecords.col(ColName.CLUSTER_COLUMN), zFieldsFromUpdatedLabelledRecords.col(ColName.COL_PREFIX + ColName.CLUSTER_COLUMN));
+		C joinCondition = unmarkedRecords.and(joinCondition1, joinCondition2);
+
 		//we are selecting columns to bring back to original shape
-		return unmarkedRecords.joinOnCol(zFieldsFromUpdatedLabelledRecords, ColName.ID_COL).select(unmarkedRecordColumns);
+		return unmarkedRecords.join(zFieldsFromUpdatedLabelledRecords, joinCondition, "inner").select(unmarkedRecordColumns);
 	}
 
 

common/core/src/main/java/zingg/common/core/executor/Trainer.java

@@ -31,7 +31,6 @@ public void execute() throws ZinggClientException {
 			ZFrame<D,R,C> traOriginal = getDSUtil().getTraining(getPipeUtil(), args, getModelHelper());
 			ZFrame<D,R,C> tra = preprocess(traOriginal).cache();
 			tra = getDSUtil().joinWithItself(tra, ColName.CLUSTER_COLUMN, true);
-			tra = tra.cache();
 			positives = tra.filter(tra.equalTo(ColName.MATCH_FLAG_COL,ColValues.MATCH_TYPE_MATCH));
 			negatives = tra.filter(tra.equalTo(ColName.MATCH_FLAG_COL,ColValues.MATCH_TYPE_NOT_A_MATCH));
 

common/core/src/main/java/zingg/common/core/similarity/function/OnlyAlphabetsExactSimilarity.java

@@ -36,7 +36,7 @@ public Double call(String first, String second) {
 			if (score1 != 1.0d && score2 != 1.0d) {
 				first = first.replaceAll("[0-9.]", "");
 				second = second.replaceAll("[0-9.]", "");
-				score = first.equalsIgnoreCase(second)? 1.0d : 0.0d;
+				score = first.equals(second)? 1.0d : 0.0d;
 			}
 			else {
 				score = 1.0d;

common/core/src/main/java/zingg/common/core/similarity/function/PinCodeMatchTypeFunction.java

@@ -25,7 +25,7 @@ public Double call(String first, String second) {
 		if (second == null || second.trim().length() ==0) return 1d;
 		first = first.split("-")[0];
 		second = second.split("-")[0];
-		double score = first.trim().equalsIgnoreCase(second.trim()) ? 1d : 0d;
+		double score = first.trim().equals(second.trim()) ? 1d : 0d;
 		return score;		
 	}	
 }

common/core/src/main/java/zingg/common/core/similarity/function/StringSimilarityDistanceFunction.java

@@ -26,7 +26,7 @@ public AbstractStringDistance getDistanceFunction(){
 	public Double call(String first, String second) {
 		if (first == null || first.trim().length() ==0) return 1d;
 		if (second == null || second.trim().length() ==0) return 1d;
-		if (first.equalsIgnoreCase(second)) return 1d;
+		if (first.equals(second)) return 1d;
 		double score = getDistanceFunction().score(first, second);
 		if (Double.isNaN(score)) return 0d; 
 		//LOG.warn(" score  " + gap +  " " + first + " " + second + " is " + score);

common/core/src/main/java/zingg/common/core/similarity/function/StringSimilarityFunction.java

@@ -22,7 +22,7 @@ public StringSimilarityFunction(String name) {
 	public Double call(String first, String second) {
 		if (first == null || first.trim().length() ==0) return 1d;
 		if (second == null || second.trim().length() ==0) return 1d;
-		double score = first.trim().equalsIgnoreCase(second.trim()) ? 1d : 0d;
+		double score = first.trim().equals(second.trim()) ? 1d : 0d;
 		return score;		
 	}
 

common/core/src/test/java/zingg/common/core/similarity/function/TestOnlyAlphabetsExactSimilarity.java

@@ -33,9 +33,4 @@ public void testDiffNoNumber() {
 		assertEquals(0d, sim.call("I have a no number", "I have r number"));
 	}	
 
-	@Test
-	public void testSameIgnoreCase() {
-		OnlyAlphabetsExactSimilarity sim = new OnlyAlphabetsExactSimilarity();
-		assertEquals(1d, sim.call("I have 1 number", "I HAVE 2 number"));
-	}
 }

spark/core/src/main/java/zingg/spark/core/executor/SparkMatcher.java

@@ -25,7 +25,7 @@
  * 
  *
  */
-public class SparkMatcher extends Matcher<SparkSession,Dataset<Row>,Row,Column,DataType> implements ISparkPreprocMapSupplier{
+public class SparkMatcher extends Matcher<SparkSession,Dataset<Row>,Row,Column,DataType> implements ISparkPreprocMapSupplier {
 
 
 	private static final long serialVersionUID = 1L;

spark/core/src/main/java/zingg/spark/core/preprocess/caseNormalize/SparkCaseNormalizer.java

@@ -31,6 +31,7 @@ public SparkCaseNormalizer(IContext<SparkSession, Dataset<Row>, Row, Column, Dat
 
     @Override
     protected ZFrame<Dataset<Row>, Row, Column> applyCaseNormalizer(ZFrame<Dataset<Row>, Row, Column> incomingDataFrame, List<String> relevantFields) {
+        String[] incomingDFColumns = incomingDataFrame.columns();
         Seq<String> columnsSeq = JavaConverters.asScalaIteratorConverter(relevantFields.iterator())
                 .asScala()
                 .toSeq();
@@ -41,6 +42,6 @@ protected ZFrame<Dataset<Row>, Row, Column> applyCaseNormalizer(ZFrame<Dataset<R
         Seq<Column> caseNormalizedSeq = JavaConverters.asScalaIteratorConverter(caseNormalizedValues.iterator())
                 .asScala()
                 .toSeq();
-        return new SparkFrame(incomingDataFrame.df().withColumns(columnsSeq, caseNormalizedSeq));
+        return new SparkFrame(incomingDataFrame.df().withColumns(columnsSeq, caseNormalizedSeq)).select(incomingDFColumns);
     }
 }

spark/core/src/test/java/zingg/spark/core/executor/labeller/ProgrammaticSparkLabeller.java

@@ -13,8 +13,9 @@
 import zingg.common.core.executor.labeller.ProgrammaticLabeller;
 import zingg.spark.core.context.ZinggSparkContext;
 import zingg.spark.core.executor.SparkLabeller;
+import zingg.spark.core.preprocess.ISparkPreprocMapSupplier;
 
-public class ProgrammaticSparkLabeller extends SparkLabeller {
+public class ProgrammaticSparkLabeller extends SparkLabeller implements ISparkPreprocMapSupplier {
 
 	private static final long serialVersionUID = 1L;