You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/03/23 06:25:24 UTC
[GitHub] [flink-ml] yunfengzhou-hub commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
yunfengzhou-hub commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r832818995
##########
File path: flink-ml-core/src/main/java/org/apache/flink/ml/linalg/BLAS.java
##########
@@ -32,9 +32,31 @@ public static double asum(DenseVector x) {
}
/** y += a * x . */
- public static void axpy(double a, DenseVector x, DenseVector y) {
+ public static void axpy(double a, Vector x, DenseVector y) {
Preconditions.checkArgument(x.size() == y.size(), "Vector size mismatched.");
- JAVA_BLAS.daxpy(x.size(), a, x.values, 1, y.values, 1);
+ if (x instanceof SparseVector) {
+ axpy(a, (SparseVector) x, y);
+ } else {
+ axpy(a, (DenseVector) x, y);
+ }
+ }
+
+ /** Computes the hadamard product of the two vectors (y = y \hdot x). */
+ public static void hDot(Vector x, Vector y) {
Review comment:
I have not found such `hDot` method in BLAS library, nor has spark or alink created such method in their `BLAS`. Thus I'm not sure whether we should add such method.
In MinMaxScaler's PR, it needs to compute the product between each element of two vectors. Weibo and I had a detailed discussion about this and our conclusion is to directly implement the for loops in the algorithm's body, instead of using or adding BLAS operations.
##########
File path: flink-ml-core/src/main/java/org/apache/flink/ml/linalg/Vectors.java
##########
@@ -30,4 +30,9 @@ public static DenseVector dense(double... values) {
public static SparseVector sparse(int size, int[] indices, double[] values) {
return new SparseVector(size, indices, values);
}
+
+ /** Converts a sparse vector to a dense vector. */
+ public static DenseVector toDense(SparseVector vec) {
Review comment:
Do you think we should define it as a member method of `Vector` or `SparseVector`, as Alink and Spark have done?
If it would be better to be a static method, would it be better to change the signature to `DenseVector toDense(Vector)`?
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/common/param/HasWithMean.java
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.common.param;
+
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.WithParams;
+
+/** Whether centers the data with mean before scaling or not. */
Review comment:
Maybe better to follow the pattern of other parameter interfaces' javadocs.
##########
File path: flink-ml-core/src/main/java/org/apache/flink/ml/linalg/DenseVector.java
##########
@@ -68,4 +68,9 @@ public boolean equals(Object obj) {
public int hashCode() {
return Arrays.hashCode(values);
}
+
+ @Override
+ public Vector clone() {
Review comment:
`clone()` has been provided by `Object`. Do you think we should remove the `Vector.clone()` API, and let this method returns `DenseVector`? Same as in `SparseVector`.
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/common/param/HasWithMean.java
##########
@@ -0,0 +1,38 @@
+/*
+ * 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.common.param;
+
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.WithParams;
+
+/** Whether centers the data with mean before scaling or not. */
+public interface HasWithMean<T> extends WithParams<T> {
Review comment:
In Alink `HasWithMean` is only used in (Vector)StandardTrainParams. Thus shall we add these methods directly to `StandardScalerParams` instead of making it a shared parameter? Same for withStd.
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.java
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.standardscaler;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+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.Vectors;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+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;
+
+/**
+ * An Estimator which implements the standard scaling algorithm.
+ *
+ * <p>Standardization is a common requirement for machine learning training because they may behave
+ * badly if the individual features of a input do not look like standard normally distributed data
+ * (e.g. Gaussian with 0 mean and unit variance).
+ *
+ * <p>This estimator standardizes the input features by removing the mean and scaling each dimension
+ * to unit variance.
+ */
+public class StandardScaler
+ implements Estimator<StandardScaler, StandardScalerModel>,
+ StandardScalerParams<StandardScaler> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public StandardScaler() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public StandardScalerModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Tuple3<DenseVector, DenseVector, Long>> sumAndSquaredSumAndWeight =
+ tEnv.toDataStream(inputs[0])
+ .transform(
+ "computeMeta",
+ new TupleTypeInfo<>(
+ TypeInformation.of(DenseVector.class),
+ TypeInformation.of(DenseVector.class),
+ BasicTypeInfo.LONG_TYPE_INFO),
+ new ComputeMeta(getFeaturesCol()));
+
+ DataStream<StandardScalerModelData> modelData =
+ sumAndSquaredSumAndWeight
+ .transform(
+ "buildModel",
+ TypeInformation.of(StandardScalerModelData.class),
+ new BuildModel())
+ .setParallelism(1);
+
+ StandardScalerModel model =
+ new StandardScalerModel().setModelData(tEnv.fromDataStream(modelData));
+ ReadWriteUtils.updateExistingParams(model, paramMap);
+ return model;
+ }
+
+ /**
+ * Builds the {@link StandardScalerModelData} using the meta data computed on each partition.
+ */
+ private static class BuildModel extends AbstractStreamOperator<StandardScalerModelData>
Review comment:
It might be better to make class names a noun, like `BuildModelOperator` or `ModelBuilder`. Same for `ComputeMeta`.
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScalerModel.java
##########
@@ -0,0 +1,185 @@
+/*
+ * 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.standardscaler;
+
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+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.SparseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.Vectors;
+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.streaming.api.environment.StreamExecutionEnvironment;
+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 transforms data using the model data computed by {@link StandardScaler}. */
+public class StandardScalerModel
+ implements Model<StandardScalerModel>, StandardScalerParams<StandardScalerModel> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+ private Table modelDataTable;
+
+ public StandardScalerModel() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked, rawtypes")
+ 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(), TypeInformation.of(Vector.class)),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getPredictionCol()));
+
+ final String broadcastModelKey = "broadcastModelKey";
+ DataStream<StandardScalerModelData> modelDataStream =
+ StandardScalerModelData.getModelDataStream(modelDataTable);
+
+ DataStream<Row> predictionResult =
+ BroadcastUtils.withBroadcastStream(
+ Collections.singletonList(inputStream),
+ Collections.singletonMap(broadcastModelKey, modelDataStream),
+ inputList -> {
+ DataStream inputData = inputList.get(0);
+ return inputData.map(
+ new PredictOutputFunction(
+ broadcastModelKey,
+ getFeaturesCol(),
+ getWithMean(),
+ getWithStd()),
+ outputTypeInfo);
+ });
+
+ return new Table[] {tEnv.fromDataStream(predictionResult)};
+ }
+
+ /** A utility function used for prediction. */
+ private static class PredictOutputFunction extends RichMapFunction<Row, Row> {
+ private final String broadcastModelKey;
+ private final String featuresCol;
+ private final boolean withMean;
+ private final boolean withStd;
+ private DenseVector mean;
+ private DenseVector scale;
+
+ public PredictOutputFunction(
+ String broadcastModelKey, String featuresCol, boolean withMean, boolean withStd) {
+ this.broadcastModelKey = broadcastModelKey;
+ this.featuresCol = featuresCol;
+ this.withMean = withMean;
+ this.withStd = withStd;
+ }
+
+ @Override
+ public Row map(Row dataPoint) {
+ if (mean == null) {
+ StandardScalerModelData modelData =
+ (StandardScalerModelData)
+ getRuntimeContext().getBroadcastVariable(broadcastModelKey).get(0);
+ mean = modelData.mean;
+ DenseVector std = modelData.std;
+
+ if (withStd) {
+ scale = std;
+ double[] scaleValues = scale.values;
+ for (int i = 0; i < scaleValues.length; i++) {
+ scaleValues[i] = scaleValues[i] == 0 ? 0 : 1 / scaleValues[i];
+ }
+ }
+ }
+
+ Vector feature = ((Vector) dataPoint.getField(featuresCol)).clone();
+
+ if (withMean && feature instanceof SparseVector) {
+ feature = Vectors.toDense((SparseVector) feature);
Review comment:
I guess using `toDense` is to accelerate the incoming BLAS operations. If so, how about optimizing this part of code like this?
- if `hDot` should also benefit from `toDense`, then change the `if` condition to `withMean || withStd`
- if not, place the `hDot` operation to before `toDense`.
- if `!withMean && !withStd`, then there is no need to do a `clone()` operation.
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.java
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.standardscaler;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+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.Vectors;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+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;
+
+/**
+ * An Estimator which implements the standard scaling algorithm.
+ *
+ * <p>Standardization is a common requirement for machine learning training because they may behave
+ * badly if the individual features of a input do not look like standard normally distributed data
+ * (e.g. Gaussian with 0 mean and unit variance).
+ *
+ * <p>This estimator standardizes the input features by removing the mean and scaling each dimension
+ * to unit variance.
+ */
+public class StandardScaler
+ implements Estimator<StandardScaler, StandardScalerModel>,
+ StandardScalerParams<StandardScaler> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public StandardScaler() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public StandardScalerModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Tuple3<DenseVector, DenseVector, Long>> sumAndSquaredSumAndWeight =
+ tEnv.toDataStream(inputs[0])
+ .transform(
+ "computeMeta",
+ new TupleTypeInfo<>(
+ TypeInformation.of(DenseVector.class),
+ TypeInformation.of(DenseVector.class),
+ BasicTypeInfo.LONG_TYPE_INFO),
+ new ComputeMeta(getFeaturesCol()));
+
+ DataStream<StandardScalerModelData> modelData =
+ sumAndSquaredSumAndWeight
+ .transform(
+ "buildModel",
+ TypeInformation.of(StandardScalerModelData.class),
+ new BuildModel())
+ .setParallelism(1);
+
+ StandardScalerModel model =
+ new StandardScalerModel().setModelData(tEnv.fromDataStream(modelData));
+ ReadWriteUtils.updateExistingParams(model, paramMap);
+ return model;
+ }
+
+ /**
+ * Builds the {@link StandardScalerModelData} using the meta data computed on each partition.
+ */
+ private static class BuildModel extends AbstractStreamOperator<StandardScalerModelData>
+ implements OneInputStreamOperator<
+ Tuple3<DenseVector, DenseVector, Long>, StandardScalerModelData>,
+ BoundedOneInput {
+ private ListState<DenseVector> sumState;
+ private ListState<DenseVector> squaredSumState;
+ private ListState<Long> numElementsState;
+ private DenseVector sum;
+ private DenseVector squaredSum;
+ private long numElements;
+
+ @Override
+ public void endInput() {
+ if (numElements > 0) {
+ BLAS.scal(1.0 / numElements, sum);
+ double[] mean = sum.values;
+ double[] std = squaredSum.values;
Review comment:
nit: maybe `double[] std = new double[squaredSum.size()]` is better? Or just write the calculated results to `squaredSum.values`.
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScalerModel.java
##########
@@ -0,0 +1,185 @@
+/*
+ * 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.standardscaler;
+
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+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.SparseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.Vectors;
+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.streaming.api.environment.StreamExecutionEnvironment;
+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 transforms data using the model data computed by {@link StandardScaler}. */
+public class StandardScalerModel
Review comment:
Do you think we can merge the model data and model operator of MinMaxScaler and Standard Scaler? I noticed that they are quite similar.
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.java
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.standardscaler;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+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.Vectors;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+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;
+
+/**
+ * An Estimator which implements the standard scaling algorithm.
+ *
+ * <p>Standardization is a common requirement for machine learning training because they may behave
+ * badly if the individual features of a input do not look like standard normally distributed data
+ * (e.g. Gaussian with 0 mean and unit variance).
+ *
+ * <p>This estimator standardizes the input features by removing the mean and scaling each dimension
+ * to unit variance.
+ */
+public class StandardScaler
+ implements Estimator<StandardScaler, StandardScalerModel>,
+ StandardScalerParams<StandardScaler> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public StandardScaler() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public StandardScalerModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Tuple3<DenseVector, DenseVector, Long>> sumAndSquaredSumAndWeight =
+ tEnv.toDataStream(inputs[0])
+ .transform(
+ "computeMeta",
+ new TupleTypeInfo<>(
+ TypeInformation.of(DenseVector.class),
+ TypeInformation.of(DenseVector.class),
+ BasicTypeInfo.LONG_TYPE_INFO),
+ new ComputeMeta(getFeaturesCol()));
+
+ DataStream<StandardScalerModelData> modelData =
+ sumAndSquaredSumAndWeight
+ .transform(
+ "buildModel",
+ TypeInformation.of(StandardScalerModelData.class),
+ new BuildModel())
+ .setParallelism(1);
+
+ StandardScalerModel model =
+ new StandardScalerModel().setModelData(tEnv.fromDataStream(modelData));
+ ReadWriteUtils.updateExistingParams(model, paramMap);
+ return model;
+ }
+
+ /**
+ * Builds the {@link StandardScalerModelData} using the meta data computed on each partition.
+ */
+ private static class BuildModel extends AbstractStreamOperator<StandardScalerModelData>
+ implements OneInputStreamOperator<
+ Tuple3<DenseVector, DenseVector, Long>, StandardScalerModelData>,
+ BoundedOneInput {
+ private ListState<DenseVector> sumState;
+ private ListState<DenseVector> squaredSumState;
+ private ListState<Long> numElementsState;
+ private DenseVector sum;
+ private DenseVector squaredSum;
+ private long numElements;
+
+ @Override
+ public void endInput() {
+ if (numElements > 0) {
+ BLAS.scal(1.0 / numElements, sum);
+ double[] mean = sum.values;
+ double[] std = squaredSum.values;
+ for (int i = 0; i < mean.length; i++) {
+ std[i] =
+ Math.sqrt(
+ (squaredSum.values[i] - numElements * mean[i] * mean[i])
+ / (numElements - 1));
Review comment:
Might throw error if `numElements == 1`.
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.java
##########
@@ -0,0 +1,290 @@
+/*
+ * 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.standardscaler;
+
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.iteration.operator.OperatorStateUtils;
+import org.apache.flink.ml.api.Estimator;
+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.Vectors;
+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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+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;
+
+/**
+ * An Estimator which implements the standard scaling algorithm.
+ *
+ * <p>Standardization is a common requirement for machine learning training because they may behave
+ * badly if the individual features of a input do not look like standard normally distributed data
+ * (e.g. Gaussian with 0 mean and unit variance).
+ *
+ * <p>This estimator standardizes the input features by removing the mean and scaling each dimension
+ * to unit variance.
+ */
+public class StandardScaler
+ implements Estimator<StandardScaler, StandardScalerModel>,
+ StandardScalerParams<StandardScaler> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public StandardScaler() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public StandardScalerModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ StreamTableEnvironment tEnv =
+ (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment();
+ DataStream<Tuple3<DenseVector, DenseVector, Long>> sumAndSquaredSumAndWeight =
+ tEnv.toDataStream(inputs[0])
+ .transform(
+ "computeMeta",
+ new TupleTypeInfo<>(
+ TypeInformation.of(DenseVector.class),
+ TypeInformation.of(DenseVector.class),
+ BasicTypeInfo.LONG_TYPE_INFO),
+ new ComputeMeta(getFeaturesCol()));
Review comment:
How about using the following structure here?
```java
.keyBy(<random_key>)
.windowAll()
.aggregate()/.reduce()
```
The expected behavior is to accumulate computed results and output when all inputs are finished, which I think can be completed by the code above. The code above also preserves parallelism of your job. Besides, it would use higher abstraction-level Flink API so you need not care about some implementation details. Using a keyBy also allows you to use `ValueState` instead of `ListState` + `getUniqueElement`.
Same for the following part of code about `BuildModel`.
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScalerModel.java
##########
@@ -0,0 +1,185 @@
+/*
+ * 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.standardscaler;
+
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+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.SparseVector;
+import org.apache.flink.ml.linalg.Vector;
+import org.apache.flink.ml.linalg.Vectors;
+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.streaming.api.environment.StreamExecutionEnvironment;
+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 transforms data using the model data computed by {@link StandardScaler}. */
+public class StandardScalerModel
+ implements Model<StandardScalerModel>, StandardScalerParams<StandardScalerModel> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+ private Table modelDataTable;
+
+ public StandardScalerModel() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked, rawtypes")
+ 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(), TypeInformation.of(Vector.class)),
+ ArrayUtils.addAll(inputTypeInfo.getFieldNames(), getPredictionCol()));
+
+ final String broadcastModelKey = "broadcastModelKey";
+ DataStream<StandardScalerModelData> modelDataStream =
+ StandardScalerModelData.getModelDataStream(modelDataTable);
+
+ DataStream<Row> predictionResult =
+ BroadcastUtils.withBroadcastStream(
+ Collections.singletonList(inputStream),
+ Collections.singletonMap(broadcastModelKey, modelDataStream),
+ inputList -> {
+ DataStream inputData = inputList.get(0);
+ return inputData.map(
+ new PredictOutputFunction(
+ broadcastModelKey,
+ getFeaturesCol(),
+ getWithMean(),
+ getWithStd()),
+ outputTypeInfo);
+ });
+
+ return new Table[] {tEnv.fromDataStream(predictionResult)};
+ }
+
+ /** A utility function used for prediction. */
+ private static class PredictOutputFunction extends RichMapFunction<Row, Row> {
+ private final String broadcastModelKey;
+ private final String featuresCol;
+ private final boolean withMean;
+ private final boolean withStd;
+ private DenseVector mean;
+ private DenseVector scale;
+
+ public PredictOutputFunction(
+ String broadcastModelKey, String featuresCol, boolean withMean, boolean withStd) {
+ this.broadcastModelKey = broadcastModelKey;
+ this.featuresCol = featuresCol;
+ this.withMean = withMean;
+ this.withStd = withStd;
+ }
+
+ @Override
+ public Row map(Row dataPoint) {
+ if (mean == null) {
+ StandardScalerModelData modelData =
+ (StandardScalerModelData)
+ getRuntimeContext().getBroadcastVariable(broadcastModelKey).get(0);
+ mean = modelData.mean;
+ DenseVector std = modelData.std;
+
+ if (withStd) {
+ scale = std;
+ double[] scaleValues = scale.values;
+ for (int i = 0; i < scaleValues.length; i++) {
+ scaleValues[i] = scaleValues[i] == 0 ? 0 : 1 / scaleValues[i];
+ }
+ }
+ }
+
+ Vector feature = ((Vector) dataPoint.getField(featuresCol)).clone();
+
+ if (withMean && feature instanceof SparseVector) {
+ feature = Vectors.toDense((SparseVector) feature);
+ }
+ if (withMean) {
Review comment:
If `withMean` or `withStd` is set to false, we may also be able to optimize the training process, as some values calculated during the training process is never used in prediction.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org