From cb8054056b631bd32372e146e9b9a6a257d79783 Mon Sep 17 00:00:00 2001 From: chandan Date: Mon, 6 Jul 2026 22:43:33 +0530 Subject: [PATCH] Use native Spark regexp_replace for stopword removal instead of a UDF Fixes #1080 --- .../preprocess/stopwords/RemoveStopWords.java | 27 --------- .../core/preprocess/stopwords/StopWords.java | 58 ------------------- .../stopwords/RemoveStopWordsUDF.java | 21 ------- .../stopwords/SparkStopWordsRemover.java | 22 +------ 4 files changed, 2 insertions(+), 126 deletions(-) delete mode 100644 common/core/src/main/java/zingg/common/core/preprocess/stopwords/RemoveStopWords.java delete mode 100644 common/core/src/main/java/zingg/common/core/preprocess/stopwords/StopWords.java delete mode 100644 spark/core/src/main/java/zingg/spark/core/preprocess/stopwords/RemoveStopWordsUDF.java 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 deleted file mode 100644 index f0c95e1aa..000000000 --- a/common/core/src/main/java/zingg/common/core/preprocess/stopwords/RemoveStopWords.java +++ /dev/null @@ -1,27 +0,0 @@ -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 deleted file mode 100644 index e7b6407d2..000000000 --- a/common/core/src/main/java/zingg/common/core/preprocess/stopwords/StopWords.java +++ /dev/null @@ -1,58 +0,0 @@ -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 deleted file mode 100644 index 2e9943b98..000000000 --- a/spark/core/src/main/java/zingg/spark/core/preprocess/stopwords/RemoveStopWordsUDF.java +++ /dev/null @@ -1,21 +0,0 @@ -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 dcac47762..36c7dc9b1 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,7 +1,6 @@ package zingg.spark.core.preprocess.stopwords; -import static org.apache.spark.sql.functions.callUDF; -import static org.apache.spark.sql.functions.lit; +import static org.apache.spark.sql.functions.regexp_replace; import java.io.Serializable; @@ -11,7 +10,6 @@ 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; @@ -19,50 +17,34 @@ 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,callUDF(udfName, ds.df().col(fieldName),lit(pattern))); + Dataset dfAfterRemoval = ds.df().withColumn(fieldName, regexp_replace(ds.df().col(fieldName), 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(); } }