diff --git a/common/core/src/main/java/zingg/common/core/preprocess/stopwords/RemoveStopWords.java b/common/core/src/main/java/zingg/common/core/preprocess/stopwords/RemoveStopWords.java new file mode 100644 index 000000000..f0c95e1aa --- /dev/null +++ b/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; + } + +} diff --git a/common/core/src/main/java/zingg/common/core/preprocess/stopwords/StopWords.java b/common/core/src/main/java/zingg/common/core/preprocess/stopwords/StopWords.java new file mode 100644 index 000000000..e7b6407d2 --- /dev/null +++ b/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 { + + 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 pipeUtil; + + public PipeUtilBase getPipeUtil() { + return pipeUtil; + } + + public void setPipeUtil(PipeUtilBase pipeUtil) { + this.pipeUtil = pipeUtil; + } + + + + public ZFrame preprocessForStopWords(S session, IArguments args, ZFrame ds) throws ZinggClientException { + /* + List wordList = new ArrayList(); + for (FieldDefinition def : args.getFieldDefinition()) { + if (!(def.getStopWords() == null || def.getStopWords() == "")) { + ZFrame 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); + } + */ +} diff --git a/spark/core/src/main/java/zingg/spark/core/preprocess/stopwords/RemoveStopWordsUDF.java b/spark/core/src/main/java/zingg/spark/core/preprocess/stopwords/RemoveStopWordsUDF.java new file mode 100644 index 000000000..2e9943b98 --- /dev/null +++ b/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{ + + private static final long serialVersionUID = 1L; + + public RemoveStopWordsUDF() { + super(); + } + + @Override + public String call(String s,String stopWordsRegexString) throws Exception { + return removeStopWordsUsingRegex(s,stopWordsRegexString); + } + +} + diff --git a/spark/core/src/main/java/zingg/spark/core/preprocess/stopwords/SparkStopWordsRemover.java b/spark/core/src/main/java/zingg/spark/core/preprocess/stopwords/SparkStopWordsRemover.java index 36c7dc9b1..dcac47762 100644 --- a/spark/core/src/main/java/zingg/spark/core/preprocess/stopwords/SparkStopWordsRemover.java +++ b/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,6 +11,7 @@ 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; @@ -17,34 +19,50 @@ 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,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, Row, Column,DataType> context) { super(context); + registerUDF(); } public SparkStopWordsRemover(IContext, Row, Column,DataType> context, FieldDefinition fd) { super(context,fd); + registerUDF(); } @Override protected ZFrame, Row, Column> removeStopWordsFromDF(ZFrame, Row, Column> ds, String fieldName, String pattern) { - Dataset dfAfterRemoval = ds.df().withColumn(fieldName, regexp_replace(ds.df().col(fieldName), pattern, "")); + Dataset 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(); } }