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/21 12:12:10 UTC
[GitHub] [flink-ml] zhipeng93 opened a new pull request #73: [FLINK-26626] Add Transformer And Estimator of StandardScaler
zhipeng93 opened a new pull request #73:
URL: https://github.com/apache/flink-ml/pull/73
... Not ready for review.
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r834119649
##########
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:
Good question! There has also been a discussion of the reason that BLAS does not provide a hadamard dot [1] --- Mostly saying that there are no performance gains.
However, I'd like to say that introducing `hDot` here is simply for convienice. I need to use it twice.
By the way, we can replace the logic in MinMaxScaler if we have this method. This will certainly simplify the implementation.
What do you think?
[1] https://github.com/xianyi/OpenBLAS/issues/1083
--
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
[GitHub] [flink-ml] yunfengzhou-hub commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
yunfengzhou-hub commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r836024449
##########
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:
I think window operation would only cache data if the following operator is `apply`. In cases when the following is `aggregate/reduce`, the window operation processes a stream record as soon as it is received, so it would only need to cache the accumulator / reduce result.
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r840212185
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.java
##########
@@ -0,0 +1,288 @@
+/*
+ * 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.Arrays;
+import java.util.Collections;
+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 ComputeMetaOperator(getFeaturesCol()));
+
+ DataStream<StandardScalerModelData> modelData =
+ sumAndSquaredSumAndWeight
+ .transform(
+ "buildModel",
+ TypeInformation.of(StandardScalerModelData.class),
+ new BuildModelOperator())
+ .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 BuildModelOperator 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;
+ if (numElements > 1) {
+ for (int i = 0; i < mean.length; i++) {
+ std[i] =
+ Math.sqrt(
+ (squaredSum.values[i] - numElements * mean[i] * mean[i])
+ / (numElements - 1));
+ }
+ } else {
+ Arrays.fill(std, 0.0);
+ }
+
+ output.collect(
+ new StreamRecord<>(
+ new StandardScalerModelData(
+ Vectors.dense(mean), Vectors.dense(std))));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple3<DenseVector, DenseVector, Long>> element) {
+ Tuple3<DenseVector, DenseVector, Long> value = element.getValue();
+ if (sum == null) {
+ sum = value.f0;
+ squaredSum = value.f1;
+ numElements = value.f2;
+ } else {
+ BLAS.axpy(1, value.f0, sum);
+ BLAS.axpy(1, value.f1, squaredSum);
+ numElements += value.f2;
+ }
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
+ sumState.update(Collections.singletonList(sum));
+ squaredSumState.update(Collections.singletonList(squaredSum));
+ numElementsState.update(Collections.singletonList(numElements));
+ }
+ }
+ }
+
+ /** Computes sum, squared sum and number of elements in each partition. */
+ private static class ComputeMetaOperator
+ extends AbstractStreamOperator<Tuple3<DenseVector, DenseVector, Long>>
+ implements OneInputStreamOperator<Row, Tuple3<DenseVector, DenseVector, Long>>,
+ BoundedOneInput {
+ private ListState<DenseVector> sumState;
+ private ListState<DenseVector> squaredSumState;
+ private ListState<Long> numElementsState;
+ private DenseVector sum;
+ private DenseVector squaredSum;
+ private long numElements;
+
+ private final String featuresCol;
+
+ public ComputeMetaOperator(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public void endInput() {
+ if (numElements > 0) {
+ output.collect(new StreamRecord<>(Tuple3.of(sum, squaredSum, numElements)));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> element) throws Exception {
+ Vector feature = (Vector) element.getValue().getField(featuresCol);
+ if (sum == null) {
+ sum = new DenseVector(feature.size());
+ squaredSum = new DenseVector(feature.size());
+ }
+ BLAS.axpy(1, feature, sum);
+ BLAS.hDot(feature, feature);
+ BLAS.axpy(1, feature, squaredSum);
+ numElements++;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
Review comment:
Good point.
I agree with the behavior that when the train data is empty, we should output a `special model data` to indicate that we cannot do inference. I have updated the code to handle this case and also added one test case. (See for `StandardScalerTest#testFitOnEmptyData`).
For other checks about when numElements is zero or not, we probably need to keep it.
For example, when the training data is not empty, there could be one worker that contains zero data points. If we remove the check, it throws an nullpointer exception.
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.java
##########
@@ -0,0 +1,288 @@
+/*
+ * 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.Arrays;
+import java.util.Collections;
+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 ComputeMetaOperator(getFeaturesCol()));
+
+ DataStream<StandardScalerModelData> modelData =
+ sumAndSquaredSumAndWeight
+ .transform(
+ "buildModel",
+ TypeInformation.of(StandardScalerModelData.class),
+ new BuildModelOperator())
+ .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 BuildModelOperator 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;
+ if (numElements > 1) {
+ for (int i = 0; i < mean.length; i++) {
+ std[i] =
+ Math.sqrt(
+ (squaredSum.values[i] - numElements * mean[i] * mean[i])
+ / (numElements - 1));
+ }
+ } else {
+ Arrays.fill(std, 0.0);
+ }
+
+ output.collect(
+ new StreamRecord<>(
+ new StandardScalerModelData(
+ Vectors.dense(mean), Vectors.dense(std))));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple3<DenseVector, DenseVector, Long>> element) {
+ Tuple3<DenseVector, DenseVector, Long> value = element.getValue();
+ if (sum == null) {
+ sum = value.f0;
+ squaredSum = value.f1;
+ numElements = value.f2;
+ } else {
+ BLAS.axpy(1, value.f0, sum);
+ BLAS.axpy(1, value.f1, squaredSum);
+ numElements += value.f2;
+ }
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
+ sumState.update(Collections.singletonList(sum));
+ squaredSumState.update(Collections.singletonList(squaredSum));
+ numElementsState.update(Collections.singletonList(numElements));
+ }
+ }
+ }
+
+ /** Computes sum, squared sum and number of elements in each partition. */
+ private static class ComputeMetaOperator
+ extends AbstractStreamOperator<Tuple3<DenseVector, DenseVector, Long>>
+ implements OneInputStreamOperator<Row, Tuple3<DenseVector, DenseVector, Long>>,
+ BoundedOneInput {
+ private ListState<DenseVector> sumState;
+ private ListState<DenseVector> squaredSumState;
+ private ListState<Long> numElementsState;
+ private DenseVector sum;
+ private DenseVector squaredSum;
+ private long numElements;
+
+ private final String featuresCol;
+
+ public ComputeMetaOperator(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public void endInput() {
+ if (numElements > 0) {
+ output.collect(new StreamRecord<>(Tuple3.of(sum, squaredSum, numElements)));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> element) throws Exception {
+ Vector feature = (Vector) element.getValue().getField(featuresCol);
+ if (sum == null) {
+ sum = new DenseVector(feature.size());
+ squaredSum = new DenseVector(feature.size());
+ }
+ BLAS.axpy(1, feature, sum);
+ BLAS.hDot(feature, feature);
+ BLAS.axpy(1, feature, squaredSum);
+ numElements++;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
Review comment:
Good point.
I agree with the behavior that when the train data is empty, we should output a `special model data` to indicate that we cannot do inference. I have updated the code to handle this case and also added one test case. (See for `StandardScalerTest#testFitOnEmptyData`).
For other checks about when numElements is zero or not, we probably need to keep it. For example, when the training data is not empty, there could be one worker that contains zero data points. If we remove the check, it throws an nullpointer exception.
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r840204804
##########
File path: flink-ml-core/src/main/java/org/apache/flink/ml/linalg/BLAS.java
##########
@@ -89,4 +111,56 @@ public static void gemv(
y.values,
1);
}
+
+ private static void axpy(double a, DenseVector x, DenseVector y) {
+ JAVA_BLAS.daxpy(x.size(), a, x.values, 1, y.values, 1);
+ }
+
+ private static void axpy(double a, SparseVector x, DenseVector y) {
+ for (int i = 0; i < x.indices.length; i++) {
+ int index = x.indices[i];
+ y.values[index] += a * x.values[i];
+ }
+ }
+
+ private static void hDot(SparseVector x, SparseVector y) {
+ int idx = 0;
+ int idy = 0;
+ while (idx < x.indices.length && idy < y.indices.length) {
+ int indexX = x.indices[idx];
+ while (idy < y.indices.length && y.indices[idy] < indexX) {
+ y.values[idy] = 0;
+ idy++;
+ }
+ if (idy < y.indices.length && y.indices[idy] == indexX) {
+ y.values[idy] *= x.values[idx];
+ idy++;
+ }
+ idx++;
+ }
+ }
+
+ private static void hDot(SparseVector x, DenseVector y) {
+ int idx = 0;
+ for (int i = 0; i < y.size(); i++) {
+ if (x.indices[idx] == i) {
Review comment:
Thanks for pointing this out. It is fixed.
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r834170216
##########
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:
They are indeed similar in -- both of them are composed of two vectors.
I am okay to merge them as one. My Concern is that what are we going to name the new class? Do you have some suggestions?
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r840992081
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.java
##########
@@ -0,0 +1,288 @@
+/*
+ * 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.Arrays;
+import java.util.Collections;
+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 ComputeMetaOperator(getFeaturesCol()));
+
+ DataStream<StandardScalerModelData> modelData =
+ sumAndSquaredSumAndWeight
+ .transform(
+ "buildModel",
+ TypeInformation.of(StandardScalerModelData.class),
+ new BuildModelOperator())
+ .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 BuildModelOperator 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;
+ if (numElements > 1) {
+ for (int i = 0; i < mean.length; i++) {
+ std[i] =
+ Math.sqrt(
+ (squaredSum.values[i] - numElements * mean[i] * mean[i])
+ / (numElements - 1));
+ }
+ } else {
+ Arrays.fill(std, 0.0);
+ }
+
+ output.collect(
+ new StreamRecord<>(
+ new StandardScalerModelData(
+ Vectors.dense(mean), Vectors.dense(std))));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple3<DenseVector, DenseVector, Long>> element) {
+ Tuple3<DenseVector, DenseVector, Long> value = element.getValue();
+ if (sum == null) {
+ sum = value.f0;
+ squaredSum = value.f1;
+ numElements = value.f2;
+ } else {
+ BLAS.axpy(1, value.f0, sum);
+ BLAS.axpy(1, value.f1, squaredSum);
+ numElements += value.f2;
+ }
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
+ sumState.update(Collections.singletonList(sum));
+ squaredSumState.update(Collections.singletonList(squaredSum));
+ numElementsState.update(Collections.singletonList(numElements));
+ }
+ }
+ }
+
+ /** Computes sum, squared sum and number of elements in each partition. */
+ private static class ComputeMetaOperator
+ extends AbstractStreamOperator<Tuple3<DenseVector, DenseVector, Long>>
+ implements OneInputStreamOperator<Row, Tuple3<DenseVector, DenseVector, Long>>,
+ BoundedOneInput {
+ private ListState<DenseVector> sumState;
+ private ListState<DenseVector> squaredSumState;
+ private ListState<Long> numElementsState;
+ private DenseVector sum;
+ private DenseVector squaredSum;
+ private long numElements;
+
+ private final String featuresCol;
+
+ public ComputeMetaOperator(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public void endInput() {
+ if (numElements > 0) {
+ output.collect(new StreamRecord<>(Tuple3.of(sum, squaredSum, numElements)));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> element) throws Exception {
+ Vector feature = (Vector) element.getValue().getField(featuresCol);
+ if (sum == null) {
+ sum = new DenseVector(feature.size());
+ squaredSum = new DenseVector(feature.size());
+ }
+ BLAS.axpy(1, feature, sum);
+ BLAS.hDot(feature, feature);
+ BLAS.axpy(1, feature, squaredSum);
+ numElements++;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
Review comment:
Thanks for pointint out that "The error should be moved closer to where it happens" and I also believe it is reasonable to throw exception when the training set is empty.
I have addressed this in the latest commit.
--
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
[GitHub] [flink-ml] yunfengzhou-hub commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
yunfengzhou-hub commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r836020739
##########
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:
Sounds good.
--
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
[GitHub] [flink-ml] yunfengzhou-hub commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
yunfengzhou-hub commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r836022735
##########
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:
I think we need more offline discussion about this API. Another choice is to still keep `MinMaxScalerModel` and `StandardScalerModel`, but they actually extend the same parent class without overriding any one of the parent class's methods.
--
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
[GitHub] [flink-ml] yunfengzhou-hub commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
yunfengzhou-hub commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r836019685
##########
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:
I think the second one would be better, as it reuses existing API and is more straight forward.
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r834144685
##########
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:
`toDense` is not used for performance here...
It is because we can only support in-place adding a `SparseVector/DenseVector` to a `DenseVecor` in `axpy`.
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r840189898
##########
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:
As we discussed offline, we agree to keep the code for now because it is an implementation issue and do not affect end users.
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r834124527
##########
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:
Thanks for pointing this out. The `clone()` here is to create a new instance of this Vector for further in-place operations, in which case we may not want to update the vectors directly.
I think there could be two alternatives:
- How about we rename the method to `duplicate` or something like `copy`?
- Or how about we remove the `clone()` in `Vector`, but keep it in `SparseVector` and `DenseVector`, given that we are indeed cloning vectors.
What do you think?
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r836088275
##########
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:
With some offline discussion, Weibo is fine with the change and also thinks it is better to keep the code consistent.
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r836088275
##########
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 talked with Weibo. He is fine with the change and also thinks it is better to keep the code consistent.
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r841019214
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScalerParams.java
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params for {@link StandardScaler}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface StandardScalerParams<T> extends HasFeaturesCol<T>, HasPredictionCol<T> {
Review comment:
Thanks for raising up this discussion.
As `StandardScaler` is an `estmator`, it aims to train on the input features and predict on the output. So I thought it might be ok to keep the current naming. So soes `MinMaxScaler`.
> HasFeaturesCol should probably be renamed to HasFeatureCol as it represents just one column.
Here `features` is one column.
--
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
[GitHub] [flink-ml] yunfengzhou-hub commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
yunfengzhou-hub commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r836021390
##########
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:
Got it. Then I'm good with the current implementation.
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r836247834
##########
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:
Thanks for the detail explanation. When using `keyBy().window().aggregate()`, I came into another problem [1]. It seems that it is not solved yet. So let's use the current implementation for now.
[1] https://lists.apache.org/thread/7jm3sdh2h3y2gks3rp8kn9yb44b55dzv
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r840215290
##########
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) {
+ Preconditions.checkArgument(x.size() == y.size(), "Vector size mismatched.");
+ if (y instanceof DenseVector) {
+ if (x instanceof SparseVector) {
Review comment:
Thanks for the comments. I have reordered the order of `a instanceof b`.
BTW, we do not support `axpy` when `y` is a SparseVector because it may lead to some memory re-allocation.
--
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
[GitHub] [flink-ml] lindong28 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r840980426
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.java
##########
@@ -0,0 +1,288 @@
+/*
+ * 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.Arrays;
+import java.util.Collections;
+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 ComputeMetaOperator(getFeaturesCol()));
+
+ DataStream<StandardScalerModelData> modelData =
+ sumAndSquaredSumAndWeight
+ .transform(
+ "buildModel",
+ TypeInformation.of(StandardScalerModelData.class),
+ new BuildModelOperator())
+ .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 BuildModelOperator 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;
+ if (numElements > 1) {
+ for (int i = 0; i < mean.length; i++) {
+ std[i] =
+ Math.sqrt(
+ (squaredSum.values[i] - numElements * mean[i] * mean[i])
+ / (numElements - 1));
+ }
+ } else {
+ Arrays.fill(std, 0.0);
+ }
+
+ output.collect(
+ new StreamRecord<>(
+ new StandardScalerModelData(
+ Vectors.dense(mean), Vectors.dense(std))));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple3<DenseVector, DenseVector, Long>> element) {
+ Tuple3<DenseVector, DenseVector, Long> value = element.getValue();
+ if (sum == null) {
+ sum = value.f0;
+ squaredSum = value.f1;
+ numElements = value.f2;
+ } else {
+ BLAS.axpy(1, value.f0, sum);
+ BLAS.axpy(1, value.f1, squaredSum);
+ numElements += value.f2;
+ }
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
+ sumState.update(Collections.singletonList(sum));
+ squaredSumState.update(Collections.singletonList(squaredSum));
+ numElementsState.update(Collections.singletonList(numElements));
+ }
+ }
+ }
+
+ /** Computes sum, squared sum and number of elements in each partition. */
+ private static class ComputeMetaOperator
+ extends AbstractStreamOperator<Tuple3<DenseVector, DenseVector, Long>>
+ implements OneInputStreamOperator<Row, Tuple3<DenseVector, DenseVector, Long>>,
+ BoundedOneInput {
+ private ListState<DenseVector> sumState;
+ private ListState<DenseVector> squaredSumState;
+ private ListState<Long> numElementsState;
+ private DenseVector sum;
+ private DenseVector squaredSum;
+ private long numElements;
+
+ private final String featuresCol;
+
+ public ComputeMetaOperator(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public void endInput() {
+ if (numElements > 0) {
Review comment:
Sounds good.
--
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
[GitHub] [flink-ml] yunfengzhou-hub commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
yunfengzhou-hub commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r836019213
##########
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 think it would be a good idea to replace the logic in MinMaxScaler. But Weibo was concerned about the performance, as the implementation would be changed from something like
```java
for (int i = 0; i < n; i++) {
y[i] += a[i] * x[i];
}
```
to
```java
for (int i = 0; i < n; i++) {
x[i] *= a[i];
}
for (int i = 0; i < n; i++) {
y[i] += x[i];
}
```
Thus I would like to hear from Weibo's opinion about this replacement.
--
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
[GitHub] [flink-ml] lindong28 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r840980952
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.java
##########
@@ -0,0 +1,288 @@
+/*
+ * 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.Arrays;
+import java.util.Collections;
+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 ComputeMetaOperator(getFeaturesCol()));
+
+ DataStream<StandardScalerModelData> modelData =
+ sumAndSquaredSumAndWeight
+ .transform(
+ "buildModel",
+ TypeInformation.of(StandardScalerModelData.class),
+ new BuildModelOperator())
+ .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 BuildModelOperator 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;
+ if (numElements > 1) {
+ for (int i = 0; i < mean.length; i++) {
+ std[i] =
+ Math.sqrt(
+ (squaredSum.values[i] - numElements * mean[i] * mean[i])
+ / (numElements - 1));
+ }
+ } else {
+ Arrays.fill(std, 0.0);
+ }
+
+ output.collect(
+ new StreamRecord<>(
+ new StandardScalerModelData(
+ Vectors.dense(mean), Vectors.dense(std))));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple3<DenseVector, DenseVector, Long>> element) {
+ Tuple3<DenseVector, DenseVector, Long> value = element.getValue();
+ if (sum == null) {
+ sum = value.f0;
+ squaredSum = value.f1;
+ numElements = value.f2;
+ } else {
+ BLAS.axpy(1, value.f0, sum);
+ BLAS.axpy(1, value.f1, squaredSum);
+ numElements += value.f2;
+ }
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
+ sumState.update(Collections.singletonList(sum));
+ squaredSumState.update(Collections.singletonList(squaredSum));
+ numElementsState.update(Collections.singletonList(numElements));
+ }
+ }
+ }
+
+ /** Computes sum, squared sum and number of elements in each partition. */
+ private static class ComputeMetaOperator
+ extends AbstractStreamOperator<Tuple3<DenseVector, DenseVector, Long>>
+ implements OneInputStreamOperator<Row, Tuple3<DenseVector, DenseVector, Long>>,
+ BoundedOneInput {
+ private ListState<DenseVector> sumState;
+ private ListState<DenseVector> squaredSumState;
+ private ListState<Long> numElementsState;
+ private DenseVector sum;
+ private DenseVector squaredSum;
+ private long numElements;
+
+ private final String featuresCol;
+
+ public ComputeMetaOperator(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public void endInput() {
+ if (numElements > 0) {
+ output.collect(new StreamRecord<>(Tuple3.of(sum, squaredSum, numElements)));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> element) throws Exception {
+ Vector feature = (Vector) element.getValue().getField(featuresCol);
+ if (sum == null) {
+ sum = new DenseVector(feature.size());
+ squaredSum = new DenseVector(feature.size());
+ }
+ BLAS.axpy(1, feature, sum);
+ BLAS.hDot(feature, feature);
+ BLAS.axpy(1, feature, squaredSum);
+ numElements++;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
Review comment:
After thinking about this more, I think it might be better to let `estimator.fit(...)` if input data is empty during runtime, so that the error could be moved closer to where it happens, i.e. in the operator that does training instead of the operator that does inference. And this could also make the logic in the model simpler and consistent with other existing models.
What do you think?
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r841032715
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScalerParams.java
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params for {@link StandardScaler}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface StandardScalerParams<T> extends HasFeaturesCol<T>, HasPredictionCol<T> {
Review comment:
After some offline discussion, we agree to follow the following rules:
- use `inputCol/outputCol/inputCols/outputCos` when doing feature engineering. (e.g., code under package `org.apache.flink.ml.feature`)
- use `featuresCol/predictionCol` when doing (un)supervised machine learning training (e.g., code under package `org.apache.flink.ml.classification, org.apache.flink.ml.clustering`
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r840212010
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.java
##########
@@ -0,0 +1,288 @@
+/*
+ * 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.Arrays;
+import java.util.Collections;
+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 ComputeMetaOperator(getFeaturesCol()));
+
+ DataStream<StandardScalerModelData> modelData =
+ sumAndSquaredSumAndWeight
+ .transform(
+ "buildModel",
+ TypeInformation.of(StandardScalerModelData.class),
+ new BuildModelOperator())
+ .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 BuildModelOperator 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;
+ if (numElements > 1) {
+ for (int i = 0; i < mean.length; i++) {
+ std[i] =
+ Math.sqrt(
+ (squaredSum.values[i] - numElements * mean[i] * mean[i])
+ / (numElements - 1));
+ }
+ } else {
+ Arrays.fill(std, 0.0);
+ }
+
+ output.collect(
+ new StreamRecord<>(
+ new StandardScalerModelData(
+ Vectors.dense(mean), Vectors.dense(std))));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple3<DenseVector, DenseVector, Long>> element) {
+ Tuple3<DenseVector, DenseVector, Long> value = element.getValue();
+ if (sum == null) {
+ sum = value.f0;
+ squaredSum = value.f1;
+ numElements = value.f2;
+ } else {
+ BLAS.axpy(1, value.f0, sum);
+ BLAS.axpy(1, value.f1, squaredSum);
+ numElements += value.f2;
+ }
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
+ sumState.update(Collections.singletonList(sum));
+ squaredSumState.update(Collections.singletonList(squaredSum));
+ numElementsState.update(Collections.singletonList(numElements));
+ }
+ }
+ }
+
+ /** Computes sum, squared sum and number of elements in each partition. */
+ private static class ComputeMetaOperator
+ extends AbstractStreamOperator<Tuple3<DenseVector, DenseVector, Long>>
+ implements OneInputStreamOperator<Row, Tuple3<DenseVector, DenseVector, Long>>,
+ BoundedOneInput {
+ private ListState<DenseVector> sumState;
+ private ListState<DenseVector> squaredSumState;
+ private ListState<Long> numElementsState;
+ private DenseVector sum;
+ private DenseVector squaredSum;
+ private long numElements;
+
+ private final String featuresCol;
+
+ public ComputeMetaOperator(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public void endInput() {
+ if (numElements > 0) {
Review comment:
It throws an exception here if we remove it when there is no data on this partition.
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r834164536
##########
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:
Good point. My concern is that using`windowAll/window` may lead to a large memory consumption. As I understand, window operations need to cache the data in state. (Please correct me if I am wrong...
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r834181740
##########
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:
Good point. I also thought about this. But I didn't do the optimization here because of the following use case:
A user calls StandardScaler#fit and get a StandardScalerModel, using `withStd as false`. Then the user wants to do a inference with the trained model using `withStd as true`. In this case, it fails as there is no std in the model data.
Another reason is that computing `std` and `mean` is not that expensive comparing with IO...
What do you think?
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r834134335
##########
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:
Thanks for the comment. Let's follow the best practice :~
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r834175679
##########
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:
Using `double[] std = new double[squaredSum.size()]` needs to allocate memory for the new matrix. If the dimension of the input is high, it is worse.
We are actually writing the calculated results back to `squaredSum.values`. `std` here is a reference for better understanding.
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r834124527
##########
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:
Thanks for pointing this out. The `clone()` here is to create a new instance of this Vector for further in-place operations, since we may not want to update the vectors directly.
I think there could be two alternatives:
- How about we rename the method to `duplicate` or something like `copy`?
- Or how about we remove the `clone()` in `Vector`, but keep it in `SparseVector` and `DenseVector`, given that we are indeed cloning vectors.
What do you think?
--
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
[GitHub] [flink-ml] yunfengzhou-hub commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r834170129
##########
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:
They are indeed similar in -- both of them are composed of two vectors.
I am okay to merge them as one. My Concern is that what are we going to name the new class? Do you have some suggestions?
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r841032715
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScalerParams.java
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params for {@link StandardScaler}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface StandardScalerParams<T> extends HasFeaturesCol<T>, HasPredictionCol<T> {
Review comment:
After some offline discussion, we agree to follow the following rules:
- use `inputCol/outputCol/inputCols/outputCos` when doing feature engineering. (e.g., code under package `org.apache.flink.ml.feature`)
- use `featuresCol/predictionCol` when doing (un)supervised machine learning training (e.g., code under package `org.apache.flink.ml.classification, org.apache.flink.ml.clustering`)
--
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
[GitHub] [flink-ml] lindong28 merged pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
lindong28 merged pull request #73:
URL: https://github.com/apache/flink-ml/pull/73
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r840212185
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.java
##########
@@ -0,0 +1,288 @@
+/*
+ * 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.Arrays;
+import java.util.Collections;
+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 ComputeMetaOperator(getFeaturesCol()));
+
+ DataStream<StandardScalerModelData> modelData =
+ sumAndSquaredSumAndWeight
+ .transform(
+ "buildModel",
+ TypeInformation.of(StandardScalerModelData.class),
+ new BuildModelOperator())
+ .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 BuildModelOperator 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;
+ if (numElements > 1) {
+ for (int i = 0; i < mean.length; i++) {
+ std[i] =
+ Math.sqrt(
+ (squaredSum.values[i] - numElements * mean[i] * mean[i])
+ / (numElements - 1));
+ }
+ } else {
+ Arrays.fill(std, 0.0);
+ }
+
+ output.collect(
+ new StreamRecord<>(
+ new StandardScalerModelData(
+ Vectors.dense(mean), Vectors.dense(std))));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple3<DenseVector, DenseVector, Long>> element) {
+ Tuple3<DenseVector, DenseVector, Long> value = element.getValue();
+ if (sum == null) {
+ sum = value.f0;
+ squaredSum = value.f1;
+ numElements = value.f2;
+ } else {
+ BLAS.axpy(1, value.f0, sum);
+ BLAS.axpy(1, value.f1, squaredSum);
+ numElements += value.f2;
+ }
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
+ sumState.update(Collections.singletonList(sum));
+ squaredSumState.update(Collections.singletonList(squaredSum));
+ numElementsState.update(Collections.singletonList(numElements));
+ }
+ }
+ }
+
+ /** Computes sum, squared sum and number of elements in each partition. */
+ private static class ComputeMetaOperator
+ extends AbstractStreamOperator<Tuple3<DenseVector, DenseVector, Long>>
+ implements OneInputStreamOperator<Row, Tuple3<DenseVector, DenseVector, Long>>,
+ BoundedOneInput {
+ private ListState<DenseVector> sumState;
+ private ListState<DenseVector> squaredSumState;
+ private ListState<Long> numElementsState;
+ private DenseVector sum;
+ private DenseVector squaredSum;
+ private long numElements;
+
+ private final String featuresCol;
+
+ public ComputeMetaOperator(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public void endInput() {
+ if (numElements > 0) {
+ output.collect(new StreamRecord<>(Tuple3.of(sum, squaredSum, numElements)));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> element) throws Exception {
+ Vector feature = (Vector) element.getValue().getField(featuresCol);
+ if (sum == null) {
+ sum = new DenseVector(feature.size());
+ squaredSum = new DenseVector(feature.size());
+ }
+ BLAS.axpy(1, feature, sum);
+ BLAS.hDot(feature, feature);
+ BLAS.axpy(1, feature, squaredSum);
+ numElements++;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
Review comment:
Good point.
I agree with the behavior that when the train data is empty, we should output a `special` vector to indicate that we cannot do inference. I have updated the code to handle this case and also added one test case. (See for StandardScalerTest#testFitOnEmptyData).
For other checks about when numElements is zero or not, we probably need to keep it.
For example, when the training data is not empty, there could be one worker that contains zero data points. If we remove the check, it throws an nullpointer exception.
--
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
[GitHub] [flink-ml] lindong28 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r841012351
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScalerParams.java
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params for {@link StandardScaler}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface StandardScalerParams<T> extends HasFeaturesCol<T>, HasPredictionCol<T> {
Review comment:
Spark's `StandardScaler` uses `inputCol` and `outputCol` as its parameters. Would it be better to follow the same approach here?
Note that this `StandardScaler` is used to transform features rather than making predictions. Thus it does not seem intuitive to name its output as `predictionCol`. We can submit a followup PR to fix this problem.
`HasFeaturesCol` should probably be renamed to `HasFeatureCol` as it represents just one column.
And `HasInputCol` is more consistent with `HasOutputCol` than `HasFeatureCol` does.
##########
File path: flink-ml-lib/src/test/java/org/apache/flink/ml/feature/StandardScalerTest.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.restartstrategy.RestartStrategies;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.feature.standardscaler.StandardScaler;
+import org.apache.flink.ml.feature.standardscaler.StandardScalerModel;
+import org.apache.flink.ml.feature.standardscaler.StandardScalerModelData;
+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.util.ReadWriteUtils;
+import org.apache.flink.ml.util.StageTestUtils;
+import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.types.Row;
+
+import org.apache.commons.collections.IteratorUtils;
+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.fail;
+
+/** Tests {@link StandardScaler} and {@link StandardScalerModel}. */
+public class StandardScalerTest extends AbstractTestBase {
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+ private Table denseTable;
+
+ private final List<Row> denseInput =
+ Arrays.asList(
+ Row.of(Vectors.dense(-2.5, 9, 1)),
+ Row.of(Vectors.dense(1.4, -5, 1)),
+ Row.of(Vectors.dense(2, -1, -2)));
+
+ private final List<DenseVector> expectedResWithMean =
+ Arrays.asList(
+ Vectors.dense(-2.8, 8, 1),
+ Vectors.dense(1.1, -6, 1),
+ Vectors.dense(1.7, -2, -2));
+
+ private final List<DenseVector> expectedResWithStd =
+ Arrays.asList(
+ Vectors.dense(-1.0231819, 1.2480754, 0.5773502),
+ Vectors.dense(0.5729819, -0.6933752, 0.5773503),
+ Vectors.dense(0.8185455, -0.1386750, -1.1547005));
+
+ private final List<DenseVector> expectedResWithMeanAndStd =
+ Arrays.asList(
+ Vectors.dense(-1.1459637, 1.1094004, 0.5773503),
+ Vectors.dense(0.45020003, -0.8320503, 0.5773503),
+ Vectors.dense(0.69576368, -0.2773501, -1.1547005));
+
+ private final double[] expectedMean = new double[] {0.3, 1, 0};
+ private final double[] expectedStd = new double[] {2.4433583, 7.2111026, 1.7320508};
+ private static final double TOLERANCE = 1e-7;
+
+ @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);
+ denseTable = tEnv.fromDataStream(env.fromCollection(denseInput)).as("features");
+ }
+
+ @SuppressWarnings("unchecked")
+ private void verifyPredictionResult(
+ List<DenseVector> expectedOutput, Table output, String predictionCol) throws Exception {
+ List<Row> collectedResult =
+ IteratorUtils.toList(tEnv.toDataStream(output).executeAndCollect());
+ List<DenseVector> predictions = new ArrayList<>(collectedResult.size());
+
+ for (Row r : collectedResult) {
+ Vector vec = (Vector) r.getField(predictionCol);
+ predictions.add(vec.toDense());
+ }
+
+ assertEquals(expectedOutput.size(), predictions.size());
+
+ predictions.sort(
+ (vec1, vec2) -> {
+ int size = Math.min(vec1.size(), vec2.size());
+ for (int i = 0; i < size; i++) {
+ int cmp = Double.compare(vec1.get(i), vec2.get(i));
+ if (cmp != 0) {
+ return cmp;
+ }
+ }
+ return 0;
+ });
+
+ for (int i = 0; i < predictions.size(); i++) {
+ assertArrayEquals(expectedOutput.get(i).values, predictions.get(i).values, TOLERANCE);
+ }
+ }
+
+ @Test
+ public void testParam() {
+ StandardScaler standardScaler = new StandardScaler();
+
+ assertEquals("features", standardScaler.getFeaturesCol());
+ assertEquals(false, standardScaler.getWithMean());
+ assertEquals(true, standardScaler.getWithStd());
+ assertEquals("prediction", standardScaler.getPredictionCol());
+
+ standardScaler
+ .setFeaturesCol("test_features")
+ .setWithMean(true)
+ .setWithStd(false)
+ .setPredictionCol("test_prediction");
+
+ assertEquals("test_features", standardScaler.getFeaturesCol());
+ assertEquals(true, standardScaler.getWithMean());
+ assertEquals(false, standardScaler.getWithStd());
+ assertEquals("test_prediction", standardScaler.getPredictionCol());
+ }
+
+ @Test
+ public void testFeaturePredictionParam() {
Review comment:
nits: suppose we change the parameter to inputCol and outputCol, it might not be intuitive to name this method as `testFeaturePredictionParam`. How about changing it to `testOutputSchema()`? We can use this name consistently across other tests.
--
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
[GitHub] [flink-ml] yunfengzhou-hub commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
yunfengzhou-hub commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r836021889
##########
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:
That's a good point. A trained Model might be used in either `withStd = true` or `withStd = false` situations. I agree with the current implementation.
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r834170129
##########
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:
They are indeed similar in -- both of them are composed of two vectors.
I am okay to merge them as one. My Concern is that what are we going to name the new class? Do you have some suggestions?
--
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
[GitHub] [flink-ml] yunfengzhou-hub commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
yunfengzhou-hub commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r836355995
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.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.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.Arrays;
+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 ComputeMetaOperator(getFeaturesCol()));
+
+ DataStream<StandardScalerModelData> modelData =
+ sumAndSquaredSumAndWeight
+ .transform(
+ "buildModel",
+ TypeInformation.of(StandardScalerModelData.class),
+ new BuildModelOperator())
+ .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 BuildModelOperator 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;
+ if (numElements > 1) {
+ for (int i = 0; i < mean.length; i++) {
+ std[i] =
+ Math.sqrt(
+ (squaredSum.values[i] - numElements * mean[i] * mean[i])
+ / (numElements - 1));
+ }
+ } else {
+ Arrays.fill(std, 0.0);
+ }
+
+ output.collect(
+ new StreamRecord<>(
+ new StandardScalerModelData(
+ Vectors.dense(mean), Vectors.dense(std))));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple3<DenseVector, DenseVector, Long>> element) {
+ Tuple3<DenseVector, DenseVector, Long> value = element.getValue();
+ if (sum == null) {
+ sum = value.f0;
+ squaredSum = value.f1;
+ numElements = value.f2;
+ } else {
+ BLAS.axpy(1, value.f0, sum);
+ BLAS.axpy(1, value.f1, squaredSum);
+ numElements += value.f2;
+ }
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ sumState.clear();
+ squaredSumState.clear();
+ numElementsState.clear();
+
+ if (numElements > 0) {
+ sumState.add(sum);
+ squaredSumState.add(squaredSum);
+ numElementsState.add(numElements);
+ }
+ }
+ }
+
+ /** Computes sum, squared sum and number of elements in each partition. */
+ private static class ComputeMetaOperator
+ extends AbstractStreamOperator<Tuple3<DenseVector, DenseVector, Long>>
+ implements OneInputStreamOperator<Row, Tuple3<DenseVector, DenseVector, Long>>,
+ BoundedOneInput {
+ private ListState<DenseVector> sumState;
+ private ListState<DenseVector> squaredSumState;
+ private ListState<Long> numElementsState;
+ private DenseVector sum;
+ private DenseVector squaredSum;
+ private long numElements;
+
+ private final String featuresCol;
+
+ public ComputeMetaOperator(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public void endInput() {
+ if (numElements > 0) {
+ output.collect(new StreamRecord<>(Tuple3.of(sum, squaredSum, numElements)));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> element) throws Exception {
+ Vector feature = (Vector) element.getValue().getField(featuresCol);
+ if (sum == null) {
+ sum = new DenseVector(feature.size());
+ squaredSum = new DenseVector(feature.size());
+ }
+ BLAS.axpy(1, feature, sum);
+ BLAS.hDot(feature, feature);
+ BLAS.axpy(1, feature, squaredSum);
+ numElements++;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ sumState.clear();
+ squaredSumState.clear();
+ numElementsState.clear();
+
+ if (numElements > 0) {
+ sumState.add(sum);
Review comment:
We can do `state.update()` instead of `.clear().add()` when `numElements > 0`.
--
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
[GitHub] [flink-ml] lindong28 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
lindong28 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r837111842
##########
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) {
+ Preconditions.checkArgument(x.size() == y.size(), "Vector size mismatched.");
+ if (y instanceof DenseVector) {
+ if (x instanceof SparseVector) {
Review comment:
nits: it seems that we use both `instanceof DenseVector` and `instanceof SparseVector` in this class. Would it be slightly better to consistently use one of the two in this class?
##########
File path: flink-ml-core/src/main/java/org/apache/flink/ml/linalg/BLAS.java
##########
@@ -89,4 +111,56 @@ public static void gemv(
y.values,
1);
}
+
+ private static void axpy(double a, DenseVector x, DenseVector y) {
+ JAVA_BLAS.daxpy(x.size(), a, x.values, 1, y.values, 1);
+ }
+
+ private static void axpy(double a, SparseVector x, DenseVector y) {
+ for (int i = 0; i < x.indices.length; i++) {
+ int index = x.indices[i];
+ y.values[index] += a * x.values[i];
+ }
+ }
+
+ private static void hDot(SparseVector x, SparseVector y) {
+ int idx = 0;
+ int idy = 0;
+ while (idx < x.indices.length && idy < y.indices.length) {
+ int indexX = x.indices[idx];
+ while (idy < y.indices.length && y.indices[idy] < indexX) {
+ y.values[idy] = 0;
+ idy++;
+ }
+ if (idy < y.indices.length && y.indices[idy] == indexX) {
+ y.values[idy] *= x.values[idx];
+ idy++;
+ }
+ idx++;
+ }
+ }
+
+ private static void hDot(SparseVector x, DenseVector y) {
+ int idx = 0;
+ for (int i = 0; i < y.size(); i++) {
+ if (x.indices[idx] == i) {
Review comment:
Should we use the following check:
`if (idx < x.indices.length && x.indices[idx] == i)`
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.java
##########
@@ -0,0 +1,288 @@
+/*
+ * 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.Arrays;
+import java.util.Collections;
+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 ComputeMetaOperator(getFeaturesCol()));
+
+ DataStream<StandardScalerModelData> modelData =
+ sumAndSquaredSumAndWeight
+ .transform(
+ "buildModel",
+ TypeInformation.of(StandardScalerModelData.class),
+ new BuildModelOperator())
+ .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 BuildModelOperator 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;
+ if (numElements > 1) {
+ for (int i = 0; i < mean.length; i++) {
+ std[i] =
+ Math.sqrt(
+ (squaredSum.values[i] - numElements * mean[i] * mean[i])
+ / (numElements - 1));
+ }
+ } else {
+ Arrays.fill(std, 0.0);
+ }
+
+ output.collect(
+ new StreamRecord<>(
+ new StandardScalerModelData(
+ Vectors.dense(mean), Vectors.dense(std))));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple3<DenseVector, DenseVector, Long>> element) {
+ Tuple3<DenseVector, DenseVector, Long> value = element.getValue();
+ if (sum == null) {
+ sum = value.f0;
+ squaredSum = value.f1;
+ numElements = value.f2;
+ } else {
+ BLAS.axpy(1, value.f0, sum);
+ BLAS.axpy(1, value.f1, squaredSum);
+ numElements += value.f2;
+ }
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
+ sumState.update(Collections.singletonList(sum));
+ squaredSumState.update(Collections.singletonList(squaredSum));
+ numElementsState.update(Collections.singletonList(numElements));
+ }
+ }
+ }
+
+ /** Computes sum, squared sum and number of elements in each partition. */
+ private static class ComputeMetaOperator
+ extends AbstractStreamOperator<Tuple3<DenseVector, DenseVector, Long>>
+ implements OneInputStreamOperator<Row, Tuple3<DenseVector, DenseVector, Long>>,
+ BoundedOneInput {
+ private ListState<DenseVector> sumState;
+ private ListState<DenseVector> squaredSumState;
+ private ListState<Long> numElementsState;
+ private DenseVector sum;
+ private DenseVector squaredSum;
+ private long numElements;
+
+ private final String featuresCol;
+
+ public ComputeMetaOperator(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public void endInput() {
+ if (numElements > 0) {
+ output.collect(new StreamRecord<>(Tuple3.of(sum, squaredSum, numElements)));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> element) throws Exception {
+ Vector feature = (Vector) element.getValue().getField(featuresCol);
+ if (sum == null) {
+ sum = new DenseVector(feature.size());
+ squaredSum = new DenseVector(feature.size());
+ }
+ BLAS.axpy(1, feature, sum);
+ BLAS.hDot(feature, feature);
+ BLAS.axpy(1, feature, squaredSum);
+ numElements++;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
Review comment:
This code aims to handle the scenario where `fit(...)` is called with an empty table. Under this scenario, the current implementation would generate an empty modelDataTable.
Should we instead generate a modelTable with one element, whose value indicates that the model data can not be used for inference?
By emitting a model data table with this value, model.transform(...) could throw proper exception to indicate that it can not do inference. In comparison, if we generate a model data table with no value, model.transform(...) would have to block forever, since it will not be able to differentiate this scenario from the scenario where source is taking a long time to read the model data.
Same for other usages of `if (numElements > 0)` in this PR.
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScalerModel.java
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.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));
+ Vector output;
+ if (feature instanceof DenseVector) {
Review comment:
Would it be simpler to add the method `Vector::clone()` similar to Spark's `Vector::copy()`?
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScalerParams.java
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params for {@link StandardScaler}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface StandardScalerParams<T> extends HasFeaturesCol<T>, HasPredictionCol<T> {
+ Param<Boolean> WITH_MEAN =
+ new BooleanParam(
+ "withMean", "Whether centers the data with mean before scaling.", false);
+
+ default Boolean getWithMean() {
+ return get(WITH_MEAN);
+ }
+
+ default T setWithMean(boolean withMean) {
+ return set(WITH_MEAN, withMean);
+ }
+
+ Param<Boolean> WITH_STD =
Review comment:
Could we move the variable declaration to be before the method declarations, similar to other Java classes in Flink ML?
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/minmaxscaler/MinMaxScalerModel.java
##########
@@ -172,11 +173,9 @@ public Row map(Row row) {
}
}
DenseVector feature = (DenseVector) row.getField(featureCol);
- DenseVector outputVector = new DenseVector(scaleVector.size());
- for (int i = 0; i < scaleVector.size(); ++i) {
- outputVector.values[i] =
- feature.values[i] * scaleVector.values[i] + offsetVector.values[i];
- }
+ DenseVector outputVector = feature.clone();
+ BLAS.hDot(scaleVector, outputVector);
+ BLAS.axpy(1, offsetVector, outputVector);
Review comment:
The new implementation seems to be strictly slower than the previous implementation. Should we keep the previous implementation?
The previous implementation just needs one for loop over the the `outputVector`.
The new implementation needs two for loops (i.e. `clone()` and `hDot()`) over the `outputVector`, plus one `JAVA_BLAS.daxpy()` call.
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.java
##########
@@ -0,0 +1,288 @@
+/*
+ * 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.Arrays;
+import java.util.Collections;
+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 ComputeMetaOperator(getFeaturesCol()));
+
+ DataStream<StandardScalerModelData> modelData =
+ sumAndSquaredSumAndWeight
+ .transform(
+ "buildModel",
+ TypeInformation.of(StandardScalerModelData.class),
+ new BuildModelOperator())
+ .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 BuildModelOperator 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;
+ if (numElements > 1) {
+ for (int i = 0; i < mean.length; i++) {
+ std[i] =
+ Math.sqrt(
+ (squaredSum.values[i] - numElements * mean[i] * mean[i])
+ / (numElements - 1));
+ }
+ } else {
+ Arrays.fill(std, 0.0);
+ }
+
+ output.collect(
+ new StreamRecord<>(
+ new StandardScalerModelData(
+ Vectors.dense(mean), Vectors.dense(std))));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple3<DenseVector, DenseVector, Long>> element) {
+ Tuple3<DenseVector, DenseVector, Long> value = element.getValue();
+ if (sum == null) {
+ sum = value.f0;
+ squaredSum = value.f1;
+ numElements = value.f2;
+ } else {
+ BLAS.axpy(1, value.f0, sum);
+ BLAS.axpy(1, value.f1, squaredSum);
+ numElements += value.f2;
+ }
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
+ sumState.update(Collections.singletonList(sum));
+ squaredSumState.update(Collections.singletonList(squaredSum));
+ numElementsState.update(Collections.singletonList(numElements));
+ }
+ }
+ }
+
+ /** Computes sum, squared sum and number of elements in each partition. */
+ private static class ComputeMetaOperator
+ extends AbstractStreamOperator<Tuple3<DenseVector, DenseVector, Long>>
+ implements OneInputStreamOperator<Row, Tuple3<DenseVector, DenseVector, Long>>,
+ BoundedOneInput {
+ private ListState<DenseVector> sumState;
+ private ListState<DenseVector> squaredSumState;
+ private ListState<Long> numElementsState;
+ private DenseVector sum;
+ private DenseVector squaredSum;
+ private long numElements;
+
+ private final String featuresCol;
+
+ public ComputeMetaOperator(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public void endInput() {
+ if (numElements > 0) {
Review comment:
Would it be simpler to remove this check?
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r834170216
##########
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:
They are indeed similar in code -- both of them are composed of two vectors.
I am okay to merge them as one. My Concern is that what are we going to name the new class? Do you have some suggestions?
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r840212185
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScaler.java
##########
@@ -0,0 +1,288 @@
+/*
+ * 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.Arrays;
+import java.util.Collections;
+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 ComputeMetaOperator(getFeaturesCol()));
+
+ DataStream<StandardScalerModelData> modelData =
+ sumAndSquaredSumAndWeight
+ .transform(
+ "buildModel",
+ TypeInformation.of(StandardScalerModelData.class),
+ new BuildModelOperator())
+ .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 BuildModelOperator 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;
+ if (numElements > 1) {
+ for (int i = 0; i < mean.length; i++) {
+ std[i] =
+ Math.sqrt(
+ (squaredSum.values[i] - numElements * mean[i] * mean[i])
+ / (numElements - 1));
+ }
+ } else {
+ Arrays.fill(std, 0.0);
+ }
+
+ output.collect(
+ new StreamRecord<>(
+ new StandardScalerModelData(
+ Vectors.dense(mean), Vectors.dense(std))));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Tuple3<DenseVector, DenseVector, Long>> element) {
+ Tuple3<DenseVector, DenseVector, Long> value = element.getValue();
+ if (sum == null) {
+ sum = value.f0;
+ squaredSum = value.f1;
+ numElements = value.f2;
+ } else {
+ BLAS.axpy(1, value.f0, sum);
+ BLAS.axpy(1, value.f1, squaredSum);
+ numElements += value.f2;
+ }
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
+ sumState.update(Collections.singletonList(sum));
+ squaredSumState.update(Collections.singletonList(squaredSum));
+ numElementsState.update(Collections.singletonList(numElements));
+ }
+ }
+ }
+
+ /** Computes sum, squared sum and number of elements in each partition. */
+ private static class ComputeMetaOperator
+ extends AbstractStreamOperator<Tuple3<DenseVector, DenseVector, Long>>
+ implements OneInputStreamOperator<Row, Tuple3<DenseVector, DenseVector, Long>>,
+ BoundedOneInput {
+ private ListState<DenseVector> sumState;
+ private ListState<DenseVector> squaredSumState;
+ private ListState<Long> numElementsState;
+ private DenseVector sum;
+ private DenseVector squaredSum;
+ private long numElements;
+
+ private final String featuresCol;
+
+ public ComputeMetaOperator(String featuresCol) {
+ this.featuresCol = featuresCol;
+ }
+
+ @Override
+ public void endInput() {
+ if (numElements > 0) {
+ output.collect(new StreamRecord<>(Tuple3.of(sum, squaredSum, numElements)));
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<Row> element) throws Exception {
+ Vector feature = (Vector) element.getValue().getField(featuresCol);
+ if (sum == null) {
+ sum = new DenseVector(feature.size());
+ squaredSum = new DenseVector(feature.size());
+ }
+ BLAS.axpy(1, feature, sum);
+ BLAS.hDot(feature, feature);
+ BLAS.axpy(1, feature, squaredSum);
+ numElements++;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "sumState", TypeInformation.of(DenseVector.class)));
+ squaredSumState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "squaredSumState",
+ TypeInformation.of(DenseVector.class)));
+ numElementsState =
+ context.getOperatorStateStore()
+ .getListState(
+ new ListStateDescriptor<>(
+ "numElementsState", BasicTypeInfo.LONG_TYPE_INFO));
+
+ sum = OperatorStateUtils.getUniqueElement(sumState, "sumState").orElse(null);
+ squaredSum =
+ OperatorStateUtils.getUniqueElement(squaredSumState, "squaredSumState")
+ .orElse(null);
+ numElements =
+ OperatorStateUtils.getUniqueElement(numElementsState, "numElementsState")
+ .orElse(0L);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ if (numElements > 0) {
Review comment:
Good point.
I agree with the behavior that when the train data is empty, we should output a `special model data` to indicate that we cannot do inference. I have updated the code to handle this case and also added one test case. (See for StandardScalerTest#testFitOnEmptyData).
For other checks about when numElements is zero or not, we probably need to keep it.
For example, when the training data is not empty, there could be one worker that contains zero data points. If we remove the check, it throws an nullpointer exception.
--
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
[GitHub] [flink-ml] zhipeng93 commented on pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#issuecomment-1085405076
@lindong28 Thanks for the review. I have adressed your comments in the latest commit.
--
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
[GitHub] [flink-ml] lindong28 commented on pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
lindong28 commented on pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#issuecomment-1086582269
Thanks for the update! LGTM.
--
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
[GitHub] [flink-ml] zhipeng93 commented on a change in pull request #73: [FLINK-26626] Add Transformer and Estimator for StandardScaler
Posted by GitBox <gi...@apache.org>.
zhipeng93 commented on a change in pull request #73:
URL: https://github.com/apache/flink-ml/pull/73#discussion_r841019214
##########
File path: flink-ml-lib/src/main/java/org/apache/flink/ml/feature/standardscaler/StandardScalerParams.java
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.ml.common.param.HasFeaturesCol;
+import org.apache.flink.ml.common.param.HasPredictionCol;
+import org.apache.flink.ml.param.BooleanParam;
+import org.apache.flink.ml.param.Param;
+
+/**
+ * Params for {@link StandardScaler}.
+ *
+ * @param <T> The class type of this instance.
+ */
+public interface StandardScalerParams<T> extends HasFeaturesCol<T>, HasPredictionCol<T> {
Review comment:
Thanks for raising up this discussion.
As `StandardScaler` is an `estmator`, it aims to train on the input features and predict on the output. So I thought it might be ok to keep the current naming. So soes `MinMaxScaler`.
> HasFeaturesCol should probably be renamed to HasFeatureCol as it represents just one column.
Here `features` is one column.
--
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