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/10/09 02:56:48 UTC
[flink-ml] branch master updated: [FLINK-29409] Add Transformer and Estimator for VarianceThresholdSelector
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 ebc80ce [FLINK-29409] Add Transformer and Estimator for VarianceThresholdSelector
ebc80ce is described below
commit ebc80ceab35ebee33e33f85704e3f4d6a85a9fd4
Author: JiangXin <ji...@alibaba-inc.com>
AuthorDate: Sun Oct 9 10:56:42 2022 +0800
[FLINK-29409] Add Transformer and Estimator for VarianceThresholdSelector
This closes #158.
---
.../operators/feature/variancethresholdselector.md | 182 ++++++++++++++
.../feature/VarianceThresholdSelectorExample.java | 78 ++++++
.../VarianceThresholdSelector.java | 172 +++++++++++++
.../VarianceThresholdSelectorModel.java | 170 +++++++++++++
.../VarianceThresholdSelectorModelData.java | 119 +++++++++
.../VarianceThresholdSelectorModelParams.java | 30 +++
.../VarianceThresholdSelectorParams.java | 47 ++++
.../ml/feature/VarianceThresholdSelectorTest.java | 279 +++++++++++++++++++++
.../feature/variancethresholdselector_example.py | 67 +++++
.../tests/test_variancethresholdselector.py | 138 ++++++++++
.../ml/lib/feature/variancethresholdselector.py | 95 +++++++
flink-ml-python/pyflink/ml/tests/test_utils.py | 14 ++
12 files changed, 1391 insertions(+)
diff --git a/docs/content/docs/operators/feature/variancethresholdselector.md b/docs/content/docs/operators/feature/variancethresholdselector.md
new file mode 100644
index 0000000..bc839ab
--- /dev/null
+++ b/docs/content/docs/operators/feature/variancethresholdselector.md
@@ -0,0 +1,182 @@
+---
+title: "Variance Threshold Selector"
+weight: 1
+type: docs
+aliases:
+- /operators/feature/variancethresholdselector.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.
+-->
+
+## Variance Threshold Selector
+
+Variance Threshold Selector is a selector that removes low-variance features.
+Features with a variance not greater than the varianceThreshold will be removed.
+If not set, varianceThreshold defaults to 0, which means only features with
+variance 0 (i.e. features that have the same value in all samples) will be removed.
+
+### Input Columns
+
+| Param name | Type | Default | Description |
+|:------------|:-------|:-------------|:----------------|
+| featuresCol | Vector | `"features"` | Feature vector. |
+
+### Output Columns
+
+| Param name | Type | Default | Description |
+|:-----------|:-------|:-----------|:-----------------|
+| outputCol | Vector | `"output"` | Scaled features. |
+
+### Parameters
+
+| Key | Default | Type | Required | Description |
+|-------------------|--------------|--------|----------|---------------------------------------------------------------------------|
+| featuresCol | `"features"` | String | no | Features column name. |
+| outputCol | `"output"` | String | no | Output column name. |
+| varianceThreshold | `0.0` | Double | no | Features with a variance not greater than this threshold will be removed. |
+
+### Examples
+
+{{< tabs examples >}}
+
+{{< tab "Java">}}
+
+```java
+import org.apache.flink.ml.feature.variancethresholdselector.VarianceThresholdSelector;
+import org.apache.flink.ml.feature.variancethresholdselector.VarianceThresholdSelectorModel;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vectors;
+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;
+
+/**
+ * Simple program that trains a {@link VarianceThresholdSelector} model and uses it for feature
+ * selection.
+ */
+public class VarianceThresholdSelectorExample {
+
+ public static void main(String[] args) {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ // Generates input training and prediction data.
+ DataStream<Row> trainStream =
+ env.fromElements(
+ Row.of(1, Vectors.dense(5.0, 7.0, 0.0, 7.0, 6.0, 0.0)),
+ Row.of(2, Vectors.dense(0.0, 9.0, 6.0, 0.0, 5.0, 9.0)),
+ Row.of(3, Vectors.dense(0.0, 9.0, 3.0, 0.0, 5.0, 5.0)),
+ Row.of(4, Vectors.dense(1.0, 9.0, 8.0, 5.0, 7.0, 4.0)),
+ Row.of(5, Vectors.dense(9.0, 8.0, 6.0, 5.0, 4.0, 4.0)),
+ Row.of(6, Vectors.dense(6.0, 9.0, 7.0, 0.0, 2.0, 0.0)));
+ Table trainTable = tEnv.fromDataStream(trainStream).as("id", "features");
+
+ // Create a VarianceThresholdSelector object and initialize its parameters
+ double threshold = 8.0;
+ VarianceThresholdSelector varianceThresholdSelector =
+ new VarianceThresholdSelector()
+ .setVarianceThreshold(threshold)
+ .setFeaturesCol("features");
+
+ // Train the VarianceThresholdSelector model.
+ VarianceThresholdSelectorModel model = varianceThresholdSelector.fit(trainTable);
+
+ // Uses the VarianceThresholdSelector model for predictions.
+ Table outputTable = model.transform(trainTable)[0];
+
+ // Extracts and displays the results.
+ System.out.printf("Variance Threshold: %s\n", threshold);
+ for (CloseableIterator<Row> it = outputTable.execute().collect(); it.hasNext(); ) {
+ Row row = it.next();
+ DenseVector inputValue =
+ (DenseVector) row.getField(varianceThresholdSelector.getFeaturesCol());
+ DenseVector outputValue =
+ (DenseVector) row.getField(varianceThresholdSelector.getOutputCol());
+ System.out.printf(
+ "Original Features: %-15s\tSelected Features: %s\n", inputValue, outputValue);
+ }
+ }
+}
+
+```
+
+{{< /tab>}}
+
+{{< tab "Python">}}
+
+```python
+# Simple program that trains a VarianceThresholdSelector model and uses it for feature
+# selection.
+
+from pyflink.common import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.ml.core.linalg import Vectors, DenseVectorTypeInfo
+from pyflink.ml.lib.feature.variancethresholdselector import VarianceThresholdSelector
+from pyflink.table import StreamTableEnvironment
+
+# create a new StreamExecutionEnvironment
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# create a StreamTableEnvironment
+t_env = StreamTableEnvironment.create(env)
+
+# generate input training and prediction data
+train_data = t_env.from_data_stream(
+ env.from_collection([
+ (1, Vectors.dense(5.0, 7.0, 0.0, 7.0, 6.0, 0.0),),
+ (2, Vectors.dense(0.0, 9.0, 6.0, 0.0, 5.0, 9.0),),
+ (3, Vectors.dense(0.0, 9.0, 3.0, 0.0, 5.0, 5.0),),
+ (4, Vectors.dense(1.0, 9.0, 8.0, 5.0, 7.0, 4.0),),
+ (5, Vectors.dense(9.0, 8.0, 6.0, 5.0, 4.0, 4.0),),
+ (6, Vectors.dense(6.0, 9.0, 7.0, 0.0, 2.0, 0.0),),
+ ],
+ type_info=Types.ROW_NAMED(
+ ['id', 'features'],
+ [Types.INT(), DenseVectorTypeInfo()])
+ ))
+
+# create a VarianceThresholdSelector object and initialize its parameters
+threshold = 8.0
+variance_thread_selector = VarianceThresholdSelector()\
+ .set_features_col("features")\
+ .set_variance_threshold(threshold)
+
+# train the VarianceThresholdSelector model
+model = variance_thread_selector.fit(train_data)
+
+# use the VarianceThresholdSelector model for predictions
+output = model.transform(train_data)[0]
+
+# extract and display the results
+print("Variance Threshold: " + str(threshold))
+field_names = output.get_schema().get_field_names()
+for result in t_env.to_data_stream(output).execute_and_collect():
+ input_value = result[field_names.index(variance_thread_selector.get_features_col())]
+ output_value = result[field_names.index(variance_thread_selector.get_output_col())]
+ print('Original Features: ' + str(input_value) + ' \tSelected Features: ' + str(output_value))
+
+```
+
+{{< /tab>}}
+
+{{< /tabs>}}
diff --git a/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/feature/VarianceThresholdSelectorExample.java b/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/feature/VarianceThresholdSelectorExample.java
new file mode 100644
index 0000000..9783fd8
--- /dev/null
+++ b/flink-ml-examples/src/main/java/org/apache/flink/ml/examples/feature/VarianceThresholdSelectorExample.java
@@ -0,0 +1,78 @@
+/*
+ * 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.variancethresholdselector.VarianceThresholdSelector;
+import org.apache.flink.ml.feature.variancethresholdselector.VarianceThresholdSelectorModel;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vectors;
+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;
+
+/**
+ * Simple program that trains a {@link VarianceThresholdSelector} model and uses it for feature
+ * selection.
+ */
+public class VarianceThresholdSelectorExample {
+
+ public static void main(String[] args) {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ // Generates input training and prediction data.
+ DataStream<Row> trainStream =
+ env.fromElements(
+ Row.of(1, Vectors.dense(5.0, 7.0, 0.0, 7.0, 6.0, 0.0)),
+ Row.of(2, Vectors.dense(0.0, 9.0, 6.0, 0.0, 5.0, 9.0)),
+ Row.of(3, Vectors.dense(0.0, 9.0, 3.0, 0.0, 5.0, 5.0)),
+ Row.of(4, Vectors.dense(1.0, 9.0, 8.0, 5.0, 7.0, 4.0)),
+ Row.of(5, Vectors.dense(9.0, 8.0, 6.0, 5.0, 4.0, 4.0)),
+ Row.of(6, Vectors.dense(6.0, 9.0, 7.0, 0.0, 2.0, 0.0)));
+ Table trainTable = tEnv.fromDataStream(trainStream).as("id", "features");
+
+ // Create a VarianceThresholdSelector object and initialize its parameters
+ double threshold = 8.0;
+ VarianceThresholdSelector varianceThresholdSelector =
+ new VarianceThresholdSelector()
+ .setVarianceThreshold(threshold)
+ .setFeaturesCol("features");
+
+ // Train the VarianceThresholdSelector model.
+ VarianceThresholdSelectorModel model = varianceThresholdSelector.fit(trainTable);
+
+ // Uses the VarianceThresholdSelector model for predictions.
+ Table outputTable = model.transform(trainTable)[0];
+
+ // Extracts and displays the results.
+ System.out.printf("Variance Threshold: %s\n", threshold);
+ for (CloseableIterator<Row> it = outputTable.execute().collect(); it.hasNext(); ) {
+ Row row = it.next();
+ DenseVector inputValue =
+ (DenseVector) row.getField(varianceThresholdSelector.getFeaturesCol());
+ DenseVector outputValue =
+ (DenseVector) row.getField(varianceThresholdSelector.getOutputCol());
+ System.out.printf(
+ "Original Features: %-15s\tSelected Features: %s\n", inputValue, outputValue);
+ }
+ }
+}
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelector.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelector.java
new file mode 100644
index 0000000..90db8fc
--- /dev/null
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelector.java
@@ -0,0 +1,172 @@
+/*
+ * 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.variancethresholdselector;
+
+import org.apache.flink.api.common.functions.AggregateFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.linalg.BLAS;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+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.streaming.api.datastream.DataStream;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.IntStream;
+
+/**
+ * An Estimator which implements the VarianceThresholdSelector algorithm. The algorithm removes all
+ * low-variance features. Features with a variance not greater than the threshold will be removed.
+ * The default is to keep all features with non-zero variance, i.e. remove the features that have
+ * the same value in all samples.
+ */
+public class VarianceThresholdSelector
+ implements Estimator<VarianceThresholdSelector, VarianceThresholdSelectorModel>,
+ VarianceThresholdSelectorParams<VarianceThresholdSelector> {
+
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public VarianceThresholdSelector() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public VarianceThresholdSelectorModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ final String featuresCol = getFeaturesCol();
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<DenseVector> inputData =
+ tEnv.toDataStream(inputs[0])
+ .map(
+ (MapFunction<Row, DenseVector>)
+ value -> ((Vector) value.getField(featuresCol)).toDense());
+
+ DataStream<VarianceThresholdSelectorModelData> modelData =
+ DataStreamUtils.aggregate(
+ inputData, new VarianceThresholdSelectorAggregator(getVarianceThreshold()));
+
+ VarianceThresholdSelectorModel model =
+ new VarianceThresholdSelectorModel().setModelData(tEnv.fromDataStream(modelData));
+ ReadWriteUtils.updateExistingParams(model, getParamMap());
+ return model;
+ }
+
+ /**
+ * A stream operator to compute the variance from feature column of the input bounded data
+ * stream.
+ */
+ private static class VarianceThresholdSelectorAggregator
+ implements AggregateFunction<
+ DenseVector,
+ Tuple3<Long, DenseVector, DenseVector>,
+ VarianceThresholdSelectorModelData> {
+
+ private final double varianceThreshold;
+
+ public VarianceThresholdSelectorAggregator(double varianceThreshold) {
+ this.varianceThreshold = varianceThreshold;
+ }
+
+ @Override
+ public Tuple3<Long, DenseVector, DenseVector> createAccumulator() {
+ return Tuple3.of(0L, new DenseVector(new double[0]), new DenseVector(new double[0]));
+ }
+
+ @Override
+ public Tuple3<Long, DenseVector, DenseVector> add(
+ DenseVector vector, Tuple3<Long, DenseVector, DenseVector> numAndSums) {
+ if (numAndSums.f0 == 0) {
+ numAndSums.f1 = new DenseVector(vector.size());
+ numAndSums.f2 = new DenseVector(vector.size());
+ }
+ numAndSums.f0 += 1L;
+ BLAS.axpy(1.0, vector, numAndSums.f1);
+ for (int i = 0; i < vector.size(); i++) {
+ numAndSums.f2.values[i] += vector.values[i] * vector.values[i];
+ }
+ return numAndSums;
+ }
+
+ @Override
+ public VarianceThresholdSelectorModelData getResult(
+ Tuple3<Long, DenseVector, DenseVector> numAndSums) {
+ long numRows = numAndSums.f0;
+ DenseVector sumVector = numAndSums.f1;
+ DenseVector squareSumVector = numAndSums.f2;
+ Preconditions.checkState(numRows > 0, "The training set is empty.");
+
+ int[] indices =
+ IntStream.range(0, sumVector.size())
+ .filter(
+ i ->
+ squareSumVector.values[i] / numRows
+ - (sumVector.values[i] / numRows)
+ * (sumVector.values[i]
+ / numRows)
+ > varianceThreshold)
+ .toArray();
+
+ return new VarianceThresholdSelectorModelData(sumVector.size(), indices);
+ }
+
+ @Override
+ public Tuple3<Long, DenseVector, DenseVector> merge(
+ Tuple3<Long, DenseVector, DenseVector> numAndSums1,
+ Tuple3<Long, DenseVector, DenseVector> acc) {
+ if (numAndSums1.f0 == 0) {
+ return acc;
+ }
+
+ if (acc.f0 == 0) {
+ return numAndSums1;
+ }
+ acc.f0 += numAndSums1.f0;
+ BLAS.axpy(1, numAndSums1.f1, acc.f1);
+ BLAS.axpy(1, numAndSums1.f2, acc.f2);
+ return acc;
+ }
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ }
+
+ public static VarianceThresholdSelector load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ return ReadWriteUtils.loadStageParam(path);
+ }
+}
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelectorModel.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelectorModel.java
new file mode 100644
index 0000000..62c4909
--- /dev/null
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelectorModel.java
@@ -0,0 +1,170 @@
+/*
+ * 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.variancethresholdselector;
+
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.ml.api.Model;
+import org.apache.flink.ml.common.broadcast.BroadcastUtils;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo;
+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.streaming.api.datastream.DataStream;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.commons.lang3.ArrayUtils;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A Model which removes low-variance data using the model data computed by {@link
+ * VarianceThresholdSelector}.
+ */
+public class VarianceThresholdSelectorModel
+ implements Model<VarianceThresholdSelectorModel>,
+ VarianceThresholdSelectorModelParams<VarianceThresholdSelectorModel> {
+
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+ private Table modelDataTable;
+
+ public VarianceThresholdSelectorModel() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public VarianceThresholdSelectorModel setModelData(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ modelDataTable = inputs[0];
+ return this;
+ }
+
+ @Override
+ public Table[] getModelData() {
+ return new Table[] {modelDataTable};
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+
+ @Override
+ public void save(String path) throws IOException {
+ ReadWriteUtils.saveMetadata(this, path);
+ ReadWriteUtils.saveModelData(
+ VarianceThresholdSelectorModelData.getModelDataStream(modelDataTable),
+ path,
+ new VarianceThresholdSelectorModelData.ModelDataEncoder());
+ }
+
+ public static VarianceThresholdSelectorModel load(StreamTableEnvironment tEnv, String path)
+ throws IOException {
+ VarianceThresholdSelectorModel model = ReadWriteUtils.loadStageParam(path);
+ Table modelDataTable =
+ ReadWriteUtils.loadModelData(
+ tEnv, path, new VarianceThresholdSelectorModelData.ModelDataDecoder());
+ return model.setModelData(modelDataTable);
+ }
+
+ @Override
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Row> data = tEnv.toDataStream(inputs[0]);
+ DataStream<VarianceThresholdSelectorModelData> varianceThresholdSelectorModel =
+ VarianceThresholdSelectorModelData.getModelDataStream(modelDataTable);
+
+ final String broadcastModelKey = "broadcastModelKey";
+ RowTypeInfo inputTypeInfo = TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(
+ inputTypeInfo.getFieldTypes(), DenseVectorTypeInfo.INSTANCE),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getOutputCol()));
+
+ DataStream<Row> output =
+ BroadcastUtils.withBroadcastStream(
+ Collections.singletonList(data),
+ Collections.singletonMap(broadcastModelKey, varianceThresholdSelectorModel),
+ inputList -> {
+ DataStream input = inputList.get(0);
+ return input.map(
+ new PredictOutputFunction(getFeaturesCol(), broadcastModelKey),
+ outputTypeInfo);
+ });
+
+ return new Table[] {tEnv.fromDataStream(output)};
+ }
+
+ /** This operator loads model data and predicts result. */
+ private static class PredictOutputFunction extends RichMapFunction<Row, Row> {
+
+ private final String featureCol;
+ private final String broadcastKey;
+ private int expectedNumOfFeatures;
+ private int[] indices = null;
+
+ public PredictOutputFunction(String featureCol, String broadcastKey) {
+ this.featureCol = featureCol;
+ this.broadcastKey = broadcastKey;
+ }
+
+ @Override
+ public Row map(Row row) {
+ if (indices == null) {
+ VarianceThresholdSelectorModelData varianceThresholdSelectorModelData =
+ (VarianceThresholdSelectorModelData)
+ getRuntimeContext().getBroadcastVariable(broadcastKey).get(0);
+ expectedNumOfFeatures = varianceThresholdSelectorModelData.numOfFeatures;
+ indices = varianceThresholdSelectorModelData.indices;
+ }
+
+ DenseVector inputVec = ((Vector) row.getField(featureCol)).toDense();
+ Preconditions.checkArgument(
+ inputVec.size() == expectedNumOfFeatures,
+ "%s has %s features, but VarianceThresholdSelector is expecting %s features as input.",
+ featureCol,
+ inputVec.size(),
+ expectedNumOfFeatures);
+ if (indices.length == 0) {
+ return Row.join(row, Row.of(Vectors.dense()));
+ } else {
+ DenseVector outputVec = new DenseVector(indices.length);
+ for (int i = 0; i < indices.length; i++) {
+ outputVec.values[i] = inputVec.get(indices[i]);
+ }
+ return Row.join(row, Row.of(outputVec));
+ }
+ }
+ }
+}
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelectorModelData.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelectorModelData.java
new file mode 100644
index 0000000..897fa20
--- /dev/null
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelectorModelData.java
@@ -0,0 +1,119 @@
+/*
+ * 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.variancethresholdselector;
+
+import org.apache.flink.api.common.serialization.Encoder;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.array.IntPrimitiveArraySerializer;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.file.src.reader.SimpleStreamFormat;
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * Model data of {@link VarianceThresholdSelectorModel}.
+ *
+ * <p>This class also provides methods to convert model data from Table to a data stream, and
+ * classes to save/load model data.
+ */
+public class VarianceThresholdSelectorModelData {
+
+ public int numOfFeatures;
+ public int[] indices;
+
+ public VarianceThresholdSelectorModelData() {}
+
+ public VarianceThresholdSelectorModelData(int numOfFeatures, int[] indices) {
+ this.numOfFeatures = numOfFeatures;
+ this.indices = indices;
+ }
+
+ /**
+ * Converts the table model to a data stream.
+ *
+ * @param modelDataTable The table model data.
+ * @return The data stream model data.
+ */
+ public static DataStream<VarianceThresholdSelectorModelData> getModelDataStream(
+ Table modelDataTable) {
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) modelDataTable).getTableEnvironment();
+ return tEnv.toDataStream(modelDataTable)
+ .map(
+ x ->
+ new VarianceThresholdSelectorModelData(
+ (int) x.getField(0), (int[]) x.getField(1)));
+ }
+
+ /** Encoder for {@link VarianceThresholdSelectorModelData}. */
+ public static class ModelDataEncoder implements Encoder<VarianceThresholdSelectorModelData> {
+ @Override
+ public void encode(VarianceThresholdSelectorModelData modelData, OutputStream outputStream)
+ throws IOException {
+ DataOutputView dataOutputView = new DataOutputViewStreamWrapper(outputStream);
+ IntSerializer.INSTANCE.serialize(modelData.numOfFeatures, dataOutputView);
+ IntPrimitiveArraySerializer.INSTANCE.serialize(modelData.indices, dataOutputView);
+ }
+ }
+
+ /** Decoder for {@link VarianceThresholdSelectorModelData}. */
+ public static class ModelDataDecoder
+ extends SimpleStreamFormat<VarianceThresholdSelectorModelData> {
+ @Override
+ public Reader<VarianceThresholdSelectorModelData> createReader(
+ Configuration config, FSDataInputStream stream) {
+ return new Reader<VarianceThresholdSelectorModelData>() {
+
+ @Override
+ public VarianceThresholdSelectorModelData read() throws IOException {
+ DataInputView source = new DataInputViewStreamWrapper(stream);
+ try {
+ int numFeatures = IntSerializer.INSTANCE.deserialize(source);
+ int[] indices = IntPrimitiveArraySerializer.INSTANCE.deserialize(source);
+ return new VarianceThresholdSelectorModelData(numFeatures, indices);
+ } catch (EOFException e) {
+ return null;
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ stream.close();
+ }
+ };
+ }
+
+ @Override
+ public TypeInformation<VarianceThresholdSelectorModelData> getProducedType() {
+ return TypeInformation.of(VarianceThresholdSelectorModelData.class);
+ }
+ }
+}
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelectorModelParams.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelectorModelParams.java
new file mode 100644
index 0000000..8892257
--- /dev/null
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelectorModelParams.java
@@ -0,0 +1,30 @@
+/*
+ * 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.variancethresholdselector;
+
+import org.apache.flink.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasOutputCol;
+
+/**
+ * Params for {@link VarianceThresholdSelectorModel}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface VarianceThresholdSelectorModelParams<T>
+ extends HasFeaturesCol<T>, HasOutputCol<T> {}
diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelectorParams.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelectorParams.java
new file mode 100644
index 0000000..f8e0692
--- /dev/null
+++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/feature/variancethresholdselector/VarianceThresholdSelectorParams.java
@@ -0,0 +1,47 @@
+/*
+ * 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.variancethresholdselector;
+
+import org.apache.flink.ml.param.DoubleParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+
+/**
+ * Params of VarianceThresholdSelectorModel.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface VarianceThresholdSelectorParams<T>
+ extends VarianceThresholdSelectorModelParams<T> {
+
+ Param<Double> VARIANCE_THRESHOLD =
+ new DoubleParam(
+ "varianceThreshold",
+ "Features with a variance not greater than this threshold will be removed.",
+ 0.0,
+ ParamValidators.gtEq(0.0));
+
+ default double getVarianceThreshold() {
+ return get(VARIANCE_THRESHOLD);
+ }
+
+ default T setVarianceThreshold(double value) {
+ return set(VARIANCE_THRESHOLD, value);
+ }
+}
diff --git a/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/VarianceThresholdSelectorTest.java b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/VarianceThresholdSelectorTest.java
new file mode 100644
index 0000000..e4e841c
--- /dev/null
+++ b/flink-ml-lib/src/test/java/org/apache/flink/ml/feature/VarianceThresholdSelectorTest.java
@@ -0,0 +1,279 @@
+/*
+ * 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.functions.MapFunction;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.feature.variancethresholdselector.VarianceThresholdSelector;
+import org.apache.flink.ml.feature.variancethresholdselector.VarianceThresholdSelectorModel;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.SparseVector;
+import org.apache.flink.ml.linalg.Vectors;
+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.Expressions;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.internal.TableImpl;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/** Tests {@link VarianceThresholdSelector} and {@link VarianceThresholdSelectorModel}. */
+public class VarianceThresholdSelectorTest extends AbstractTestBase {
+
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+ private Table trainDataTable;
+ private Table predictDataTable;
+
+ private static final double EPS = 1.0e-5;
+ private static final List<Row> TRAIN_DATA =
+ new ArrayList<>(
+ Arrays.asList(
+ Row.of(1, Vectors.dense(5.0, 7.0, 0.0, 7.0, 6.0, 0.0)),
+ Row.of(2, Vectors.dense(0.0, 9.0, 6.0, 0.0, 5.0, 9.0)),
+ Row.of(3, Vectors.dense(0.0, 9.0, 3.0, 0.0, 5.0, 5.0)),
+ Row.of(4, Vectors.dense(1.0, 9.0, 8.0, 5.0, 7.0, 4.0)),
+ Row.of(5, Vectors.dense(9.0, 8.0, 6.0, 5.0, 4.0, 4.0)),
+ Row.of(6, Vectors.dense(6.0, 9.0, 7.0, 0.0, 2.0, 0.0))));
+
+ private static final List<Row> PREDICT_DATA =
+ new ArrayList<>(
+ Arrays.asList(
+ Row.of(Vectors.dense(1.0, 2.0, 3.0, 4.0, 5.0, 6.0)),
+ Row.of(Vectors.dense(0.1, 0.2, 0.3, 0.4, 0.5, 0.6))));
+
+ private static final List<DenseVector> EXPECTED_OUTPUT =
+ new ArrayList<>(
+ Arrays.asList(Vectors.dense(1.0, 4.0, 6.0), Vectors.dense(0.1, 0.4, 0.6)));
+
+ @Before
+ public void before() {
+ Configuration config = new Configuration();
+ config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
+ env = StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.setParallelism(4);
+ env.enableCheckpointing(100);
+ env.setRestartStrategy(RestartStrategies.noRestart());
+ tEnv = StreamTableEnvironment.create(env);
+
+ trainDataTable = tEnv.fromDataStream(env.fromCollection(TRAIN_DATA)).as("id", "features");
+ predictDataTable = tEnv.fromDataStream(env.fromCollection(PREDICT_DATA)).as("features");
+ }
+
+ private static void verifyPredictionResult(
+ Table output, String outputCol, List<DenseVector> expected) throws Exception {
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) output).getTableEnvironment();
+ DataStream<DenseVector> stream =
+ tEnv.toDataStream(output)
+ .map(
+ (MapFunction<Row, DenseVector>)
+ row -> (DenseVector) row.getField(outputCol));
+ List<DenseVector> result = IteratorUtils.toList(stream.executeAndCollect());
+ compareResultCollections(expected, result, TestUtils::compare);
+ }
+
+ @Test
+ public void testParam() {
+ VarianceThresholdSelector varianceThresholdSelector = new VarianceThresholdSelector();
+ assertEquals("features", varianceThresholdSelector.getFeaturesCol());
+ assertEquals("output", varianceThresholdSelector.getOutputCol());
+ assertEquals(0.0, varianceThresholdSelector.getVarianceThreshold(), EPS);
+
+ varianceThresholdSelector
+ .setFeaturesCol("test_feature")
+ .setOutputCol("test_output")
+ .setVarianceThreshold(0.5);
+ assertEquals("test_feature", varianceThresholdSelector.getFeaturesCol());
+ assertEquals(0.5, varianceThresholdSelector.getVarianceThreshold(), EPS);
+ assertEquals("test_output", varianceThresholdSelector.getOutputCol());
+ }
+
+ @Test
+ public void testOutputSchema() {
+ VarianceThresholdSelector varianceThresholdSelector =
+ new VarianceThresholdSelector().setOutputCol("output").setVarianceThreshold(0.5);
+ VarianceThresholdSelectorModel model = varianceThresholdSelector.fit(trainDataTable);
+ Table output = model.transform(trainDataTable)[0];
+ assertEquals(
+ Arrays.asList("id", "features", "output"),
+ output.getResolvedSchema().getColumnNames());
+ }
+
+ @Test
+ public void testFitAndPredict() throws Exception {
+ VarianceThresholdSelector varianceThresholdSelector =
+ new VarianceThresholdSelector().setVarianceThreshold(8.0);
+ VarianceThresholdSelectorModel model = varianceThresholdSelector.fit(trainDataTable);
+ Table predictTableOutput = model.transform(predictDataTable)[0];
+ verifyPredictionResult(
+ predictTableOutput, varianceThresholdSelector.getOutputCol(), EXPECTED_OUTPUT);
+ }
+
+ @Test
+ public void testNonSelectedFeatures() throws Exception {
+ VarianceThresholdSelector varianceThresholdSelector =
+ new VarianceThresholdSelector().setVarianceThreshold(100.0);
+ VarianceThresholdSelectorModel model = varianceThresholdSelector.fit(trainDataTable);
+ Table predictTableOutput = model.transform(predictDataTable)[0];
+ verifyPredictionResult(
+ predictTableOutput,
+ varianceThresholdSelector.getOutputCol(),
+ new ArrayList<>(Arrays.asList(Vectors.dense(), Vectors.dense())));
+ }
+
+ @Test
+ public void testInputTypeConversion() throws Exception {
+ trainDataTable =
+ TestUtils.convertDataTypesToSparseInt(
+ tEnv, trainDataTable.select(Expressions.$("features")));
+ predictDataTable = TestUtils.convertDataTypesToSparseInt(tEnv, predictDataTable);
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class}, TestUtils.getColumnDataTypes(trainDataTable));
+ assertArrayEquals(
+ new Class<?>[] {SparseVector.class},
+ TestUtils.getColumnDataTypes(predictDataTable));
+
+ VarianceThresholdSelector varianceThresholdSelector =
+ new VarianceThresholdSelector()
+ .setFeaturesCol("features")
+ .setVarianceThreshold(8.0);
+ VarianceThresholdSelectorModel model = varianceThresholdSelector.fit(trainDataTable);
+ Table output = model.transform(predictDataTable)[0];
+ verifyPredictionResult(output, varianceThresholdSelector.getOutputCol(), EXPECTED_OUTPUT);
+ }
+
+ @Test
+ public void testSaveLoadAndPredict() throws Exception {
+ VarianceThresholdSelector varianceThresholdSelector =
+ new VarianceThresholdSelector()
+ .setFeaturesCol("features")
+ .setVarianceThreshold(8.0);
+ VarianceThresholdSelector loadedVarianceThresholdSelector =
+ TestUtils.saveAndReload(
+ tEnv, varianceThresholdSelector, tempFolder.newFolder().getAbsolutePath());
+ VarianceThresholdSelectorModel model = loadedVarianceThresholdSelector.fit(trainDataTable);
+ VarianceThresholdSelectorModel loadedModel =
+ TestUtils.saveAndReload(tEnv, model, tempFolder.newFolder().getAbsolutePath());
+ assertEquals(
+ Arrays.asList("numOfFeatures", "indices"),
+ model.getModelData()[0].getResolvedSchema().getColumnNames());
+ Table output = loadedModel.transform(predictDataTable)[0];
+ verifyPredictionResult(output, varianceThresholdSelector.getOutputCol(), EXPECTED_OUTPUT);
+ }
+
+ @Test
+ public void testFitOnEmptyData() {
+ Table emptyTable =
+ tEnv.fromDataStream(env.fromCollection(TRAIN_DATA).filter(x -> x.getArity() == 0))
+ .as("id", "features");
+ VarianceThresholdSelector varianceThresholdSelector = new VarianceThresholdSelector();
+ VarianceThresholdSelectorModel model = varianceThresholdSelector.fit(emptyTable);
+ Table modelDataTable = model.getModelData()[0];
+
+ try {
+ modelDataTable.execute().print();
+ fail();
+ } catch (Throwable e) {
+ assertEquals("The training set is empty.", ExceptionUtils.getRootCause(e).getMessage());
+ }
+ }
+
+ @Test
+ public void testIncompatibleNumOfFeatures() {
+ VarianceThresholdSelector varianceThresholdSelector =
+ new VarianceThresholdSelector().setVarianceThreshold(8.0);
+ VarianceThresholdSelectorModel model = varianceThresholdSelector.fit(trainDataTable);
+
+ List<Row> predictData =
+ new ArrayList<>(
+ Arrays.asList(
+ Row.of(Vectors.dense(1.0, 2.0, 3.0, 4.0)),
+ Row.of(Vectors.dense(0.1, 0.2, 0.3, 0.4))));
+ Table predictTable = tEnv.fromDataStream(env.fromCollection(predictData)).as("features");
+ Table output = model.transform(predictTable)[0];
+ try {
+ output.execute().print();
+ fail();
+ } catch (Throwable e) {
+ assertTrue(
+ ExceptionUtils.getRootCause(e)
+ .getMessage()
+ .contains("but VarianceThresholdSelector is expecting"));
+ }
+ }
+
+ @Test
+ public void testGetModelData() throws Exception {
+ VarianceThresholdSelector varianceThresholdSelector =
+ new VarianceThresholdSelector()
+ .setFeaturesCol("features")
+ .setVarianceThreshold(8.0);
+ VarianceThresholdSelectorModel model = varianceThresholdSelector.fit(trainDataTable);
+ Table modelData = model.getModelData()[0];
+ assertEquals(
+ Arrays.asList("numOfFeatures", "indices"),
+ modelData.getResolvedSchema().getColumnNames());
+ DataStream<Row> output = tEnv.toDataStream(modelData);
+ List<Row> modelRows = IteratorUtils.toList(output.executeAndCollect());
+ long numOfFeatures = (int) modelRows.get(0).getField(0);
+ int[] indices = (int[]) modelRows.get(0).getField(1);
+ assertEquals(6, numOfFeatures);
+ int[] expectedIndices = {0, 3, 5};
+ for (int i = 0; i < indices.length; i++) {
+ assertEquals(expectedIndices[i], indices[i]);
+ }
+ }
+
+ @Test
+ public void testSetModelData() throws Exception {
+ VarianceThresholdSelector varianceThresholdSelector =
+ new VarianceThresholdSelector()
+ .setFeaturesCol("features")
+ .setVarianceThreshold(8.0);
+ VarianceThresholdSelectorModel modelA = varianceThresholdSelector.fit(trainDataTable);
+
+ Table modelData = modelA.getModelData()[0];
+ VarianceThresholdSelectorModel modelB =
+ new VarianceThresholdSelectorModel().setModelData(modelData);
+ Table output = modelB.transform(predictDataTable)[0];
+ verifyPredictionResult(output, varianceThresholdSelector.getOutputCol(), EXPECTED_OUTPUT);
+ }
+}
diff --git a/flink-ml-python/pyflink/examples/ml/feature/variancethresholdselector_example.py b/flink-ml-python/pyflink/examples/ml/feature/variancethresholdselector_example.py
new file mode 100644
index 0000000..3cf7ad6
--- /dev/null
+++ b/flink-ml-python/pyflink/examples/ml/feature/variancethresholdselector_example.py
@@ -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.
+################################################################################
+
+# Simple program that trains a VarianceThresholdSelector model and uses it for feature
+# selection.
+
+from pyflink.common import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.ml.core.linalg import Vectors, DenseVectorTypeInfo
+from pyflink.ml.lib.feature.variancethresholdselector import VarianceThresholdSelector
+from pyflink.table import StreamTableEnvironment
+
+# create a new StreamExecutionEnvironment
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# create a StreamTableEnvironment
+t_env = StreamTableEnvironment.create(env)
+
+# generate input training and prediction data
+train_data = t_env.from_data_stream(
+ env.from_collection([
+ (1, Vectors.dense(5.0, 7.0, 0.0, 7.0, 6.0, 0.0),),
+ (2, Vectors.dense(0.0, 9.0, 6.0, 0.0, 5.0, 9.0),),
+ (3, Vectors.dense(0.0, 9.0, 3.0, 0.0, 5.0, 5.0),),
+ (4, Vectors.dense(1.0, 9.0, 8.0, 5.0, 7.0, 4.0),),
+ (5, Vectors.dense(9.0, 8.0, 6.0, 5.0, 4.0, 4.0),),
+ (6, Vectors.dense(6.0, 9.0, 7.0, 0.0, 2.0, 0.0),),
+ ],
+ type_info=Types.ROW_NAMED(
+ ['id', 'features'],
+ [Types.INT(), DenseVectorTypeInfo()])
+ ))
+
+# create a VarianceThresholdSelector object and initialize its parameters
+threshold = 8.0
+variance_thread_selector = VarianceThresholdSelector()\
+ .set_features_col("features")\
+ .set_variance_threshold(threshold)
+
+# train the VarianceThresholdSelector model
+model = variance_thread_selector.fit(train_data)
+
+# use the VarianceThresholdSelector model for predictions
+output = model.transform(train_data)[0]
+
+# extract and display the results
+print("Variance Threshold: " + str(threshold))
+field_names = output.get_schema().get_field_names()
+for result in t_env.to_data_stream(output).execute_and_collect():
+ input_value = result[field_names.index(variance_thread_selector.get_features_col())]
+ output_value = result[field_names.index(variance_thread_selector.get_output_col())]
+ print('Original Features: ' + str(input_value) + ' \tSelected Features: ' + str(output_value))
diff --git a/flink-ml-python/pyflink/ml/lib/feature/tests/test_variancethresholdselector.py b/flink-ml-python/pyflink/ml/lib/feature/tests/test_variancethresholdselector.py
new file mode 100644
index 0000000..854b9bd
--- /dev/null
+++ b/flink-ml-python/pyflink/ml/lib/feature/tests/test_variancethresholdselector.py
@@ -0,0 +1,138 @@
+################################################################################
+# 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 typing import List
+
+from pyflink.common import Types
+from pyflink.table import Table
+
+from pyflink.ml.core.linalg import Vectors, DenseVectorTypeInfo, DenseVector
+from pyflink.ml.lib.feature.variancethresholdselector import VarianceThresholdSelector
+from pyflink.ml.tests.test_utils import PyFlinkMLTestCase
+
+
+class VarianceThresholdSelectorTest(PyFlinkMLTestCase):
+ def setUp(self):
+ super(VarianceThresholdSelectorTest, self).setUp()
+ self.train_table = self.t_env.from_data_stream(
+ self.env.from_collection([
+ (1, Vectors.dense(5.0, 7.0, 0.0, 7.0, 6.0, 0.0),),
+ (2, Vectors.dense(0.0, 9.0, 6.0, 0.0, 5.0, 9.0),),
+ (3, Vectors.dense(0.0, 9.0, 3.0, 0.0, 5.0, 5.0),),
+ (4, Vectors.dense(1.0, 9.0, 8.0, 5.0, 7.0, 4.0),),
+ (5, Vectors.dense(9.0, 8.0, 6.0, 5.0, 4.0, 4.0),),
+ (6, Vectors.dense(6.0, 9.0, 7.0, 0.0, 2.0, 0.0),),
+ ],
+ type_info=Types.ROW_NAMED(
+ ['id', 'features'],
+ [Types.INT(), DenseVectorTypeInfo()])
+ ))
+
+ self.predict_table = self.t_env.from_data_stream(
+ self.env.from_collection([
+ (Vectors.dense(1.0, 2.0, 3.0, 4.0, 5.0, 6.0),),
+ (Vectors.dense(0.1, 0.2, 0.3, 0.4, 0.5, 0.6),),
+ ],
+ type_info=Types.ROW_NAMED(
+ ['features'],
+ [DenseVectorTypeInfo()])
+ ))
+ self.expected_output = [
+ Vectors.dense(0.1, 0.4, 0.6),
+ Vectors.dense(1.0, 4.0, 6.0)]
+
+ def test_param(self):
+ variance_threshold_selector = VarianceThresholdSelector()
+ self.assertEqual("features", variance_threshold_selector.features_col)
+ self.assertEqual("output", variance_threshold_selector.output_col)
+ self.assertEqual(0.0, variance_threshold_selector.variance_threshold)
+
+ variance_threshold_selector.set_output_col("test_output")\
+ .set_variance_threshold(8.0)
+ self.assertEqual("features", variance_threshold_selector.features_col)
+ self.assertEqual("test_output", variance_threshold_selector.output_col)
+ self.assertEqual(8.0, variance_threshold_selector.variance_threshold)
+
+ def test_output_schema(self):
+ variance_threshold_selector = VarianceThresholdSelector() \
+ .set_output_col('test_output') \
+ .set_variance_threshold(8.0)
+
+ model = variance_threshold_selector.fit(self.train_table)
+ output = model.transform(self.predict_table.alias('test_input'))[0]
+ self.assertEqual(
+ ['test_input', 'test_output'],
+ output.get_schema().get_field_names())
+
+ def test_fit_and_predict(self):
+ variance_threshold_selector = VarianceThresholdSelector() \
+ .set_variance_threshold(8.0)
+ model = variance_threshold_selector.fit(self.train_table)
+ output = model.transform(self.predict_table)[0]
+ self.verify_output_result(
+ output,
+ variance_threshold_selector.get_output_col(),
+ output.get_schema().get_field_names(),
+ self.expected_output)
+
+ def test_incompatible_num_of_features(self):
+ variance_threshold_selector = VarianceThresholdSelector() \
+ .set_variance_threshold(8.0)
+ model = variance_threshold_selector.fit(self.train_table)
+ predict_table = self.t_env.from_data_stream(
+ self.env.from_collection([
+ (Vectors.dense(1.0, 2.0, 3.0, 4.0, 5.0),),
+ (Vectors.dense(0.1, 0.2, 0.3, 0.4, 0.5),),
+ ],
+ type_info=Types.ROW_NAMED(
+ ['features'],
+ [DenseVectorTypeInfo()])
+ ))
+ with self.assertRaisesRegex(Exception, 'but VarianceThresholdSelector is expecting'):
+ output = model.transform(predict_table)[0]
+ self.verify_output_result(
+ output,
+ variance_threshold_selector.get_output_col(),
+ output.get_schema().get_field_names(),
+ self.expected_output)
+
+ def test_save_load_predict(self):
+ variance_threshold_selector = VarianceThresholdSelector() \
+ .set_variance_threshold(8.0)
+ reloaded_variance_threshold_selector = self.save_and_reload(variance_threshold_selector)
+ model = reloaded_variance_threshold_selector.fit(self.train_table)
+ reloaded_model = self.save_and_reload(model)
+ output = reloaded_model.transform(self.predict_table)[0]
+ self.verify_output_result(
+ output,
+ variance_threshold_selector.get_output_col(),
+ output.get_schema().get_field_names(),
+ self.expected_output)
+
+ def verify_output_result(
+ self, output: Table,
+ output_col: str,
+ field_names: List[str],
+ expected_result: List[DenseVector]):
+ collected_results = [result for result in
+ self.t_env.to_data_stream(output).execute_and_collect()]
+ results = []
+ for item in collected_results:
+ item.set_field_names(field_names)
+ results.append(item[output_col])
+ results.sort(key=lambda x: x[0])
+ self.assertEqual(expected_result, results)
diff --git a/flink-ml-python/pyflink/ml/lib/feature/variancethresholdselector.py b/flink-ml-python/pyflink/ml/lib/feature/variancethresholdselector.py
new file mode 100644
index 0000000..23b2c6e
--- /dev/null
+++ b/flink-ml-python/pyflink/ml/lib/feature/variancethresholdselector.py
@@ -0,0 +1,95 @@
+################################################################################
+# 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 pyflink.ml.core.param import Param, FloatParam, ParamValidators
+from pyflink.ml.core.wrapper import JavaWithParams
+from pyflink.ml.lib.feature.common import JavaFeatureModel, JavaFeatureEstimator
+from pyflink.ml.lib.param import HasFeaturesCol, HasOutputCol
+
+
+class _VarianceThresholdSelectorParams(
+ JavaWithParams,
+ HasFeaturesCol,
+ HasOutputCol
+):
+ """
+ Params for :class:`VarianceThresholdSelector`.
+ """
+
+ VARIANCE_THRESHOLD: Param[float] = FloatParam(
+ "variance_threshold",
+ "Features with a variance not greater than this threshold will be removed.",
+ 0.0,
+ ParamValidators.gt_eq(0.0))
+
+ def __init__(self, java_params):
+ super(_VarianceThresholdSelectorParams, self).__init__(java_params)
+
+ def set_variance_threshold(self, value: float):
+ return typing.cast(_VarianceThresholdSelectorParams,
+ self.set(self.VARIANCE_THRESHOLD, value))
+
+ def get_variance_threshold(self):
+ return self.get(self.VARIANCE_THRESHOLD)
+
+ @property
+ def variance_threshold(self):
+ return self.get_variance_threshold()
+
+
+class VarianceThresholdSelectorModel(JavaFeatureModel, _VarianceThresholdSelectorParams):
+ """
+ A Model which transforms data using the model data
+ computed by :class:`VarianceThresholdSelector`.
+ """
+
+ def __init__(self, java_model=None):
+ super(VarianceThresholdSelectorModel, self).__init__(java_model)
+
+ @classmethod
+ def _java_model_package_name(cls) -> str:
+ return "variancethresholdselector"
+
+ @classmethod
+ def _java_model_class_name(cls) -> str:
+ return "VarianceThresholdSelectorModel"
+
+
+class VarianceThresholdSelector(JavaFeatureEstimator, _VarianceThresholdSelectorParams):
+ """
+ An Estimator which implements the VarianceThresholdSelector algorithm. The algorithm
+ removes all low-variance features. Features with a variance not greater than the
+ threshold will be removed. The default is to keep all features with non-zero variance,
+ i.e. remove the features that have the same value in all samples.
+ """
+
+ def __init__(self):
+ super(VarianceThresholdSelector, self).__init__()
+
+ @classmethod
+ def _create_model(cls, java_model) -> VarianceThresholdSelectorModel:
+ return VarianceThresholdSelectorModel(java_model)
+
+ @classmethod
+ def _java_estimator_package_name(cls) -> str:
+ return "variancethresholdselector"
+
+ @classmethod
+ def _java_estimator_class_name(cls) -> str:
+ return "VarianceThresholdSelector"
diff --git a/flink-ml-python/pyflink/ml/tests/test_utils.py b/flink-ml-python/pyflink/ml/tests/test_utils.py
index 7cea8e7..04104c4 100644
--- a/flink-ml-python/pyflink/ml/tests/test_utils.py
+++ b/flink-ml-python/pyflink/ml/tests/test_utils.py
@@ -20,6 +20,7 @@ import os
import shutil
import tempfile
import unittest
+import uuid
from pyflink.common import RestartStrategies, Configuration
from pyflink.datastream import StreamExecutionEnvironment
@@ -70,3 +71,16 @@ class PyFlinkMLTestCase(unittest.TestCase):
FLINK_ML_LIB_SOURCE_PATH, "target", "flink-ml-lib-*-tests.jar"))[0]
self.env.add_classpaths("file://{0}".format(ml_test_jar))
+
+ def save_and_reload(self, stage):
+ path = os.path.join(self.temp_dir, 'test_save_and_reload', str(uuid.uuid1()))
+ stage.save(path)
+ try:
+ self.env.execute('save_stage')
+ except Exception as e:
+ if "No operators defined in streaming topology." in str(e):
+ pass
+ else:
+ raise e
+ load_func = getattr(stage, 'load')
+ return load_func(self.t_env, path)