lindong28 commented on a change in pull request #37:
URL: https://github.com/apache/flink-ml/pull/37#discussion_r765496493
##########
File path:
flink-ml-core/src/main/java/org/apache/flink/ml/param/ParamValidators.java
##########
@@ -95,4 +95,9 @@ public boolean validate(T value) {
}
};
}
+
+ // Check if the parameter value array is not empty array.
+ public static <T> ParamValidator<T> nonEmptyArray() {
Review comment:
Could you test this validator in `StageTest::testValidators(...)`?
##########
File path:
flink-ml-lib/src/main/java/org/apache/flink/ml/common/param/HasHandleInvalid.java
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringParam;
+import org.apache.flink.ml.param.WithParams;
+
+/** Interface for the shared handleInvalid param. */
+public interface HasHandleInvalid<T> extends WithParams<T> {
+ Param<String> HANDLE_INVALID =
+ new StringParam(
+ "handleInvalid",
+ "Strategy to handle invalid entries.",
+ "ERROR",
Review comment:
Could we use lower-case letters (i.e. `error`) as the value here for
consistency with `HasDistanceMeasure`?
##########
File path:
flink-ml-core/src/main/java/org/apache/flink/ml/linalg/SparseVector.java
##########
@@ -0,0 +1,177 @@
+/*
+ * 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.linalg;
+
+import org.apache.flink.api.common.typeinfo.TypeInfo;
+import org.apache.flink.ml.linalg.typeinfo.SparseVectorTypeInfoFactory;
+import org.apache.flink.util.Preconditions;
+
+import java.util.Arrays;
+import java.util.Objects;
+
+/** A sparse vector of double values. */
+@TypeInfo(SparseVectorTypeInfoFactory.class)
+public class SparseVector implements Vector {
+ public final int n;
+ public final int[] indices;
+ public final double[] values;
+
+ public SparseVector(int n, int[] indices, double[] values) {
+ this.n = n;
+ this.indices = indices;
+ this.values = values;
+ checkSizeAndIndicesRange();
Review comment:
Instead of making three passes over the vector (i.e.
`checkSizeAndIndicesRange`, `isIndicesSorted` and `checkDuplicatedIndices`).
Could we make just one pass to improve efficiency?
##########
File path:
flink-ml-lib/src/main/java/org/apache/flink/ml/feature/onehotencoder/OneHotEncoder.java
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.onehotencoder;
+
+import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.functions.MapPartitionFunction;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.common.datastream.MapPartitionFunctionWrapper;
+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.Collector;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * An Estimator which implements the one-hot encoding algorithm.
+ *
+ * <p>See https://en.wikipedia.org/wiki/One-hot.
+ */
+public class OneHotEncoder
+ implements Estimator<OneHotEncoder, OneHotEncoderModel>,
+ OneHotEncoderParams<OneHotEncoder> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public OneHotEncoder() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public OneHotEncoderModel fit(Table... inputs) {
+ Preconditions.checkArgument(inputs.length == 1);
+ Preconditions.checkArgument(getHandleInvalid().equals("ERROR"));
Review comment:
Instead of asking individual algorithms to specify the raw string, could
we add a `public static final String ERROR_INVALID = "error"`, similar to
`Bucketizer::ERROR_INVALID` in Spark and `EuclideanDistanceMeasure::NAME` in
Flink ML?
##########
File path:
flink-ml-core/src/test/java/org/apache/flink/ml/linalg/VectorsTest.java
##########
@@ -0,0 +1,69 @@
+/*
+ * 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.linalg;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+
+/** Tests the behavior of Vectors. */
+public class VectorsTest {
Review comment:
If this test is supposed to cover only `SparseVector`, could we name it
as `SparseVectorTest`? Otherwise, could we specify in the test method name
whether the test is for `SparseVector` or `DenseVector`?
##########
File path:
flink-ml-lib/src/main/java/org/apache/flink/ml/common/param/HasHandleInvalid.java
##########
@@ -0,0 +1,43 @@
+/*
+ * 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.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringParam;
+import org.apache.flink.ml.param.WithParams;
+
+/** Interface for the shared handleInvalid param. */
+public interface HasHandleInvalid<T> extends WithParams<T> {
Review comment:
Could we add Java doc explaining what happens when the strategy is
`ERROR`?
##########
File path:
flink-ml-lib/src/test/java/org/apache/flink/ml/feature/OneHotEncoderTest.java
##########
@@ -0,0 +1,282 @@
+/*
+ * 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.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.feature.onehotencoder.OneHotEncoder;
+import org.apache.flink.ml.feature.onehotencoder.OneHotEncoderModel;
+import org.apache.flink.ml.feature.onehotencoder.OneHotEncoderModelData;
+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.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/** Tests OneHotEncoder and OneHotEncoderModel. */
+public class OneHotEncoderTest {
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+
+ private StreamExecutionEnvironment env;
+ private StreamTableEnvironment tEnv;
+ private Table trainTable;
+ private Table predictTable;
+ private Map<Double, Vector>[] expectedOutput;
+ private OneHotEncoder estimator;
+
+ @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);
+
+ List<Row> trainData = Arrays.asList(Row.of(0.0), Row.of(1.0),
Row.of(2.0), Row.of(0.0));
+
+ trainTable =
tEnv.fromDataStream(env.fromCollection(trainData)).as("input");
+
+ List<Row> predictData = Arrays.asList(Row.of(0.0), Row.of(1.0),
Row.of(2.0));
+
+ predictTable =
tEnv.fromDataStream(env.fromCollection(predictData)).as("input");
+
+ expectedOutput =
+ new HashMap[] {
+ new HashMap<Double, Vector>() {
+ {
+ put(0.0, Vectors.sparse(2, new int[] {0}, new
double[] {1.0}));
+ put(1.0, Vectors.sparse(2, new int[] {1}, new
double[] {1.0}));
+ put(2.0, Vectors.sparse(2, new int[0], new
double[0]));
+ }
+ }
+ };
+
+ estimator = new
OneHotEncoder().setInputCols("input").setOutputCols("output");
+ }
+
+ /**
+ * Executes a given table and collect its results. Results are returned as
a map array. Each
+ * element in the array is a map corresponding to a input column whose key
is the original value
+ * in the input column, value is the one-hot encoding result of that value.
+ *
+ * @param table A table to be executed and to have its result collected
+ * @param inputCols Name of the input columns
+ * @param outputCols Name of the output columns containing one-hot
encoding result
+ * @return An array of map containing the collected results for each input
column
+ */
+ private static Map<Double, Vector>[] executeAndCollect(
+ Table table, String[] inputCols, String[] outputCols) {
+ Map<Double, Vector>[] maps = new HashMap[inputCols.length];
+ for (int i = 0; i < inputCols.length; i++) {
+ maps[i] = new HashMap<>();
+ }
+ for (CloseableIterator<Row> it = table.execute().collect();
it.hasNext(); ) {
+ Row row = it.next();
+ for (int i = 0; i < inputCols.length; i++) {
+ maps[i].put(
+ ((Number) row.getField(inputCols[i])).doubleValue(),
+ (Vector) row.getField(outputCols[i]));
+ }
+ }
+ return maps;
+ }
+
+ @Test
+ public void testParam() {
+ OneHotEncoder estimator = new OneHotEncoder();
+
+ assertTrue(estimator.getDropLast());
+
+
estimator.setInputCols("test_input").setOutputCols("test_output").setDropLast(false);
+
+ assertArrayEquals(new String[] {"test_input"},
estimator.getInputCols());
+ assertArrayEquals(new String[] {"test_output"},
estimator.getOutputCols());
+ assertFalse(estimator.getDropLast());
+
+ OneHotEncoderModel model = new OneHotEncoderModel();
+
+ assertTrue(model.getDropLast());
+
+
model.setInputCols("test_input").setOutputCols("test_output").setDropLast(false);
+
+ assertArrayEquals(new String[] {"test_input"}, model.getInputCols());
+ assertArrayEquals(new String[] {"test_output"}, model.getOutputCols());
+ assertFalse(model.getDropLast());
+ }
+
+ @Test
+ public void testFitAndPredict() {
+ OneHotEncoderModel model = estimator.fit(trainTable);
+ Table outputTable = model.transform(predictTable)[0];
+ Map<Double, Vector>[] actualOutput =
+ executeAndCollect(outputTable, model.getInputCols(),
model.getOutputCols());
+ assertArrayEquals(expectedOutput, actualOutput);
+ }
+
+ @Test
+ public void testDropLast() {
+ estimator.setDropLast(false);
+
+ expectedOutput =
+ new HashMap[] {
+ new HashMap<Double, Vector>() {
+ {
+ put(0.0, Vectors.sparse(3, new int[] {0}, new
double[] {1.0}));
+ put(1.0, Vectors.sparse(3, new int[] {1}, new
double[] {1.0}));
+ put(2.0, Vectors.sparse(3, new int[] {2}, new
double[] {1.0}));
+ }
+ }
+ };
+
+ OneHotEncoderModel model = estimator.fit(trainTable);
+ Table outputTable = model.transform(predictTable)[0];
+ Map<Double, Vector>[] actualOutput =
+ executeAndCollect(outputTable, model.getInputCols(),
model.getOutputCols());
+ assertArrayEquals(expectedOutput, actualOutput);
+ }
+
+ @Test
+ public void testInputDataType() {
+ List<Row> trainData = Arrays.asList(Row.of(0), Row.of(1), Row.of(2),
Row.of(0));
+
+ trainTable =
tEnv.fromDataStream(env.fromCollection(trainData)).as("input");
+
+ List<Row> predictData = Arrays.asList(Row.of(0), Row.of(1), Row.of(2));
+ predictTable =
tEnv.fromDataStream(env.fromCollection(predictData)).as("input");
+
+ expectedOutput =
+ new HashMap[] {
+ new HashMap<Double, Vector>() {
+ {
+ put(0.0, Vectors.sparse(2, new int[] {0}, new
double[] {1.0}));
+ put(1.0, Vectors.sparse(2, new int[] {1}, new
double[] {1.0}));
+ put(2.0, Vectors.sparse(2, new int[0], new
double[0]));
+ }
+ }
+ };
+
+ OneHotEncoderModel model = estimator.fit(trainTable);
+ Table outputTable = model.transform(predictTable)[0];
+ Map<Double, Vector>[] actualOutput =
+ executeAndCollect(outputTable, model.getInputCols(),
model.getOutputCols());
+ assertArrayEquals(expectedOutput, actualOutput);
+ }
+
+ @Test
+ public void testNonIntegerDouble() {
+ List<Row> trainData = Arrays.asList(Row.of(0.5), Row.of(1.0),
Row.of(2.0), Row.of(0.0));
+
+ trainTable =
tEnv.fromDataStream(env.fromCollection(trainData)).as("input");
+ OneHotEncoderModel model = estimator.fit(trainTable);
+ Table outputTable = model.transform(predictTable)[0];
+ try {
+ outputTable.execute().collect().next();
+ Assert.fail("Expected IllegalArgumentException");
+ } catch (Exception e) {
+ Throwable exception = e;
+ while (exception.getCause() != null) {
+ exception = exception.getCause();
+ }
+ assertEquals(IllegalArgumentException.class, exception.getClass());
+ assertEquals("Value 0.5 cannot be parsed as indexed integer.",
exception.getMessage());
+ }
+ }
+
+ @Test
+ public void testNonIntegerDouble2() {
Review comment:
Could we make the name more self-explanatory regarding its difference
with `testNonIntegerDouble`?
Maybe something like `testIllegalTrainData` and `testIllegalPredictData`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
