yunfengzhou-hub commented on code in PR #172:
URL: https://github.com/apache/flink-ml/pull/172#discussion_r1017456006
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/common/param/HasRelativeError.java:
##########
@@ -36,7 +36,7 @@ default double getRelativeError() {
return get(RELATIVE_ERROR);
}
- default T setFeaturesCol(double value) {
+ default T setRelativeError(double value) {
Review Comment:
Let's add tests for `setRelativeError` and `getRelativeError` in
`ImputerTest.testParams`.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/feature/robustscaler/RobustScaler.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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.robustscaler;
+
+import org.apache.flink.api.common.functions.AggregateFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.util.QuantileSummary;
+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.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Scale features using statistics that are robust to outliers.
Review Comment:
Let's make the JavaDoc begin with a noun or definition. For example,
- An algorithm that scales xxx
- RobustScaler is an algorithm that scales xxx
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/feature/robustscaler/RobustScaler.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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.robustscaler;
+
+import org.apache.flink.api.common.functions.AggregateFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.util.QuantileSummary;
+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.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Scale features using statistics that are robust to outliers.
+ *
+ * <p>This Scaler removes the median and scales the data according to the
quantile range (defaults
+ * to IQR: Interquartile Range). The IQR is the range between the 1st quartile
(25th quantile) and
+ * the 3rd quartile (75th quantile) but can be configured.
+ *
+ * <p>Centering and scaling happen independently on each feature by computing
the relevant
+ * statistics on the samples in the training set. Median and quantile range
are then stored to be
+ * used on later data using the transform method.
+ *
+ * <p>Standardization of a dataset is a common requirement for many machine
learning estimators.
+ * Typically this is done by removing the mean and scaling to unit variance.
However, outliers can
+ * often influence the sample mean / variance in a negative way. In such
cases, the median and the
+ * interquartile range often give better results.
+ */
+public class RobustScaler
+ implements Estimator<RobustScaler, RobustScalerModel>,
RobustScalerParams<RobustScaler> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public RobustScaler() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public RobustScalerModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl)
inputs[0]).getTableEnvironment();
+ DataStream<DenseVector> inputData =
+ tEnv.toDataStream(inputs[0])
+ .map(
+ (MapFunction<Row, DenseVector>)
+ value ->
+ ((Vector)
value.getField(getInputCol())).toDense());
+ DataStream<RobustScalerModelData> modelData =
+ DataStreamUtils.aggregate(
+ inputData,
+ new QuantileAggregator(getRelativeError(), getLower(),
getUpper()));
+ RobustScalerModel model =
+ new
RobustScalerModel().setModelData(tEnv.fromDataStream(modelData));
+ ReadWriteUtils.updateExistingParams(model, getParamMap());
+ return model;
+ }
+
+ /**
+ * A stream operator to compute the medians and quantile ranges from
feature column of the input
Review Comment:
It might be better to change "stream operator" to "function" or "aggregate
function".
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/feature/robustscaler/RobustScalerModel.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.robustscaler;
+
+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.BLAS;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.typeinfo.VectorTypeInfo;
+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.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/** A Model which transforms data using the model data computed by {@link
RobustScaler}. */
+public class RobustScalerModel
+ implements Model<RobustScalerModel>,
RobustScalerModelParams<RobustScalerModel> {
+
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+ private Table modelDataTable;
+
+ public RobustScalerModel() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public Table[] transform(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl)
inputs[0]).getTableEnvironment();
+ DataStream<Row> inputStream = tEnv.toDataStream(inputs[0]);
+
+ RowTypeInfo inputTypeInfo =
TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema());
+ RowTypeInfo outputTypeInfo =
+ new RowTypeInfo(
+ ArrayUtils.addAll(inputTypeInfo.getFieldTypes(),
VectorTypeInfo.INSTANCE),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(),
getOutputCol()));
+ final String broadcastModelKey = "broadcastModelKey";
+ DataStream<RobustScalerModelData> modelDataStream =
+ RobustScalerModelData.getModelDataStream(modelDataTable);
+
+ DataStream<Row> output =
+ BroadcastUtils.withBroadcastStream(
+ Collections.singletonList(inputStream),
+ Collections.singletonMap(broadcastModelKey,
modelDataStream),
+ inputList -> {
+ DataStream inputData = inputList.get(0);
+ return inputData.map(
+ new PredictOutputFunction(
+ broadcastModelKey,
+ getInputCol(),
+ getWithCentering(),
+ getWithScaling()),
+ 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 broadcastModelKey;
+ private final String inputCol;
+ private final boolean withCentering;
+ private final boolean withScaling;
+
+ private DenseVector medians;
+ private DenseVector scales;
+
+ public PredictOutputFunction(
+ String broadcastModelKey,
+ String inputCol,
+ boolean withCentering,
+ boolean withScaling) {
+ this.broadcastModelKey = broadcastModelKey;
+ this.inputCol = inputCol;
+ this.withCentering = withCentering;
+ this.withScaling = withScaling;
+ }
+
+ @Override
+ public Row map(Row row) throws Exception {
+ if (medians == null) {
+ RobustScalerModelData modelData =
+ (RobustScalerModelData)
+
getRuntimeContext().getBroadcastVariable(broadcastModelKey).get(0);
+ medians = modelData.medians;
+ scales =
+ new DenseVector(
+ Arrays.stream(modelData.ranges.values)
+ .map(range -> range == 0 ? 1 : 1 /
range)
Review Comment:
Could you please explain why this PR chooses to make scale = 1 when range
== 0? Spark seems to make scale = 0 in this case.
##########
flink-ml-lib/src/test/java/org/apache/flink/ml/feature/RobustScalerTest.java:
##########
@@ -0,0 +1,295 @@
+/*
+ * 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.robustscaler.RobustScaler;
+import org.apache.flink.ml.feature.robustscaler.RobustScalerModel;
+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.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/** Tests {@link RobustScaler} and {@link RobustScalerModel}. */
+public class RobustScalerTest extends AbstractTestBase {
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+ private Table trainDataTable;
+ private Table predictDataTable;
+
+ private static final List<Row> TRAIN_DATA =
+ new ArrayList<>(
+ Arrays.asList(
+ Row.of(0, Vectors.dense(0.0, 0.0)),
+ Row.of(1, Vectors.dense(1.0, -1.0)),
+ Row.of(2, Vectors.dense(2.0, -2.0)),
+ Row.of(3, Vectors.dense(3.0, -3.0)),
+ Row.of(4, Vectors.dense(4.0, -4.0)),
+ Row.of(5, Vectors.dense(5.0, -5.0)),
+ Row.of(6, Vectors.dense(6.0, -6.0)),
+ Row.of(7, Vectors.dense(7.0, -7.0)),
+ Row.of(8, Vectors.dense(8.0, -8.0))));
+ private static final List<Row> PREDICT_DATA =
+ new ArrayList<>(
+ Arrays.asList(
+ Row.of(Vectors.dense(3.0, -3.0)),
+ Row.of(Vectors.dense(6.0, -6.0)),
+ Row.of(Vectors.dense(99.0, -99.0))));
+ private static final double EPS = 1.0e-5;
+
+ private static final List<DenseVector> EXPECTED_OUTPUT =
+ new ArrayList<>(
+ Arrays.asList(
+ Vectors.dense(0.75, -0.75),
+ Vectors.dense(1.5, -1.5),
+ Vectors.dense(24.75, -24.75)));
+
+ @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", "input");
+ predictDataTable =
tEnv.fromDataStream(env.fromCollection(PREDICT_DATA)).as("input");
+ }
+
+ 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() {
+ RobustScaler robustScaler = new RobustScaler();
+ assertEquals("input", robustScaler.getInputCol());
+ assertEquals("output", robustScaler.getOutputCol());
+ assertEquals(0.25, robustScaler.getLower(), EPS);
+ assertEquals(0.75, robustScaler.getUpper(), EPS);
+ assertEquals(0.001, robustScaler.getRelativeError(), EPS);
+ assertFalse(robustScaler.getWithCentering());
+ assertTrue(robustScaler.getWithScaling());
+
+ robustScaler
+ .setInputCol("test_input")
+ .setOutputCol("test_output")
+ .setLower(0.1)
+ .setUpper(0.9)
+ .setRelativeError(0.01)
+ .setWithCentering(false)
Review Comment:
It might be better to set it to a non-default value.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/feature/robustscaler/RobustScaler.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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.robustscaler;
+
+import org.apache.flink.api.common.functions.AggregateFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.util.QuantileSummary;
+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.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Scale features using statistics that are robust to outliers.
+ *
+ * <p>This Scaler removes the median and scales the data according to the
quantile range (defaults
+ * to IQR: Interquartile Range). The IQR is the range between the 1st quartile
(25th quantile) and
+ * the 3rd quartile (75th quantile) but can be configured.
+ *
+ * <p>Centering and scaling happen independently on each feature by computing
the relevant
+ * statistics on the samples in the training set. Median and quantile range
are then stored to be
+ * used on later data using the transform method.
+ *
+ * <p>Standardization of a dataset is a common requirement for many machine
learning estimators.
+ * Typically this is done by removing the mean and scaling to unit variance.
However, outliers can
+ * often influence the sample mean / variance in a negative way. In such
cases, the median and the
+ * interquartile range often give better results.
Review Comment:
I guess "range" should be "ranges". Let's also check if there are other
grammar errors.
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/feature/robustscaler/RobustScaler.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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.robustscaler;
+
+import org.apache.flink.api.common.functions.AggregateFunction;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.common.util.QuantileSummary;
+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.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Scale features using statistics that are robust to outliers.
+ *
+ * <p>This Scaler removes the median and scales the data according to the
quantile range (defaults
+ * to IQR: Interquartile Range). The IQR is the range between the 1st quartile
(25th quantile) and
+ * the 3rd quartile (75th quantile) but can be configured.
+ *
+ * <p>Centering and scaling happen independently on each feature by computing
the relevant
+ * statistics on the samples in the training set. Median and quantile range
are then stored to be
+ * used on later data using the transform method.
+ *
+ * <p>Standardization of a dataset is a common requirement for many machine
learning estimators.
+ * Typically this is done by removing the mean and scaling to unit variance.
However, outliers can
+ * often influence the sample mean / variance in a negative way. In such
cases, the median and the
+ * interquartile range often give better results.
+ */
+public class RobustScaler
+ implements Estimator<RobustScaler, RobustScalerModel>,
RobustScalerParams<RobustScaler> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public RobustScaler() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public RobustScalerModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl)
inputs[0]).getTableEnvironment();
+ DataStream<DenseVector> inputData =
+ tEnv.toDataStream(inputs[0])
+ .map(
+ (MapFunction<Row, DenseVector>)
+ value ->
+ ((Vector)
value.getField(getInputCol())).toDense());
Review Comment:
It might be better to create a final variable of `getInputCol()`'s value and
read the value in this map function.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
