Revert "Use native Spark regexp_replace for stopword removal instead of a UDF…"

This reverts commit 4f067fd.

Author: sonalgoyal

common/core/src/main/java/zingg/common/core/preprocess/stopwords/RemoveStopWords.java

@@ -0,0 +1,27 @@
+package zingg.common.core.preprocess.stopwords;
+
+import java.io.Serializable;
+
+public class RemoveStopWords implements Serializable {
+	
+	private static final long serialVersionUID = 1L;
+	private String name = "removeStopWordsUDF";
+	
+	public RemoveStopWords() {
+		super();
+	}
+
+	protected String removeStopWordsUsingRegex(String s,String stopWordsRegexString) {
+		if (s == null || stopWordsRegexString==null) return null;
+		return s.replaceAll(stopWordsRegexString, "");
+	}
+	
+	public String getName() {
+		return name;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+	
+}

common/core/src/main/java/zingg/common/core/preprocess/stopwords/StopWords.java

@@ -0,0 +1,58 @@
+package zingg.common.core.preprocess.stopwords;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import zingg.common.client.arguments.model.IArguments;
+import zingg.common.client.ZFrame;
+import zingg.common.client.ZinggClientException;
+import zingg.common.client.util.ColName;
+import zingg.common.client.util.PipeUtilBase;
+
+public class StopWords<S,D,R,C,T> {
+
+	protected static String name = "zingg.preprocess.stopwords.StopWords";
+	public static final Log LOG = LogFactory.getLog(StopWords.class);
+	protected static String stopWordColumn = ColName.COL_WORD;
+	protected static final int COLUMN_INDEX_DEFAULT = 0;
+	protected PipeUtilBase<S,D,R,C> pipeUtil;
+
+	public PipeUtilBase<S, D, R, C> getPipeUtil() {
+		return pipeUtil;
+	}
+
+	public void setPipeUtil(PipeUtilBase<S, D, R, C> pipeUtil) {
+		this.pipeUtil = pipeUtil;
+	}
+
+	
+
+    public ZFrame<D,R,C> preprocessForStopWords(S session, IArguments args, ZFrame<D,R,C> ds) throws ZinggClientException {
+		/* 
+		List<String> wordList = new ArrayList<String>();
+		for (FieldDefinition def : args.getFieldDefinition()) {
+			if (!(def.getStopWords() == null || def.getStopWords() == "")) {
+				ZFrame<D,R,C> stopWords = getPipeUtil().read(false, false, getPipeUtil().getStopWordsPipe(args, def.getStopWords()));
+				//if (!Arrays.asList(stopWords.schema().fieldNames()).contains(stopWordColumn)) {
+					stopWordColumn = stopWords.columns()[COLUMN_INDEX_DEFAULT];
+				//}
+				wordList = stopWords.select(stopWordColumn).as(Encoders.STRING()).collectAsList();
+				String pattern = wordList.stream().collect(Collectors.joining("|", "\\b(", ")\\b\\s?"));
+				ds = ds.withColumn(def.getFieldName(), removeStopWords(pattern.toLowerCase()).apply(ds.col(def.getFieldName())));
+			}
+		}
+
+		return ds;
+		*/
+		return ds;
+	}
+	
+	/* 
+	public static UserDefinedFunction removeStopWords(String stopWordsRegexString) {
+		return udf((String s) -> {
+				if (s == null) return null;
+				return s.toLowerCase().replaceAll(stopWordsRegexString, "");
+			}, DataTypes.StringType);
+	}
+	*/
+}

spark/core/src/main/java/zingg/spark/core/preprocess/stopwords/RemoveStopWordsUDF.java

@@ -0,0 +1,21 @@
+package zingg.spark.core.preprocess.stopwords;
+
+import org.apache.spark.sql.api.java.UDF2;
+
+import zingg.common.core.preprocess.stopwords.RemoveStopWords;
+
+public class RemoveStopWordsUDF extends RemoveStopWords implements UDF2<String,String,String>{
+	
+	private static final long serialVersionUID = 1L;
+	
+	public RemoveStopWordsUDF() {
+		super();
+	}
+	
+	@Override
+	public String call(String s,String stopWordsRegexString) throws Exception {
+		return removeStopWordsUsingRegex(s,stopWordsRegexString);
+	}
+	
+}
+

spark/core/src/main/java/zingg/spark/core/preprocess/stopwords/SparkStopWordsRemover.java

@@ -1,6 +1,7 @@
 package zingg.spark.core.preprocess.stopwords;
 
-import static org.apache.spark.sql.functions.regexp_replace;
+import static org.apache.spark.sql.functions.callUDF;
+import static org.apache.spark.sql.functions.lit;
 
 import java.io.Serializable;
 
@@ -10,41 +11,58 @@
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DataTypes;
 
 import zingg.common.client.FieldDefinition;
 import zingg.common.client.ZFrame;
 import zingg.common.core.context.IContext;
 import zingg.common.core.preprocess.stopwords.StopWordsRemover;
 import zingg.spark.client.SparkFrame;
 import org.apache.spark.sql.SparkSession;
+import zingg.spark.core.util.SparkFnRegistrar;
 
 public class SparkStopWordsRemover extends StopWordsRemover<SparkSession,Dataset<Row>,Row,Column,DataType>  implements Serializable {
 
 	private static final long serialVersionUID = 1L;
 	protected static String name = "zingg.spark.preprocess.SparkStopWordsRemover";
 	public static final Log LOG = LogFactory.getLog(SparkStopWordsRemover.class);
+	
+	private String udfName;
 
 	public SparkStopWordsRemover(){
     }
 
 	public SparkStopWordsRemover(IContext<SparkSession, Dataset<Row>, Row, Column,DataType> context) {
 		super(context);
+		registerUDF();
 	}
 
 	public SparkStopWordsRemover(IContext<SparkSession, Dataset<Row>, Row, Column,DataType> context, FieldDefinition fd) {
 		super(context,fd);
+		registerUDF();
 	}
 
  	@Override
 	protected ZFrame<Dataset<Row>, Row, Column> removeStopWordsFromDF(ZFrame<Dataset<Row>, Row, Column> ds,
 			String fieldName, String pattern) {
-		Dataset<Row> dfAfterRemoval = ds.df().withColumn(fieldName, regexp_replace(ds.df().col(fieldName), pattern, ""));
+		Dataset<Row> dfAfterRemoval = ds.df().withColumn(fieldName,callUDF(udfName, ds.df().col(fieldName),lit(pattern)));
 
 		return new SparkFrame(dfAfterRemoval);
 	}
 
+	protected void registerUDF() {
+		RemoveStopWordsUDF removeStopWordsUDF = new RemoveStopWordsUDF();
+		// Each field will have different pattern
+		this.udfName = removeStopWordsUDF.getName();
+		// register the UDF
+		SparkSession zSession = getContext().getSession();
+
+		SparkFnRegistrar.registerUDF2(zSession, udfName, removeStopWordsUDF, DataTypes.StringType);
+	}
+
 	@Override
 	public void init() {
+		registerUDF();
 	}
 
 }