You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2022/12/09 04:59:38 UTC
[flink-ml] branch master updated: [FLINK-29603] Add Transformer for StopWordsRemover (#183)
This is an automated email from the ASF dual-hosted git repository.
lindong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-ml.git
The following commit(s) were added to refs/heads/master by this push:
new 2fbf793 [FLINK-29603] Add Transformer for StopWordsRemover (#183)
2fbf793 is described below
commit 2fbf7936320c91159f2025204cf9d8c93be73820
Author: yunfengzhou-hub <yu...@outlook.com>
AuthorDate: Fri Dec 9 12:59:32 2022 +0800
[FLINK-29603] Add Transformer for StopWordsRemover (#183)
This closes #183.
---
.github/workflows/python-checks.yml | 5 +-
README.md | 2 +
.../docs/operators/feature/stopwordsremover.md | 165 +++++++
.../examples/feature/StopWordsRemoverExample.java | 67 +++
.../feature/stopwordsremover/StopWordsRemover.java | 173 ++++++++
.../stopwordsremover/StopWordsRemoverParams.java | 83 ++++
.../stopwordsremover/StopWordsRemoverUtils.java | 99 +++++
.../org/apache/flink/ml/feature/stopwords/README | 12 +
.../apache/flink/ml/feature/stopwords/danish.txt | 94 ++++
.../apache/flink/ml/feature/stopwords/dutch.txt | 101 +++++
.../apache/flink/ml/feature/stopwords/english.txt | 181 ++++++++
.../apache/flink/ml/feature/stopwords/finnish.txt | 235 ++++++++++
.../apache/flink/ml/feature/stopwords/french.txt | 156 +++++++
.../apache/flink/ml/feature/stopwords/german.txt | 231 ++++++++++
.../flink/ml/feature/stopwords/hungarian.txt | 199 +++++++++
.../apache/flink/ml/feature/stopwords/italian.txt | 279 ++++++++++++
.../flink/ml/feature/stopwords/norwegian.txt | 176 ++++++++
.../flink/ml/feature/stopwords/portuguese.txt | 203 +++++++++
.../apache/flink/ml/feature/stopwords/russian.txt | 151 +++++++
.../apache/flink/ml/feature/stopwords/spanish.txt | 313 +++++++++++++
.../apache/flink/ml/feature/stopwords/swedish.txt | 114 +++++
.../apache/flink/ml/feature/stopwords/turkish.txt | 53 +++
.../flink/ml/feature/StopWordsRemoverTest.java | 492 +++++++++++++++++++++
.../ml/feature/stopwordsremover_example.py | 58 +++
.../pyflink/ml/lib/feature/stopwordsremover.py | 148 +++++++
.../ml/lib/feature/tests/test_stopwordsremover.py | 115 +++++
flink-ml-python/pyflink/ml/tests/test_utils.py | 15 -
.../apache/flink/ml/python/PythonBridgeUtils.java | 2 +-
pom.xml | 2 +
29 files changed, 3907 insertions(+), 17 deletions(-)
diff --git a/.github/workflows/python-checks.yml b/.github/workflows/python-checks.yml
index 6a5deda..dba36bc 100644
--- a/.github/workflows/python-checks.yml
+++ b/.github/workflows/python-checks.yml
@@ -29,7 +29,7 @@ jobs:
runs-on: ${{ matrix.os }}
strategy:
matrix:
- os: ['macOS-latest', 'ubuntu-latest']
+ os: ['macOS-latest', 'ubuntu-20.04']
python-version: ['3.6', '3.7', '3.8']
steps:
- name: Checkout code
@@ -47,6 +47,9 @@ jobs:
- name: Install dependencies
working-directory: flink-ml-python/dev
run: python -m pip install -r dev-requirements.txt
+ - name: Install Flink ML
+ working-directory: flink-ml-python
+ run: python setup.py sdist && python -m pip install dist/*.tar.gz
- name: Check code style
working-directory: flink-ml-python
run: python -m flake8 --config=setup.cfg
diff --git a/README.md b/README.md
index 26f637a..5435e21 100644
--- a/README.md
+++ b/README.md
@@ -46,3 +46,5 @@ overview of ongoing community work.
The code in this repository is licensed under the [Apache Software License
2](LICENSE).
+
+[//]: # (TODO: Add a guideline for developers to install Flink ML and run tests.)
diff --git a/docs/content/docs/operators/feature/stopwordsremover.md b/docs/content/docs/operators/feature/stopwordsremover.md
new file mode 100644
index 0000000..dd1933b
--- /dev/null
+++ b/docs/content/docs/operators/feature/stopwordsremover.md
@@ -0,0 +1,165 @@
+---
+title: "StopWordsRemover"
+weight: 1
+type: docs
+aliases:
+- /operators/feature/stopwordsremover.html
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+## StopWordsRemover
+
+A feature transformer that filters out stop words from input.
+
+Note: null values from input array are preserved unless adding null to stopWords
+explicitly.
+
+See Also: <a href="http://en.wikipedia.org/wiki/Stop_words">Stop words
+(Wikipedia)</a>
+
+### Input Columns
+
+| Param name | Type | Default | Description |
+|:-----------|:---------|:--------|:---------------------------------------------------|
+| inputCols | String[] | `null` | Arrays of strings containing stop words to remove. |
+
+### Output Columns
+
+| Param name | Type | Default | Description |
+|:-----------|:---------|:--------|:-------------------------------------------|
+| outputCols | String[] | `null` | Arrays of strings with stop words removed. |
+
+### Parameters
+
+| Key | Default | Type | Required | Description |
+|---------------|----------------------------------------------------|----------|----------|----------------------------------------------------------------------------------------|
+| inputCols | `null` | String[] | yes | Input column names. |
+| outputCols | `null` | String[] | yes | Output column name. |
+| stopWords | `StopWordsRemover.loadDefaultStopWords("english")` | String[] | no | The words to be filtered out. |
+| caseSensitive | `false` | Boolean | no | Whether to do a case-sensitive comparison over the stop words. |
+| locale | `StopWordsRemover.getDefaultOrUS().toString()` | String | no | Locale of the input for case insensitive matching. Ignored when caseSensitive is true. |
+
+### Examples
+
+{{< tabs examples >}}
+
+{{< tab "Java">}}
+
+```java
+import org.apache.flink.ml.feature.stopwordsremover.StopWordsRemover;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import java.util.Arrays;
+
+/** Simple program that creates a StopWordsRemover instance and uses it for feature engineering. */
+public class StopWordsRemoverExample {
+ public static void main(String[] args) {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ // Generates input data.
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of((Object) new String[] {"test", "test"}),
+ Row.of((Object) new String[] {"a", "b", "c", "d"}),
+ Row.of((Object) new String[] {"a", "the", "an"}),
+ Row.of((Object) new String[] {"A", "The", "AN"}),
+ Row.of((Object) new String[] {null}),
+ Row.of((Object) new String[] {}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("input");
+
+ // Creates a StopWordsRemover object and initializes its parameters.
+ StopWordsRemover remover =
+ new StopWordsRemover().setInputCols("input").setOutputCols("output");
+
+ // Uses the StopWordsRemover object for feature transformations.
+ Table outputTable = remover.transform(inputTable)[0];
+
+ // Extracts and displays the results.
+ for (CloseableIterator<Row> it = outputTable.execute().collect(); it.hasNext(); ) {
+ Row row = it.next();
+
+ String[] inputValues = row.getFieldAs("input");
+ String[] outputValues = row.getFieldAs("output");
+
+ System.out.printf(
+ "Input Values: %s\tOutput Values: %s\n",
+ Arrays.toString(inputValues), Arrays.toString(outputValues));
+ }
+ }
+}
+```
+
+{{< /tab>}}
+
+{{< tab "Python">}}
+
+```python
+# Simple program that creates a StopWordsRemover instance and uses it for feature
+# engineering.
+
+from pyflink.common import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.ml.lib.feature.stopwordsremover import StopWordsRemover
+from pyflink.table import StreamTableEnvironment
+
+# create a new StreamExecutionEnvironment
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# create a StreamTableEnvironment
+t_env = StreamTableEnvironment.create(env)
+
+# generate input data
+input_table = t_env.from_data_stream(
+ env.from_collection([
+ (["test", "test"],),
+ (["a", "b", "c", "d"],),
+ (["a", "the", "an"],),
+ (["A", "The", "AN"],),
+ ([None],),
+ ([],),
+ ],
+ type_info=Types.ROW_NAMED(
+ ['input'],
+ [Types.OBJECT_ARRAY(Types.STRING())])))
+
+# create a StopWordsRemover object and initialize its parameters
+remover = StopWordsRemover().set_input_cols('input').set_output_cols('output')
+
+# use the StopWordsRemover for feature engineering
+output_table = remover.transform(input_table)[0]
+
+# extract and display the results
+field_names = output_table.get_schema().get_field_names()
+for result in t_env.to_data_stream(output_table).execute_and_collect():
+ input_value = result[field_names.index('input')]
+ output_value = result[field_names.index('output')]
+ print('Input Value: ' + str(input_value) + '\tOutput Value: ' + str(output_value))
+```
+
+{{< /tab>}}
+
+{{< /tabs>}}
diff --git a/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/feature/StopWordsRemoverExample.java b/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/feature/StopWordsRemoverExample.java
new file mode 100644
index 0000000..a993b72
--- /dev/null
+++ b/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/feature/StopWordsRemoverExample.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.examples.feature;
+
+import org.apache.flink.ml.feature.stopwordsremover.StopWordsRemover;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import java.util.Arrays;
+
+/** Simple program that creates a StopWordsRemover instance and uses it for feature engineering. */
+public class StopWordsRemoverExample {
+ public static void main(String[] args) {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ // Generates input data.
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of((Object) new String[] {"test", "test"}),
+ Row.of((Object) new String[] {"a", "b", "c", "d"}),
+ Row.of((Object) new String[] {"a", "the", "an"}),
+ Row.of((Object) new String[] {"A", "The", "AN"}),
+ Row.of((Object) new String[] {null}),
+ Row.of((Object) new String[] {}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("input");
+
+ // Creates a StopWordsRemover object and initializes its parameters.
+ StopWordsRemover remover =
+ new StopWordsRemover().setInputCols("input").setOutputCols("output");
+
+ // Uses the StopWordsRemover object for feature transformations.
+ Table outputTable = remover.transform(inputTable)[0];
+
+ // Extracts and displays the results.
+ for (CloseableIterator<Row> it = outputTable.execute().collect(); it.hasNext(); ) {
+ Row row = it.next();
+
+ String[] inputValues = row.getFieldAs("input");
+ String[] outputValues = row.getFieldAs("output");
+
+ System.out.printf(
+ "Input Values: %s\tOutput Values: %s\n",
+ Arrays.toString(inputValues), Arrays.toString(outputValues));
+ }
+ }
+}
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/stopwordsremover/StopWordsRemover.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/stopwordsremover/StopWordsRemover.java
new file mode 100644
index 0000000..4994bf8
--- /dev/null
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/stopwordsremover/StopWordsRemover.java
@@ -0,0 +1,173 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.feature.stopwordsremover;
+
+import org.apache.flink.ml.api.Transformer;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.functions.FunctionContext;
+import org.apache.flink.table.functions.ScalarFunction;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.call;
+
+/**
+ * A feature transformer that filters out stop words from input.
+ *
+ * <p>Note: null values from input array are preserved unless adding null to stopWords explicitly.
+ *
+ * @see <a href="http://en.wikipedia.org/wiki/Stop_words">Stop words (Wikipedia)</a>
+ */
+public class StopWordsRemover
+ implements Transformer<StopWordsRemover>, StopWordsRemoverParams<StopWordsRemover> {
+
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public StopWordsRemover() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Preconditions.checkArgument(getInputCols().length == getOutputCols().length);
+
+ String[] inputCols = getInputCols();
+ String[] outputCols = getOutputCols();
+
+ ScalarFunction function =
+ new RemoveStopWordsFunction(
+ new HashSet<>(Arrays.asList(getStopWords())),
+ new Locale(getLocale()),
+ getCaseSensitive());
+
+ Expression[] expressions = new Expression[inputCols.length + 1];
+ expressions[0] = $("*");
+ for (int i = 0; i < inputCols.length; i++) {
+ expressions[i + 1] = call(function, $(inputCols[i])).as(outputCols[i]);
+ }
+
+ Table output = inputs[0].select(expressions);
+
+ return new Table[] {output};
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static StopWordsRemover load(StreamTableEnvironment env, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ /**
+ * Loads the default stop words for the given language.
+ *
+ * <p>Supported languages: danish, dutch, english, finnish, french, german, hungarian, italian,
+ * norwegian, portuguese, russian, spanish, swedish, turkish
+ *
+ * @see <a
+ * href="http://anoncvs.postgresql.org/cvsweb.cgi/pgsql/src/backend/snowball/stopwords/">here</a>
+ */
+ public static String[] loadDefaultStopWords(String language) {
+ return StopWordsRemoverUtils.loadDefaultStopWords(language);
+ }
+
+ /**
+ * Returns system default locale, or {@link Locale#US} if the default locale is not in available
+ * locales in JVM. The locale is returned as a String.
+ */
+ public static String getDefaultOrUS() {
+ return StopWordsRemoverUtils.getDefaultOrUS();
+ }
+
+ /**
+ * Returns a set of all installed locales. It must contain at least a {@link Locale} instance
+ * equal to {@link Locale#US}. The locales are returned as Strings.
+ *
+ * @return A set of installed locales.
+ */
+ public static Set<String> getAvailableLocales() {
+ return StopWordsRemoverUtils.getAvailableLocales();
+ }
+
+ /** A Scalar Function that removes stop words from input string array. */
+ public static class RemoveStopWordsFunction extends ScalarFunction {
+ private final Set<String> stopWords;
+ private final Locale locale;
+ private final boolean caseSensitive;
+ private transient Predicate<String> predicate;
+
+ public RemoveStopWordsFunction(
+ Set<String> stopWords, Locale locale, boolean caseSensitive) {
+ this.locale = locale;
+ this.caseSensitive = caseSensitive;
+ if (caseSensitive) {
+ this.stopWords = stopWords;
+ } else {
+ this.stopWords =
+ stopWords.stream()
+ .map(x -> x == null ? null : x.toLowerCase(locale))
+ .collect(Collectors.toSet());
+ }
+ }
+
+ @Override
+ public void open(FunctionContext context) throws Exception {
+ super.open(context);
+ if (caseSensitive) {
+ predicate = x -> !stopWords.contains(x);
+ } else {
+ predicate =
+ x -> {
+ if (x != null) {
+ x = x.toLowerCase(locale);
+ }
+ return !stopWords.contains(x);
+ };
+ }
+ }
+
+ public String[] eval(String[] input) {
+ return Arrays.stream(input).filter(predicate).toArray(String[]::new);
+ }
+ }
+}
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/stopwordsremover/StopWordsRemoverParams.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/stopwordsremover/StopWordsRemoverParams.java
new file mode 100644
index 0000000..cf16811
--- /dev/null
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/stopwordsremover/StopWordsRemoverParams.java
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.feature.stopwordsremover;
+
+import org.apache.flink.ml.common.param.HasInputCols;
+import org.apache.flink.ml.common.param.HasOutputCols;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringArrayParam;
+import org.apache.flink.ml.param.StringParam;
+
+import static org.apache.flink.ml.feature.stopwordsremover.StopWordsRemoverUtils.getAvailableLocales;
+import static org.apache.flink.ml.feature.stopwordsremover.StopWordsRemoverUtils.getDefaultOrUS;
+import static org.apache.flink.ml.feature.stopwordsremover.StopWordsRemoverUtils.loadDefaultStopWords;
+
+/**
+ * Params of {@link StopWordsRemover}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface StopWordsRemoverParams<T> extends HasInputCols<T>, HasOutputCols<T> {
+
+ Param<String[]> STOP_WORDS =
+ new StringArrayParam(
+ "stopWords",
+ "The words to be filtered out.",
+ loadDefaultStopWords("english"),
+ ParamValidators.nonEmptyArray());
+
+ Param<Boolean> CASE_SENSITIVE =
+ new BooleanParam(
+ "caseSensitive",
+ "Whether to do a case-sensitive comparison over the stop words.",
+ false);
+
+ Param<String> LOCALE =
+ new StringParam(
+ "locale",
+ "Locale of the input for case insensitive matching. Ignored when caseSensitive is true.",
+ getDefaultOrUS(),
+ ParamValidators.inArray(getAvailableLocales().toArray(new String[0])));
+
+ default String[] getStopWords() {
+ return get(STOP_WORDS);
+ }
+
+ default T setStopWords(String[] value) {
+ return set(STOP_WORDS, value);
+ }
+
+ default boolean getCaseSensitive() {
+ return get(CASE_SENSITIVE);
+ }
+
+ default T setCaseSensitive(boolean value) {
+ return set(CASE_SENSITIVE, value);
+ }
+
+ default String getLocale() {
+ return get(LOCALE);
+ }
+
+ default T setLocale(String value) {
+ return set(LOCALE, value);
+ }
+}
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/stopwordsremover/StopWordsRemoverUtils.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/stopwordsremover/StopWordsRemoverUtils.java
new file mode 100644
index 0000000..f3d776a
--- /dev/null
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/stopwordsremover/StopWordsRemoverUtils.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.feature.stopwordsremover;
+
+import org.apache.flink.util.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Set;
+
+/** Utility methods used by {@link StopWordsRemover} and {@link StopWordsRemoverParams}. */
+class StopWordsRemoverUtils {
+
+ private static final Set<String> SUPPORTED_LANGUAGES =
+ new HashSet<>(
+ Arrays.asList(
+ "danish",
+ "dutch",
+ "english",
+ "finnish",
+ "french",
+ "german",
+ "hungarian",
+ "italian",
+ "norwegian",
+ "portuguese",
+ "russian",
+ "spanish",
+ "swedish",
+ "turkish"));
+
+ private static final Logger LOG = LoggerFactory.getLogger(StopWordsRemover.class);
+
+ /** See {@link StopWordsRemover#loadDefaultStopWords(String)}. */
+ static String[] loadDefaultStopWords(String language) {
+ Preconditions.checkArgument(
+ SUPPORTED_LANGUAGES.contains(language),
+ "%s is not in the supported language list: %s.",
+ language,
+ SUPPORTED_LANGUAGES);
+
+ InputStream in =
+ StopWordsRemover.class
+ .getClassLoader()
+ .getResourceAsStream(
+ "org/apache/flink/ml/feature/stopwords/" + language + ".txt");
+
+ return new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8))
+ .lines()
+ .toArray(String[]::new);
+ }
+
+ /** See {@link StopWordsRemover#getDefaultOrUS()}. */
+ static String getDefaultOrUS() {
+ if (Arrays.asList(Locale.getAvailableLocales()).contains(Locale.getDefault())) {
+ return Locale.getDefault().toString();
+ } else {
+ LOG.warn(
+ "Default locale set was [{}]; however, it was "
+ + "not found in available locales in JVM, falling back to en_US locale. Set param `locale` "
+ + "in order to respect another locale.",
+ Locale.getDefault());
+ return Locale.US.toString();
+ }
+ }
+
+ /** See {@link StopWordsRemover#getAvailableLocales()}. */
+ static Set<String> getAvailableLocales() {
+ Set<String> locales = new HashSet<>();
+ for (Locale locale : Locale.getAvailableLocales()) {
+ locales.add(locale.toString());
+ }
+ return locales;
+ }
+}
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/README b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/README
new file mode 100755
index 0000000..e60cf65
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/README
@@ -0,0 +1,12 @@
+Stopwords Corpus
+
+This corpus contains lists of stop words for several languages. These
+are high-frequency grammatical words which are usually ignored in text
+retrieval applications.
+
+They were obtained from:
+http://anoncvs.postgresql.org/cvsweb.cgi/pgsql/src/backend/snowball/stopwords/
+
+The English list has been augmented
+https://github.com/nltk/nltk_data/issues/22
+
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/danish.txt b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/danish.txt
new file mode 100644
index 0000000..ea9e2c4
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/danish.txt
@@ -0,0 +1,94 @@
+og
+i
+jeg
+det
+at
+en
+den
+til
+er
+som
+på
+de
+med
+han
+af
+for
+ikke
+der
+var
+mig
+sig
+men
+et
+har
+om
+vi
+min
+havde
+ham
+hun
+nu
+over
+da
+fra
+du
+ud
+sin
+dem
+os
+op
+man
+hans
+hvor
+eller
+hvad
+skal
+selv
+her
+alle
+vil
+blev
+kunne
+ind
+når
+være
+dog
+noget
+ville
+jo
+deres
+efter
+ned
+skulle
+denne
+end
+dette
+mit
+også
+under
+have
+dig
+anden
+hende
+mine
+alt
+meget
+sit
+sine
+vor
+mod
+disse
+hvis
+din
+nogle
+hos
+blive
+mange
+ad
+bliver
+hendes
+været
+thi
+jer
+sådan
\ No newline at end of file
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/dutch.txt b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/dutch.txt
new file mode 100644
index 0000000..023cc2c
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/dutch.txt
@@ -0,0 +1,101 @@
+de
+en
+van
+ik
+te
+dat
+die
+in
+een
+hij
+het
+niet
+zijn
+is
+was
+op
+aan
+met
+als
+voor
+had
+er
+maar
+om
+hem
+dan
+zou
+of
+wat
+mijn
+men
+dit
+zo
+door
+over
+ze
+zich
+bij
+ook
+tot
+je
+mij
+uit
+der
+daar
+haar
+naar
+heb
+hoe
+heeft
+hebben
+deze
+u
+want
+nog
+zal
+me
+zij
+nu
+ge
+geen
+omdat
+iets
+worden
+toch
+al
+waren
+veel
+meer
+doen
+toen
+moet
+ben
+zonder
+kan
+hun
+dus
+alles
+onder
+ja
+eens
+hier
+wie
+werd
+altijd
+doch
+wordt
+wezen
+kunnen
+ons
+zelf
+tegen
+na
+reeds
+wil
+kon
+niets
+uw
+iemand
+geweest
+andere
\ No newline at end of file
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/english.txt b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/english.txt
new file mode 100644
index 0000000..d6094d7
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/english.txt
@@ -0,0 +1,181 @@
+i
+me
+my
+myself
+we
+our
+ours
+ourselves
+you
+your
+yours
+yourself
+yourselves
+he
+him
+his
+himself
+she
+her
+hers
+herself
+it
+its
+itself
+they
+them
+their
+theirs
+themselves
+what
+which
+who
+whom
+this
+that
+these
+those
+am
+is
+are
+was
+were
+be
+been
+being
+have
+has
+had
+having
+do
+does
+did
+doing
+a
+an
+the
+and
+but
+if
+or
+because
+as
+until
+while
+of
+at
+by
+for
+with
+about
+against
+between
+into
+through
+during
+before
+after
+above
+below
+to
+from
+up
+down
+in
+out
+on
+off
+over
+under
+again
+further
+then
+once
+here
+there
+when
+where
+why
+how
+all
+any
+both
+each
+few
+more
+most
+other
+some
+such
+no
+nor
+not
+only
+own
+same
+so
+than
+too
+very
+s
+t
+can
+will
+just
+don
+should
+now
+i'll
+you'll
+he'll
+she'll
+we'll
+they'll
+i'd
+you'd
+he'd
+she'd
+we'd
+they'd
+i'm
+you're
+he's
+she's
+it's
+we're
+they're
+i've
+we've
+you've
+they've
+isn't
+aren't
+wasn't
+weren't
+haven't
+hasn't
+hadn't
+don't
+doesn't
+didn't
+won't
+wouldn't
+shan't
+shouldn't
+mustn't
+can't
+couldn't
+cannot
+could
+here's
+how's
+let's
+ought
+that's
+there's
+what's
+when's
+where's
+who's
+why's
+would
\ No newline at end of file
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/finnish.txt b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/finnish.txt
new file mode 100644
index 0000000..5b0eb10
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/finnish.txt
@@ -0,0 +1,235 @@
+olla
+olen
+olet
+on
+olemme
+olette
+ovat
+ole
+oli
+olisi
+olisit
+olisin
+olisimme
+olisitte
+olisivat
+olit
+olin
+olimme
+olitte
+olivat
+ollut
+olleet
+en
+et
+ei
+emme
+ette
+eivät
+minä
+minun
+minut
+minua
+minussa
+minusta
+minuun
+minulla
+minulta
+minulle
+sinä
+sinun
+sinut
+sinua
+sinussa
+sinusta
+sinuun
+sinulla
+sinulta
+sinulle
+hän
+hänen
+hänet
+häntä
+hänessä
+hänestä
+häneen
+hänellä
+häneltä
+hänelle
+me
+meidän
+meidät
+meitä
+meissä
+meistä
+meihin
+meillä
+meiltä
+meille
+te
+teidän
+teidät
+teitä
+teissä
+teistä
+teihin
+teillä
+teiltä
+teille
+he
+heidän
+heidät
+heitä
+heissä
+heistä
+heihin
+heillä
+heiltä
+heille
+tämä
+tämän
+tätä
+tässä
+tästä
+tähän
+tallä
+tältä
+tälle
+tänä
+täksi
+tuo
+tuon
+tuotä
+tuossa
+tuosta
+tuohon
+tuolla
+tuolta
+tuolle
+tuona
+tuoksi
+se
+sen
+sitä
+siinä
+siitä
+siihen
+sillä
+siltä
+sille
+sinä
+siksi
+nämä
+näiden
+näitä
+näissä
+näistä
+näihin
+näillä
+näiltä
+näille
+näinä
+näiksi
+nuo
+noiden
+noita
+noissa
+noista
+noihin
+noilla
+noilta
+noille
+noina
+noiksi
+ne
+niiden
+niitä
+niissä
+niistä
+niihin
+niillä
+niiltä
+niille
+niinä
+niiksi
+kuka
+kenen
+kenet
+ketä
+kenessä
+kenestä
+keneen
+kenellä
+keneltä
+kenelle
+kenenä
+keneksi
+ketkä
+keiden
+ketkä
+keitä
+keissä
+keistä
+keihin
+keillä
+keiltä
+keille
+keinä
+keiksi
+mikä
+minkä
+minkä
+mitä
+missä
+mistä
+mihin
+millä
+miltä
+mille
+minä
+miksi
+mitkä
+joka
+jonka
+jota
+jossa
+josta
+johon
+jolla
+jolta
+jolle
+jona
+joksi
+jotka
+joiden
+joita
+joissa
+joista
+joihin
+joilla
+joilta
+joille
+joina
+joiksi
+että
+ja
+jos
+koska
+kuin
+mutta
+niin
+sekä
+sillä
+tai
+vaan
+vai
+vaikka
+kanssa
+mukaan
+noin
+poikki
+yli
+kun
+niin
+nyt
+itse
\ No newline at end of file
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/french.txt b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/french.txt
new file mode 100644
index 0000000..a59a042
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/french.txt
@@ -0,0 +1,156 @@
+au
+aux
+avec
+ce
+ces
+dans
+de
+des
+du
+elle
+en
+et
+eux
+il
+je
+la
+le
+les
+leur
+lui
+ma
+mais
+me
+même
+mes
+moi
+mon
+ne
+nos
+notre
+nous
+on
+ou
+par
+pas
+pour
+qu
+que
+qui
+sa
+se
+ses
+son
+sur
+ta
+te
+tes
+toi
+ton
+tu
+un
+une
+vos
+votre
+vous
+c
+d
+j
+l
+à
+m
+n
+s
+t
+y
+été
+étée
+étées
+étés
+étant
+étante
+étants
+étantes
+suis
+es
+est
+sommes
+êtes
+sont
+serai
+seras
+sera
+serons
+serez
+seront
+serais
+serait
+serions
+seriez
+seraient
+étais
+était
+étions
+étiez
+étaient
+fus
+fut
+fûmes
+fûtes
+furent
+sois
+soit
+soyons
+soyez
+soient
+fusse
+fusses
+fût
+fussions
+fussiez
+fussent
+ayant
+ayante
+ayantes
+ayants
+eu
+eue
+eues
+eus
+ai
+as
+avons
+avez
+ont
+aurai
+auras
+aura
+aurons
+aurez
+auront
+aurais
+aurait
+aurions
+auriez
+auraient
+avais
+avait
+avions
+aviez
+avaient
+eut
+eûmes
+eûtes
+eurent
+aie
+aies
+ait
+ayons
+ayez
+aient
+eusse
+eusses
+eût
+eussions
+eussiez
+eussent
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/german.txt b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/german.txt
new file mode 100644
index 0000000..7e65190
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/german.txt
@@ -0,0 +1,231 @@
+aber
+alle
+allem
+allen
+aller
+alles
+als
+also
+am
+an
+ander
+andere
+anderem
+anderen
+anderer
+anderes
+anderm
+andern
+anderr
+anders
+auch
+auf
+aus
+bei
+bin
+bis
+bist
+da
+damit
+dann
+der
+den
+des
+dem
+die
+das
+daß
+derselbe
+derselben
+denselben
+desselben
+demselben
+dieselbe
+dieselben
+dasselbe
+dazu
+dein
+deine
+deinem
+deinen
+deiner
+deines
+denn
+derer
+dessen
+dich
+dir
+du
+dies
+diese
+diesem
+diesen
+dieser
+dieses
+doch
+dort
+durch
+ein
+eine
+einem
+einen
+einer
+eines
+einig
+einige
+einigem
+einigen
+einiger
+einiges
+einmal
+er
+ihn
+ihm
+es
+etwas
+euer
+eure
+eurem
+euren
+eurer
+eures
+für
+gegen
+gewesen
+hab
+habe
+haben
+hat
+hatte
+hatten
+hier
+hin
+hinter
+ich
+mich
+mir
+ihr
+ihre
+ihrem
+ihren
+ihrer
+ihres
+euch
+im
+in
+indem
+ins
+ist
+jede
+jedem
+jeden
+jeder
+jedes
+jene
+jenem
+jenen
+jener
+jenes
+jetzt
+kann
+kein
+keine
+keinem
+keinen
+keiner
+keines
+können
+könnte
+machen
+man
+manche
+manchem
+manchen
+mancher
+manches
+mein
+meine
+meinem
+meinen
+meiner
+meines
+mit
+muss
+musste
+nach
+nicht
+nichts
+noch
+nun
+nur
+ob
+oder
+ohne
+sehr
+sein
+seine
+seinem
+seinen
+seiner
+seines
+selbst
+sich
+sie
+ihnen
+sind
+so
+solche
+solchem
+solchen
+solcher
+solches
+soll
+sollte
+sondern
+sonst
+über
+um
+und
+uns
+unse
+unsem
+unsen
+unser
+unses
+unter
+viel
+vom
+von
+vor
+während
+war
+waren
+warst
+was
+weg
+weil
+weiter
+welche
+welchem
+welchen
+welcher
+welches
+wenn
+werde
+werden
+wie
+wieder
+will
+wir
+wird
+wirst
+wo
+wollen
+wollte
+würde
+würden
+zu
+zum
+zur
+zwar
+zwischen
\ No newline at end of file
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/hungarian.txt b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/hungarian.txt
new file mode 100644
index 0000000..8d4543a
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/hungarian.txt
@@ -0,0 +1,199 @@
+a
+ahogy
+ahol
+aki
+akik
+akkor
+alatt
+által
+általában
+amely
+amelyek
+amelyekben
+amelyeket
+amelyet
+amelynek
+ami
+amit
+amolyan
+amíg
+amikor
+át
+abban
+ahhoz
+annak
+arra
+arról
+az
+azok
+azon
+azt
+azzal
+azért
+aztán
+azután
+azonban
+bár
+be
+belül
+benne
+cikk
+cikkek
+cikkeket
+csak
+de
+e
+eddig
+egész
+egy
+egyes
+egyetlen
+egyéb
+egyik
+egyre
+ekkor
+el
+elég
+ellen
+elõ
+elõször
+elõtt
+elsõ
+én
+éppen
+ebben
+ehhez
+emilyen
+ennek
+erre
+ez
+ezt
+ezek
+ezen
+ezzel
+ezért
+és
+fel
+felé
+hanem
+hiszen
+hogy
+hogyan
+igen
+így
+illetve
+ill.
+ill
+ilyen
+ilyenkor
+ison
+ismét
+itt
+jó
+jól
+jobban
+kell
+kellett
+keresztül
+keressünk
+ki
+kívül
+között
+közül
+legalább
+lehet
+lehetett
+legyen
+lenne
+lenni
+lesz
+lett
+maga
+magát
+majd
+majd
+már
+más
+másik
+meg
+még
+mellett
+mert
+mely
+melyek
+mi
+mit
+míg
+miért
+milyen
+mikor
+minden
+mindent
+mindenki
+mindig
+mint
+mintha
+mivel
+most
+nagy
+nagyobb
+nagyon
+ne
+néha
+nekem
+neki
+nem
+néhány
+nélkül
+nincs
+olyan
+ott
+össze
+õ
+õk
+õket
+pedig
+persze
+rá
+s
+saját
+sem
+semmi
+sok
+sokat
+sokkal
+számára
+szemben
+szerint
+szinte
+talán
+tehát
+teljes
+tovább
+továbbá
+több
+úgy
+ugyanis
+új
+újabb
+újra
+után
+utána
+utolsó
+vagy
+vagyis
+valaki
+valami
+valamint
+való
+vagyok
+van
+vannak
+volt
+voltam
+voltak
+voltunk
+vissza
+vele
+viszont
+volna
\ No newline at end of file
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/italian.txt b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/italian.txt
new file mode 100644
index 0000000..783b2e0
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/italian.txt
@@ -0,0 +1,279 @@
+ad
+al
+allo
+ai
+agli
+all
+agl
+alla
+alle
+con
+col
+coi
+da
+dal
+dallo
+dai
+dagli
+dall
+dagl
+dalla
+dalle
+di
+del
+dello
+dei
+degli
+dell
+degl
+della
+delle
+in
+nel
+nello
+nei
+negli
+nell
+negl
+nella
+nelle
+su
+sul
+sullo
+sui
+sugli
+sull
+sugl
+sulla
+sulle
+per
+tra
+contro
+io
+tu
+lui
+lei
+noi
+voi
+loro
+mio
+mia
+miei
+mie
+tuo
+tua
+tuoi
+tue
+suo
+sua
+suoi
+sue
+nostro
+nostra
+nostri
+nostre
+vostro
+vostra
+vostri
+vostre
+mi
+ti
+ci
+vi
+lo
+la
+li
+le
+gli
+ne
+il
+un
+uno
+una
+ma
+ed
+se
+perché
+anche
+come
+dov
+dove
+che
+chi
+cui
+non
+più
+quale
+quanto
+quanti
+quanta
+quante
+quello
+quelli
+quella
+quelle
+questo
+questi
+questa
+queste
+si
+tutto
+tutti
+a
+c
+e
+i
+l
+o
+ho
+hai
+ha
+abbiamo
+avete
+hanno
+abbia
+abbiate
+abbiano
+avrò
+avrai
+avrà
+avremo
+avrete
+avranno
+avrei
+avresti
+avrebbe
+avremmo
+avreste
+avrebbero
+avevo
+avevi
+aveva
+avevamo
+avevate
+avevano
+ebbi
+avesti
+ebbe
+avemmo
+aveste
+ebbero
+avessi
+avesse
+avessimo
+avessero
+avendo
+avuto
+avuta
+avuti
+avute
+sono
+sei
+è
+siamo
+siete
+sia
+siate
+siano
+sarò
+sarai
+sarà
+saremo
+sarete
+saranno
+sarei
+saresti
+sarebbe
+saremmo
+sareste
+sarebbero
+ero
+eri
+era
+eravamo
+eravate
+erano
+fui
+fosti
+fu
+fummo
+foste
+furono
+fossi
+fosse
+fossimo
+fossero
+essendo
+faccio
+fai
+facciamo
+fanno
+faccia
+facciate
+facciano
+farò
+farai
+farà
+faremo
+farete
+faranno
+farei
+faresti
+farebbe
+faremmo
+fareste
+farebbero
+facevo
+facevi
+faceva
+facevamo
+facevate
+facevano
+feci
+facesti
+fece
+facemmo
+faceste
+fecero
+facessi
+facesse
+facessimo
+facessero
+facendo
+sto
+stai
+sta
+stiamo
+stanno
+stia
+stiate
+stiano
+starò
+starai
+starà
+staremo
+starete
+staranno
+starei
+staresti
+starebbe
+staremmo
+stareste
+starebbero
+stavo
+stavi
+stava
+stavamo
+stavate
+stavano
+stetti
+stesti
+stette
+stemmo
+steste
+stettero
+stessi
+stesse
+stessimo
+stessero
+stando
\ No newline at end of file
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/norwegian.txt b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/norwegian.txt
new file mode 100644
index 0000000..cb91702
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/norwegian.txt
@@ -0,0 +1,176 @@
+og
+i
+jeg
+det
+at
+en
+et
+den
+til
+er
+som
+på
+de
+med
+han
+av
+ikke
+ikkje
+der
+så
+var
+meg
+seg
+men
+ett
+har
+om
+vi
+min
+mitt
+ha
+hadde
+hun
+nå
+over
+da
+ved
+fra
+du
+ut
+sin
+dem
+oss
+opp
+man
+kan
+hans
+hvor
+eller
+hva
+skal
+selv
+sjøl
+her
+alle
+vil
+bli
+ble
+blei
+blitt
+kunne
+inn
+når
+være
+kom
+noen
+noe
+ville
+dere
+som
+deres
+kun
+ja
+etter
+ned
+skulle
+denne
+for
+deg
+si
+sine
+sitt
+mot
+å
+meget
+hvorfor
+dette
+disse
+uten
+hvordan
+ingen
+din
+ditt
+blir
+samme
+hvilken
+hvilke
+sånn
+inni
+mellom
+vår
+hver
+hvem
+vors
+hvis
+både
+bare
+enn
+fordi
+før
+mange
+også
+slik
+vært
+være
+båe
+begge
+siden
+dykk
+dykkar
+dei
+deira
+deires
+deim
+di
+då
+eg
+ein
+eit
+eitt
+elles
+honom
+hjå
+ho
+hoe
+henne
+hennar
+hennes
+hoss
+hossen
+ikkje
+ingi
+inkje
+korleis
+korso
+kva
+kvar
+kvarhelst
+kven
+kvi
+kvifor
+me
+medan
+mi
+mine
+mykje
+no
+nokon
+noka
+nokor
+noko
+nokre
+si
+sia
+sidan
+so
+somt
+somme
+um
+upp
+vere
+vore
+verte
+vort
+varte
+vart
\ No newline at end of file
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/portuguese.txt b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/portuguese.txt
new file mode 100644
index 0000000..98b4fdc
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/portuguese.txt
@@ -0,0 +1,203 @@
+de
+a
+o
+que
+e
+do
+da
+em
+um
+para
+com
+não
+uma
+os
+no
+se
+na
+por
+mais
+as
+dos
+como
+mas
+ao
+ele
+das
+à
+seu
+sua
+ou
+quando
+muito
+nos
+já
+eu
+também
+só
+pelo
+pela
+até
+isso
+ela
+entre
+depois
+sem
+mesmo
+aos
+seus
+quem
+nas
+me
+esse
+eles
+você
+essa
+num
+nem
+suas
+meu
+às
+minha
+numa
+pelos
+elas
+qual
+nós
+lhe
+deles
+essas
+esses
+pelas
+este
+dele
+tu
+te
+vocês
+vos
+lhes
+meus
+minhas
+teu
+tua
+teus
+tuas
+nosso
+nossa
+nossos
+nossas
+dela
+delas
+esta
+estes
+estas
+aquele
+aquela
+aqueles
+aquelas
+isto
+aquilo
+estou
+está
+estamos
+estão
+estive
+esteve
+estivemos
+estiveram
+estava
+estávamos
+estavam
+estivera
+estivéramos
+esteja
+estejamos
+estejam
+estivesse
+estivéssemos
+estivessem
+estiver
+estivermos
+estiverem
+hei
+há
+havemos
+hão
+houve
+houvemos
+houveram
+houvera
+houvéramos
+haja
+hajamos
+hajam
+houvesse
+houvéssemos
+houvessem
+houver
+houvermos
+houverem
+houverei
+houverá
+houveremos
+houverão
+houveria
+houveríamos
+houveriam
+sou
+somos
+são
+era
+éramos
+eram
+fui
+foi
+fomos
+foram
+fora
+fôramos
+seja
+sejamos
+sejam
+fosse
+fôssemos
+fossem
+for
+formos
+forem
+serei
+será
+seremos
+serão
+seria
+seríamos
+seriam
+tenho
+tem
+temos
+tém
+tinha
+tínhamos
+tinham
+tive
+teve
+tivemos
+tiveram
+tivera
+tivéramos
+tenha
+tenhamos
+tenham
+tivesse
+tivéssemos
+tivessem
+tiver
+tivermos
+tiverem
+terei
+terá
+teremos
+terão
+teria
+teríamos
+teriam
\ No newline at end of file
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/russian.txt b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/russian.txt
new file mode 100644
index 0000000..8a800b7
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/russian.txt
@@ -0,0 +1,151 @@
+и
+в
+во
+не
+что
+он
+на
+я
+с
+со
+как
+а
+то
+все
+она
+так
+его
+но
+да
+ты
+к
+у
+же
+вы
+за
+бы
+по
+только
+ее
+мне
+было
+вот
+от
+меня
+еще
+нет
+о
+из
+ему
+теперь
+когда
+даже
+ну
+вдруг
+ли
+если
+уже
+или
+ни
+быть
+был
+него
+до
+вас
+нибудь
+опять
+уж
+вам
+ведь
+там
+потом
+себя
+ничего
+ей
+может
+они
+тут
+где
+есть
+надо
+ней
+для
+мы
+тебя
+их
+чем
+была
+сам
+чтоб
+без
+будто
+чего
+раз
+тоже
+себе
+под
+будет
+ж
+тогда
+кто
+этот
+того
+потому
+этого
+какой
+совсем
+ним
+здесь
+этом
+один
+почти
+мой
+тем
+чтобы
+нее
+сейчас
+были
+куда
+зачем
+всех
+никогда
+можно
+при
+наконец
+два
+об
+другой
+хоть
+после
+над
+больше
+тот
+через
+эти
+нас
+про
+всего
+них
+какая
+много
+разве
+три
+эту
+моя
+впрочем
+хорошо
+свою
+этой
+перед
+иногда
+лучше
+чуть
+том
+нельзя
+такой
+им
+более
+всегда
+конечно
+всю
+между
\ No newline at end of file
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/spanish.txt b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/spanish.txt
new file mode 100644
index 0000000..94f493a
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/spanish.txt
@@ -0,0 +1,313 @@
+de
+la
+que
+el
+en
+y
+a
+los
+del
+se
+las
+por
+un
+para
+con
+no
+una
+su
+al
+lo
+como
+más
+pero
+sus
+le
+ya
+o
+este
+sí
+porque
+esta
+entre
+cuando
+muy
+sin
+sobre
+también
+me
+hasta
+hay
+donde
+quien
+desde
+todo
+nos
+durante
+todos
+uno
+les
+ni
+contra
+otros
+ese
+eso
+ante
+ellos
+e
+esto
+mí
+antes
+algunos
+qué
+unos
+yo
+otro
+otras
+otra
+él
+tanto
+esa
+estos
+mucho
+quienes
+nada
+muchos
+cual
+poco
+ella
+estar
+estas
+algunas
+algo
+nosotros
+mi
+mis
+tú
+te
+ti
+tu
+tus
+ellas
+nosotras
+vosostros
+vosostras
+os
+mío
+mía
+míos
+mías
+tuyo
+tuya
+tuyos
+tuyas
+suyo
+suya
+suyos
+suyas
+nuestro
+nuestra
+nuestros
+nuestras
+vuestro
+vuestra
+vuestros
+vuestras
+esos
+esas
+estoy
+estás
+está
+estamos
+estáis
+están
+esté
+estés
+estemos
+estéis
+estén
+estaré
+estarás
+estará
+estaremos
+estaréis
+estarán
+estaría
+estarías
+estaríamos
+estaríais
+estarían
+estaba
+estabas
+estábamos
+estabais
+estaban
+estuve
+estuviste
+estuvo
+estuvimos
+estuvisteis
+estuvieron
+estuviera
+estuvieras
+estuviéramos
+estuvierais
+estuvieran
+estuviese
+estuvieses
+estuviésemos
+estuvieseis
+estuviesen
+estando
+estado
+estada
+estados
+estadas
+estad
+he
+has
+ha
+hemos
+habéis
+han
+haya
+hayas
+hayamos
+hayáis
+hayan
+habré
+habrás
+habrá
+habremos
+habréis
+habrán
+habría
+habrías
+habríamos
+habríais
+habrían
+había
+habías
+habíamos
+habíais
+habían
+hube
+hubiste
+hubo
+hubimos
+hubisteis
+hubieron
+hubiera
+hubieras
+hubiéramos
+hubierais
+hubieran
+hubiese
+hubieses
+hubiésemos
+hubieseis
+hubiesen
+habiendo
+habido
+habida
+habidos
+habidas
+soy
+eres
+es
+somos
+sois
+son
+sea
+seas
+seamos
+seáis
+sean
+seré
+serás
+será
+seremos
+seréis
+serán
+sería
+serías
+seríamos
+seríais
+serían
+era
+eras
+éramos
+erais
+eran
+fui
+fuiste
+fue
+fuimos
+fuisteis
+fueron
+fuera
+fueras
+fuéramos
+fuerais
+fueran
+fuese
+fueses
+fuésemos
+fueseis
+fuesen
+sintiendo
+sentido
+sentida
+sentidos
+sentidas
+siente
+sentid
+tengo
+tienes
+tiene
+tenemos
+tenéis
+tienen
+tenga
+tengas
+tengamos
+tengáis
+tengan
+tendré
+tendrás
+tendrá
+tendremos
+tendréis
+tendrán
+tendría
+tendrías
+tendríamos
+tendríais
+tendrían
+tenía
+tenías
+teníamos
+teníais
+tenían
+tuve
+tuviste
+tuvo
+tuvimos
+tuvisteis
+tuvieron
+tuviera
+tuvieras
+tuviéramos
+tuvierais
+tuvieran
+tuviese
+tuvieses
+tuviésemos
+tuvieseis
+tuviesen
+teniendo
+tenido
+tenida
+tenidos
+tenidas
+tened
\ No newline at end of file
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/swedish.txt b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/swedish.txt
new file mode 100644
index 0000000..9fae31c
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/swedish.txt
@@ -0,0 +1,114 @@
+och
+det
+att
+i
+en
+jag
+hon
+som
+han
+på
+den
+med
+var
+sig
+för
+så
+till
+är
+men
+ett
+om
+hade
+de
+av
+icke
+mig
+du
+henne
+då
+sin
+nu
+har
+inte
+hans
+honom
+skulle
+hennes
+där
+min
+man
+ej
+vid
+kunde
+något
+från
+ut
+när
+efter
+upp
+vi
+dem
+vara
+vad
+över
+än
+dig
+kan
+sina
+här
+ha
+mot
+alla
+under
+någon
+eller
+allt
+mycket
+sedan
+ju
+denna
+själv
+detta
+åt
+utan
+varit
+hur
+ingen
+mitt
+ni
+bli
+blev
+oss
+din
+dessa
+några
+deras
+blir
+mina
+samma
+vilken
+er
+sådan
+vår
+blivit
+dess
+inom
+mellan
+sådant
+varför
+varje
+vilka
+ditt
+vem
+vilket
+sitta
+sådana
+vart
+dina
+vars
+vårt
+våra
+ert
+era
+vilkas
\ No newline at end of file
diff --git a/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/turkish.txt b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/turkish.txt
new file mode 100644
index 0000000..4e9708d
--- /dev/null
+++ b/flink-ml-lib/src/main/resources/org/apache/flink/ml/feature/stopwords/turkish.txt
@@ -0,0 +1,53 @@
+acaba
+ama
+aslında
+az
+bazı
+belki
+biri
+birkaç
+birşey
+biz
+bu
+çok
+çünkü
+da
+daha
+de
+defa
+diye
+eğer
+en
+gibi
+hem
+hep
+hepsi
+her
+hiç
+için
+ile
+ise
+kez
+ki
+kim
+mı
+mu
+mü
+nasıl
+ne
+neden
+nerde
+nerede
+nereye
+niçin
+niye
+o
+sanki
+şey
+siz
+şu
+tüm
+ve
+veya
+ya
+yani
\ No newline at end of file
diff --git a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/StopWordsRemoverTest.java b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/StopWordsRemoverTest.java
new file mode 100644
index 0000000..05ca70c
--- /dev/null
+++ b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/StopWordsRemoverTest.java
@@ -0,0 +1,492 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.feature;
+
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.builder.PipelineModel;
+import org.apache.flink.ml.feature.stopwordsremover.StopWordsRemover;
+import org.apache.flink.ml.util.TestUtils;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/** Tests {@link StopWordsRemover}. */
+public class StopWordsRemoverTest extends AbstractTestBase {
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+
+ @Before
+ public void before() {
+ Configuration config = new Configuration();
+ config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+ env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.getConfig().enableObjectReuse();
+ env.setParallelism(4);
+ env.enableCheckpointing(100);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+
+ tEnv = StreamTableEnvironment.create(env);
+ }
+
+ private static void verifyOutputResult(StopWordsRemover remover, Table inputTable) {
+ Table outputTable = remover.transform(inputTable)[0];
+
+ int expectedLength = IteratorUtils.toList(inputTable.execute().collect()).size();
+ int count = 0;
+ for (CloseableIterator<Row> it = outputTable.execute().collect(); it.hasNext(); ) {
+ Row row = it.next();
+ String[] expected = row.getFieldAs("expected");
+ String[] actual = row.getFieldAs("filtered");
+ assertArrayEquals(expected, actual);
+ count++;
+ }
+ assertEquals(expectedLength, count);
+ }
+
+ @Test
+ public void testParams() {
+ StopWordsRemover remover = new StopWordsRemover();
+ assertTrue(Arrays.asList(remover.getStopWords()).containsAll(Arrays.asList("i", "would")));
+ assertTrue(
+ Arrays.asList(Locale.US.toString(), Locale.getDefault().toString())
+ .contains(remover.getLocale()));
+ assertFalse(remover.getCaseSensitive());
+
+ remover.setInputCols("f1", "f2")
+ .setOutputCols("o1", "o2")
+ .setStopWords(StopWordsRemover.loadDefaultStopWords("turkish"))
+ .setLocale(Locale.US.toString())
+ .setCaseSensitive(true);
+
+ assertArrayEquals(new String[] {"f1", "f2"}, remover.getInputCols());
+ assertArrayEquals(new String[] {"o1", "o2"}, remover.getOutputCols());
+ assertTrue(
+ Arrays.asList(remover.getStopWords()).containsAll(Arrays.asList("acaba", "yani")));
+ assertEquals(Locale.US.toString(), remover.getLocale());
+ assertTrue(remover.getCaseSensitive());
+ }
+
+ @Test
+ public void testOutputSchema() {
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of(new String[] {"test", "test"}, new String[] {"test", "test"}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("raw", "expected");
+
+ StopWordsRemover remover =
+ new StopWordsRemover().setInputCols("raw").setOutputCols("filtered");
+
+ Table outputTable = remover.transform(inputTable)[0];
+
+ assertEquals(
+ Arrays.asList("raw", "expected", "filtered"),
+ outputTable.getResolvedSchema().getColumnNames());
+ }
+
+ @Test
+ public void testTransform() {
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of(new String[] {"test", "test"}, new String[] {"test", "test"}),
+ Row.of(new String[] {"a", "b", "c", "d"}, new String[] {"b", "c", "d"}),
+ Row.of(new String[] {"a", "the", "an"}, new String[] {}),
+ Row.of(new String[] {"A", "The", "AN"}, new String[] {}),
+ Row.of(new String[] {null}, new String[] {null}),
+ Row.of(new String[] {}, new String[] {}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("raw", "expected");
+
+ StopWordsRemover remover =
+ new StopWordsRemover().setInputCols("raw").setOutputCols("filtered");
+
+ verifyOutputResult(remover, inputTable);
+ }
+
+ @Test
+ public void testTransformWithStopWordsList() {
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of(new String[] {"test", "test"}, new String[] {}),
+ Row.of(new String[] {"a", "b", "c", "d"}, new String[] {"b", "c", "d"}),
+ Row.of(new String[] {"a", "the", "an"}, new String[] {}),
+ Row.of(new String[] {"A", "The", "AN"}, new String[] {}),
+ Row.of(new String[] {null}, new String[] {}),
+ Row.of(new String[] {}, new String[] {}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("raw", "expected");
+
+ StopWordsRemover remover =
+ new StopWordsRemover()
+ .setInputCols("raw")
+ .setOutputCols("filtered")
+ .setStopWords(new String[] {"test", "a", "an", "the", null});
+
+ verifyOutputResult(remover, inputTable);
+ }
+
+ @Test
+ public void testTransformWithLocaledInputCaseInsensitive() {
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of(new String[] {"mİlk", "and", "nuts"}, new String[] {"and", "nuts"}),
+ Row.of(
+ new String[] {"cookIe", "and", "nuts"},
+ new String[] {"cookIe", "and", "nuts"}),
+ Row.of(new String[] {null}, new String[] {null}),
+ Row.of(new String[] {}, new String[] {}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("raw", "expected");
+
+ StopWordsRemover remover =
+ new StopWordsRemover()
+ .setInputCols("raw")
+ .setOutputCols("filtered")
+ .setStopWords(new String[] {"milk", "cookie"})
+ .setCaseSensitive(false)
+ .setLocale("tr");
+
+ verifyOutputResult(remover, inputTable);
+ }
+
+ @Test
+ public void testTransformWithLocaledInputCaseSensitive() {
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of(
+ new String[] {"mİlk", "and", "nuts"},
+ new String[] {"mİlk", "and", "nuts"}),
+ Row.of(
+ new String[] {"cookIe", "and", "nuts"},
+ new String[] {"cookIe", "and", "nuts"}),
+ Row.of(new String[] {null}, new String[] {null}),
+ Row.of(new String[] {}, new String[] {}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("raw", "expected");
+
+ StopWordsRemover remover =
+ new StopWordsRemover()
+ .setInputCols("raw")
+ .setOutputCols("filtered")
+ .setStopWords(new String[] {"milk", "cookie"})
+ .setCaseSensitive(true)
+ .setLocale("tr");
+
+ verifyOutputResult(remover, inputTable);
+ }
+
+ @Test
+ public void testInvalidLocale() {
+ try {
+ new StopWordsRemover().setLocale("rt");
+ fail();
+ } catch (Exception e) {
+ assertEquals(IllegalArgumentException.class, e.getClass());
+ assertEquals("Parameter locale is given an invalid value rt", e.getMessage());
+ }
+ }
+
+ @Test
+ public void testAvailableLocales() {
+ assertTrue(StopWordsRemover.getAvailableLocales().contains(Locale.US.toString()));
+
+ StopWordsRemover remover = new StopWordsRemover();
+ for (String locale : StopWordsRemover.getAvailableLocales()) {
+ remover.setLocale(locale);
+ }
+ }
+
+ @Test
+ public void testTransformCaseSensitive() {
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of(new String[] {"A"}, new String[] {"A"}),
+ Row.of(new String[] {"The", "the"}, new String[] {"The"}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("raw", "expected");
+
+ StopWordsRemover remover =
+ new StopWordsRemover()
+ .setInputCols("raw")
+ .setOutputCols("filtered")
+ .setCaseSensitive(true);
+
+ verifyOutputResult(remover, inputTable);
+ }
+
+ @Test
+ public void testDefaultStopWordsOfSupportedLanguagesNotEmtpy() {
+ List<String> supportedLanguages =
+ Arrays.asList(
+ "danish",
+ "dutch",
+ "english",
+ "finnish",
+ "french",
+ "german",
+ "hungarian",
+ "italian",
+ "norwegian",
+ "portuguese",
+ "russian",
+ "spanish",
+ "swedish",
+ "turkish");
+ for (String language : supportedLanguages) {
+ assertTrue(StopWordsRemover.loadDefaultStopWords(language).length > 0);
+ }
+ }
+
+ @Test
+ public void testTransformWithLanguageSelection() {
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of(new String[] {"acaba", "ama", "biri"}, new String[] {}),
+ Row.of(new String[] {"hep", "her", "scala"}, new String[] {"scala"}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("raw", "expected");
+
+ StopWordsRemover remover =
+ new StopWordsRemover()
+ .setInputCols("raw")
+ .setOutputCols("filtered")
+ .setStopWords(StopWordsRemover.loadDefaultStopWords("turkish"));
+
+ verifyOutputResult(remover, inputTable);
+ }
+
+ @Test
+ public void testTransformWithIgnoredWords() {
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of(
+ new String[] {"python", "scala", "a"},
+ new String[] {"python", "scala", "a"}),
+ Row.of(
+ new String[] {"Python", "Scala", "swift"},
+ new String[] {"Python", "Scala", "swift"}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("raw", "expected");
+
+ Set<String> stopWords =
+ new HashSet<>(Arrays.asList(StopWordsRemover.loadDefaultStopWords("english")));
+ stopWords.remove("a");
+
+ StopWordsRemover remover =
+ new StopWordsRemover()
+ .setInputCols("raw")
+ .setOutputCols("filtered")
+ .setStopWords(stopWords.toArray(new String[0]));
+
+ verifyOutputResult(remover, inputTable);
+ }
+
+ @Test
+ public void testTransformWithAdditionalWords() {
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of(new String[] {"python", "scala", "a"}, new String[] {}),
+ Row.of(new String[] {"Python", "Scala", "swift"}, new String[] {"swift"}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("raw", "expected");
+
+ Set<String> stopWords =
+ new HashSet<>(Arrays.asList(StopWordsRemover.loadDefaultStopWords("english")));
+ stopWords.addAll(Arrays.asList("python", "scala"));
+
+ StopWordsRemover remover =
+ new StopWordsRemover()
+ .setInputCols("raw")
+ .setOutputCols("filtered")
+ .setStopWords(stopWords.toArray(new String[0]));
+
+ verifyOutputResult(remover, inputTable);
+ }
+
+ @Test
+ public void testSaveLoadAndTransform() throws Exception {
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of(new String[] {"test", "test"}, new String[] {"test", "test"}),
+ Row.of(new String[] {"a", "b", "c", "d"}, new String[] {"b", "c", "d"}),
+ Row.of(new String[] {"a", "the", "an"}, new String[] {}),
+ Row.of(new String[] {"A", "The", "AN"}, new String[] {}),
+ Row.of(new String[] {null}, new String[] {null}),
+ Row.of(new String[] {}, new String[] {}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("raw", "expected");
+
+ StopWordsRemover remover =
+ new StopWordsRemover().setInputCols("raw").setOutputCols("filtered");
+
+ StopWordsRemover loadedRemover =
+ TestUtils.saveAndReload(
+ tEnv, remover, TEMPORARY_FOLDER.newFolder().getAbsolutePath());
+
+ verifyOutputResult(loadedRemover, inputTable);
+ }
+
+ @Test
+ public void testOutputColumnAlreadyExists() {
+ try {
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of(new String[] {"The", "the", "swift"}, new String[] {"swift"}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("raw", "expected");
+ StopWordsRemover remover =
+ new StopWordsRemover().setInputCols("raw").setOutputCols("expected");
+ remover.transform(inputTable);
+ fail();
+ } catch (Exception e) {
+ assertEquals(ValidationException.class, e.getClass());
+ assertEquals("Ambiguous column name: expected", e.getMessage());
+ }
+ }
+
+ @Test
+ public void testTransformMultipleColumns() {
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of(
+ new String[] {"test", "test"},
+ new String[] {"test1", "test2"},
+ new String[] {},
+ new String[] {"test1", "test2"}),
+ Row.of(
+ new String[] {"a", "b", "c", "d"},
+ new String[] {"a", "b"},
+ new String[] {"b", "c", "d"},
+ new String[] {"b"}),
+ Row.of(
+ new String[] {"a", "the", "an"},
+ new String[] {"a", "the", "test1"},
+ new String[] {},
+ new String[] {"test1"}),
+ Row.of(
+ new String[] {"A", "The", "AN"},
+ new String[] {"A", "The", "AN"},
+ new String[] {},
+ new String[] {}),
+ Row.of(
+ new String[] {null},
+ new String[] {null},
+ new String[] {null},
+ new String[] {null}),
+ Row.of(new String[] {}, new String[] {}, new String[] {}, new String[] {}));
+ Table inputTable =
+ tEnv.fromDataStream(inputStream).as("raw1", "raw2", "expected1", "expected2");
+
+ StopWordsRemover remover =
+ new StopWordsRemover()
+ .setInputCols("raw1", "raw2")
+ .setOutputCols("filtered1", "filtered2")
+ .setStopWords(new String[] {"test", "a", "an", "the"});
+
+ Table outputTable = remover.transform(inputTable)[0];
+
+ int expectedLength = IteratorUtils.toList(inputTable.execute().collect()).size();
+ int count = 0;
+ for (CloseableIterator<Row> it = outputTable.execute().collect(); it.hasNext(); ) {
+ Row row = it.next();
+
+ String[] expected = row.getFieldAs("expected1");
+ String[] actual = row.getFieldAs("filtered1");
+ assertArrayEquals(expected, actual);
+
+ expected = row.getFieldAs("expected2");
+ actual = row.getFieldAs("filtered2");
+ assertArrayEquals(expected, actual);
+
+ count++;
+ }
+ assertEquals(expectedLength, count);
+ }
+
+ @Test
+ public void testCompareSingleMultipleRemoverInPipeline() {
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of(new String[] {"test", "test"}, new String[] {"test1", "test2"}),
+ Row.of(new String[] {"a", "b", "c", "d"}, new String[] {"a", "b"}),
+ Row.of(new String[] {"a", "the", "an"}, new String[] {"a", "the", "test1"}),
+ Row.of(new String[] {"A", "The", "AN"}, new String[] {"A", "The", "AN"}),
+ Row.of(new String[] {null}, new String[] {null}),
+ Row.of(new String[] {}, new String[] {}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("input1", "input2");
+
+ PipelineModel multiColsPipeline =
+ new PipelineModel(
+ Collections.singletonList(
+ new StopWordsRemover()
+ .setInputCols("input1", "input2")
+ .setOutputCols("output1", "output2")));
+
+ PipelineModel singleColPipeline =
+ new PipelineModel(
+ Arrays.asList(
+ new StopWordsRemover()
+ .setInputCols("input1")
+ .setOutputCols("output1"),
+ new StopWordsRemover()
+ .setInputCols("input2")
+ .setOutputCols("output2")));
+
+ assertEquals(
+ new HashSet<Row>(
+ IteratorUtils.toList(
+ multiColsPipeline.transform(inputTable)[0].execute().collect())),
+ new HashSet<Row>(
+ IteratorUtils.toList(
+ singleColPipeline.transform(inputTable)[0].execute().collect())));
+ }
+
+ @Test
+ public void testMismatchInputOutputCols() {
+ try {
+ DataStream<Row> inputStream =
+ env.fromElements(
+ Row.of(new String[] {"The", "the", "swift"}, new String[] {"swift"}));
+ Table inputTable = tEnv.fromDataStream(inputStream).as("raw", "expected");
+ StopWordsRemover remover =
+ new StopWordsRemover()
+ .setInputCols("raw")
+ .setOutputCols("expected1", "expected2");
+ remover.transform(inputTable);
+ fail();
+ } catch (Exception e) {
+ assertEquals(IllegalArgumentException.class, e.getClass());
+ assertNull(e.getMessage());
+ }
+ }
+}
diff --git a/flink-ml-python/pyflink/examples/ml/feature/stopwordsremover_example.py b/flink-ml-python/pyflink/examples/ml/feature/stopwordsremover_example.py
new file mode 100644
index 0000000..48e64c9
--- /dev/null
+++ b/flink-ml-python/pyflink/examples/ml/feature/stopwordsremover_example.py
@@ -0,0 +1,58 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+# Simple program that creates a StopWordsRemover instance and uses it for feature
+# engineering.
+
+from pyflink.common import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.ml.lib.feature.stopwordsremover import StopWordsRemover
+from pyflink.table import StreamTableEnvironment
+
+# create a new StreamExecutionEnvironment
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# create a StreamTableEnvironment
+t_env = StreamTableEnvironment.create(env)
+
+# generate input data
+input_table = t_env.from_data_stream(
+ env.from_collection([
+ (["test", "test"],),
+ (["a", "b", "c", "d"],),
+ (["a", "the", "an"],),
+ (["A", "The", "AN"],),
+ ([None],),
+ ([],),
+ ],
+ type_info=Types.ROW_NAMED(
+ ['input'],
+ [Types.OBJECT_ARRAY(Types.STRING())])))
+
+# create a StopWordsRemover object and initialize its parameters
+remover = StopWordsRemover().set_input_cols('input').set_output_cols('output')
+
+# use the StopWordsRemover for feature engineering
+output_table = remover.transform(input_table)[0]
+
+# extract and display the results
+field_names = output_table.get_schema().get_field_names()
+for result in t_env.to_data_stream(output_table).execute_and_collect():
+ input_value = result[field_names.index('input')]
+ output_value = result[field_names.index('output')]
+ print('Input Value: ' + str(input_value) + '\tOutput Value: ' + str(output_value))
diff --git a/flink-ml-python/pyflink/ml/lib/feature/stopwordsremover.py b/flink-ml-python/pyflink/ml/lib/feature/stopwordsremover.py
new file mode 100644
index 0000000..872e775
--- /dev/null
+++ b/flink-ml-python/pyflink/ml/lib/feature/stopwordsremover.py
@@ -0,0 +1,148 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+import typing
+from typing import Tuple
+
+from pyflink.java_gateway import get_gateway
+from pyflink.ml.core.param import Param, StringArrayParam, BooleanParam, StringParam
+from pyflink.ml.core.wrapper import JavaWithParams
+from pyflink.ml.lib.feature.common import JavaFeatureTransformer
+from pyflink.ml.lib.param import HasInputCols, HasOutputCols
+
+
+def _load_default_stop_words(language: str) -> Tuple[str, ...]:
+ return tuple(*[get_gateway().jvm.org.apache.flink.ml.feature.
+ stopwordsremover.StopWordsRemover.loadDefaultStopWords(language)])
+
+
+def _get_default_or_us() -> str:
+ return get_gateway().jvm.org.apache.flink.ml.feature. \
+ stopwordsremover.StopWordsRemover.getDefaultOrUS()
+
+
+def _get_available_locales() -> set:
+ return {*get_gateway().jvm.org.apache.flink.ml.feature.
+ stopwordsremover.StopWordsRemover.getAvailableLocales()}
+
+
+class _StopWordsRemoverParams(
+ JavaWithParams,
+ HasInputCols,
+ HasOutputCols
+):
+ """
+ Params for :class:`StopWordsRemover`.
+ """
+
+ STOP_WORDS: Param[Tuple[str, ...]] = StringArrayParam(
+ "stop_words",
+ "The words to be filtered out.",
+ _load_default_stop_words('english'))
+
+ CASE_SENSITIVE: Param[bool] = BooleanParam(
+ "case_sensitive",
+ "Whether to do a case-sensitive comparison over the stop words.",
+ False
+ )
+
+ LOCALE: Param[str] = StringParam(
+ "locale",
+ "Locale of the input for case insensitive matching. Ignored when caseSensitive is true.",
+ _get_default_or_us())
+
+ def __init__(self, java_params):
+ super(_StopWordsRemoverParams, self).__init__(java_params)
+
+ def set_stop_words(self, *value: str):
+ return typing.cast(_StopWordsRemoverParams, self.set(self.STOP_WORDS, value))
+
+ def set_case_sensitive(self, value: bool):
+ return typing.cast(_StopWordsRemoverParams, self.set(self.CASE_SENSITIVE, value))
+
+ def set_locale(self, value: str):
+ return typing.cast(_StopWordsRemoverParams, self.set(self.LOCALE, value))
+
+ def get_stop_words(self) -> Tuple[str, ...]:
+ return self.get(self.STOP_WORDS)
+
+ def get_case_sensitive(self) -> bool:
+ return self.get(self.CASE_SENSITIVE)
+
+ def get_locale(self) -> str:
+ return self.get(self.LOCALE)
+
+ @property
+ def stop_words(self):
+ return self.get_stop_words()
+
+ @property
+ def case_sensitive(self):
+ return self.get_case_sensitive()
+
+ @property
+ def locale(self):
+ return self.get_locale()
+
+
+class StopWordsRemover(JavaFeatureTransformer, _StopWordsRemoverParams):
+ """
+ A feature transformer that filters out stop words from input.
+
+ Note: null values from input array are preserved unless adding null to stopWords explicitly.
+
+ See Also: http://en.wikipedia.org/wiki/Stop_words
+ """
+
+ def __init__(self, java_model=None):
+ super(StopWordsRemover, self).__init__(java_model)
+
+ @classmethod
+ def _java_transformer_package_name(cls) -> str:
+ return "stopwordsremover"
+
+ @classmethod
+ def _java_transformer_class_name(cls) -> str:
+ return "StopWordsRemover"
+
+ @classmethod
+ def load_default_stop_words(cls, language: str):
+ """
+ Loads the default stop words for the given language.
+
+ Supported languages: danish, dutch, english, finnish, french, german, hungarian, italian,
+ norwegian, portuguese, russian, spanish, swedish, turkish
+
+ See Also: http://anoncvs.postgresql.org/cvsweb.cgi/pgsql/src/backend/snowball/stopwords/
+ """
+ return _load_default_stop_words(language)
+
+ @classmethod
+ def get_default_or_us(cls):
+ """
+ Returns system default locale, or "en_US" if the default locale is not available. The locale
+ is returned as a String.
+ """
+ return _get_default_or_us()
+
+ @classmethod
+ def get_available_locales(cls):
+ """
+ Returns a set of all installed locales. It must contain at least a Locale
+ instance equal to "en_US". The locales are returned as Strings.
+ """
+ return _get_available_locales()
diff --git a/flink-ml-python/pyflink/ml/lib/feature/tests/test_stopwordsremover.py b/flink-ml-python/pyflink/ml/lib/feature/tests/test_stopwordsremover.py
new file mode 100644
index 0000000..26991a9
--- /dev/null
+++ b/flink-ml-python/pyflink/ml/lib/feature/tests/test_stopwordsremover.py
@@ -0,0 +1,115 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+from pyflink.common import Types
+from pyflink.ml.lib.feature.stopwordsremover import StopWordsRemover
+from pyflink.ml.tests.test_utils import PyFlinkMLTestCase
+
+
+class StopWordsRemoverTest(PyFlinkMLTestCase):
+ def setUp(self):
+ super(StopWordsRemoverTest, self).setUp()
+ self.input_table = self.t_env.from_data_stream(
+ self.env.from_collection([
+ (["test", "test"], ["test", "test"]),
+ (["a", "b", "c", "d"], ["b", "c", "d"]),
+ (["a", "the", "an"], []),
+ (["A", "The", "AN"], []),
+ ([None], [None]),
+ ([], []),
+ ],
+ type_info=Types.ROW_NAMED(
+ ['raw', 'expected'],
+ [Types.OBJECT_ARRAY(Types.STRING()), Types.OBJECT_ARRAY(Types.STRING())])))
+
+ def test_param(self):
+ remover = StopWordsRemover()
+
+ self.assertTrue({'i', 'would'} <= set(remover.stop_words))
+ self.assertEqual(remover.locale, StopWordsRemover.get_default_or_us())
+ self.assertFalse(remover.case_sensitive)
+
+ remover.set_input_cols('f1', 'f2') \
+ .set_output_cols('o1', 'o2') \
+ .set_stop_words(*StopWordsRemover.load_default_stop_words('turkish')) \
+ .set_locale('en_US') \
+ .set_case_sensitive(True)
+
+ self.assertEqual(('f1', 'f2'), remover.input_cols)
+ self.assertEqual(('o1', 'o2'), remover.output_cols)
+ self.assertTrue({'acaba', 'yani'} <= set(remover.stop_words))
+ self.assertEqual('en_US', remover.locale)
+ self.assertTrue(remover.case_sensitive)
+
+ def test_output_schema(self):
+ remover = StopWordsRemover().set_input_cols('raw').set_output_cols('filtered')
+ output_table = remover.transform(self.input_table)[0]
+
+ self.assertEqual(
+ ['raw', 'expected', 'filtered'],
+ output_table.get_schema().get_field_names())
+
+ def test_transform(self):
+ remover = StopWordsRemover().set_input_cols('raw').set_output_cols('filtered')
+ self.verify_output_result(remover, self.input_table)
+
+ def test_save_load_and_transform(self):
+ remover = StopWordsRemover().set_input_cols('raw').set_output_cols('filtered')
+ loaded_remover = self.save_and_reload(remover)
+ self.verify_output_result(loaded_remover, self.input_table)
+
+ def test_available_locales(self):
+ self.assertTrue('en_US' in StopWordsRemover.get_available_locales())
+
+ remover = StopWordsRemover()
+ for locale in StopWordsRemover.get_available_locales():
+ remover.set_locale(locale)
+
+ def test_default_language_stop_words_not_empty(self):
+ supported_languages = {
+ "danish",
+ "dutch",
+ "english",
+ "finnish",
+ "french",
+ "german",
+ "hungarian",
+ "italian",
+ "norwegian",
+ "portuguese",
+ "russian",
+ "spanish",
+ "swedish",
+ "turkish"
+ }
+
+ for language in supported_languages:
+ self.assertTrue(len(StopWordsRemover.load_default_stop_words(language)) > 0)
+
+ def verify_output_result(self, remover, input_table):
+ output_table = remover.transform(input_table)[0]
+ results = [x for x in self.t_env.to_data_stream(output_table).execute_and_collect()]
+ expected_length = len([_ for _ in
+ self.t_env.to_data_stream(input_table).execute_and_collect()])
+ self.assertEqual(expected_length, len(results))
+
+ field_names = output_table.get_schema().get_field_names()
+ for result in results:
+ expected = result[field_names.index('expected')]
+ actual = result[field_names.index('filtered')]
+ self.assertEqual(expected, actual)
diff --git a/flink-ml-python/pyflink/ml/tests/test_utils.py b/flink-ml-python/pyflink/ml/tests/test_utils.py
index bbfb82f..eca6d37 100644
--- a/flink-ml-python/pyflink/ml/tests/test_utils.py
+++ b/flink-ml-python/pyflink/ml/tests/test_utils.py
@@ -83,22 +83,7 @@ class PyFlinkMLTestCase(unittest.TestCase):
@classmethod
def _load_dependency_jars(cls):
- from pyflink.ml.version import __version__
-
- flink_version = __version__.replace(".dev0", "-SNAPSHOT")
this_directory = os.path.abspath(os.path.dirname(__file__))
- FLINK_ML_HOME = os.path.abspath(os.path.join(
- this_directory,
- "../../../../flink-ml-dist/target/flink-ml-%s-bin/flink-ml-%s" %
- (flink_version, flink_version)))
- FLINK_ML_LIB_PATH = os.path.join(FLINK_ML_HOME, "lib")
- if not os.path.isdir(FLINK_ML_LIB_PATH):
- raise Exception("You need to build Flink ML with maven you can run: "
- "mvn -DskipTests clean package")
-
- for file in os.listdir(FLINK_ML_LIB_PATH):
- if file.endswith('.jar'):
- add_jars_to_context_class_loader(["file://{0}/{1}".format(FLINK_ML_LIB_PATH, file)])
# load flink-ml-lib/flink-ml-lib-*-tests.jar
FLINK_ML_LIB_SOURCE_PATH = os.path.abspath(os.path.join(
diff --git a/flink-ml-python/src/main/java/org/apache/flink/ml/python/PythonBridgeUtils.java b/flink-ml-python/src/main/java/org/apache/flink/ml/python/PythonBridgeUtils.java
index 9b9596e..36eca36 100644
--- a/flink-ml-python/src/main/java/org/apache/flink/ml/python/PythonBridgeUtils.java
+++ b/flink-ml-python/src/main/java/org/apache/flink/ml/python/PythonBridgeUtils.java
@@ -65,7 +65,7 @@ public class PythonBridgeUtils {
null, null);
if (obj == null) {
- return new byte[0];
+ return pickler.dumps(null);
} else {
if (dataType instanceof SqlTimeTypeInfo) {
SqlTimeTypeInfo<?> sqlTimeTypeInfo =
diff --git a/pom.xml b/pom.xml
index 7d15c15..21875e7 100644
--- a/pom.xml
+++ b/pom.xml
@@ -534,6 +534,8 @@ under the License.
<exclude>**/LICENSE*</exclude>
<!-- artifacts created during release process -->
<exclude>release/**</exclude>
+ <!-- resource files used in production code -->
+ <exclude>**/src/main/resources/**</exclude>
</excludes>
</configuration>
</plugin>