lindong28 commented on a change in pull request #28: URL: https://github.com/apache/flink-ml/pull/28#discussion_r758296725
########## File path: flink-ml-lib/src/test/java/org/apache/flink/ml/classification/linear/LogisticRegressionTest.java ########## @@ -0,0 +1,277 @@ +/* + * 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.classification.linear; + +import org.apache.flink.api.common.restartstrategy.RestartStrategies; +import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.api.java.typeutils.RowTypeInfo; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.ml.util.ReadWriteUtils; +import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.functions.sink.SinkFunction; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.types.Row; + +import org.junit.Before; +import org.junit.Test; + +import java.nio.file.Files; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +/** Tests {@link LogisticRegression} and {@link LogisticRegressionModel}. */ +public class LogisticRegressionTest { + + private StreamExecutionEnvironment env; + + private StreamTableEnvironment tEnv; + + private static List<Row> trainData = + Arrays.asList( + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.)); + + private static double[] expectedCoefficient = new double[] {-0.67, 0.21, 0.32, 0.43}; Review comment: nits: `private static final` ########## File path: flink-ml-lib/src/test/java/org/apache/flink/ml/classification/linear/LogisticRegressionTest.java ########## @@ -0,0 +1,277 @@ +/* + * 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.classification.linear; + +import org.apache.flink.api.common.restartstrategy.RestartStrategies; +import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.api.java.typeutils.RowTypeInfo; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.ml.util.ReadWriteUtils; +import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.functions.sink.SinkFunction; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.types.Row; + +import org.junit.Before; +import org.junit.Test; + +import java.nio.file.Files; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +/** Tests {@link LogisticRegression} and {@link LogisticRegressionModel}. */ +public class LogisticRegressionTest { + + private StreamExecutionEnvironment env; + + private StreamTableEnvironment tEnv; + + private static List<Row> trainData = + Arrays.asList( + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.)); + + private static double[] expectedCoefficient = new double[] {-0.67, 0.21, 0.32, 0.43}; + + private static final double TOLERANCE = 1e-7; + + private Table dataTable; + + @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); + Collections.shuffle(trainData); + dataTable = + tEnv.fromDataStream( + env.fromCollection( + trainData, + new RowTypeInfo( + new TypeInformation[] { + PrimitiveArrayTypeInfo.DOUBLE_PRIMITIVE_ARRAY_TYPE_INFO, + Types.DOUBLE, + Types.DOUBLE + }, + new String[] {"features", "label", "weight"}))); + } + + private void verifyPredictionResult( + Table output, String featuresCol, String predCol, String rawPred) throws Exception { + tEnv.toDataStream(output) + .addSink( + new SinkFunction<Row>() { + @Override + public void invoke(Row value, Context context) { + double[] feature = + Objects.requireNonNull( Review comment: Can we remove `Objects.requireNonNull` to simplify the code here? The statement is also a bit too long. The following code may be a bit more readable. ``` @SuppressWarnings("ConstantConditions") double pred = (double) value.getField(predCol); @SuppressWarnings("ConstantConditions") double rawPred = ((double[]) value.getField(rawPredCol))[0]; if (Math.abs(feature[0] - 1) < TOLERANCE) { assertEquals( -1, pred, TOLERANCE); assertTrue(rawPred > 0.5); } ``` Instead of using `datastream.sink(...)`, could we use `stream.executeAndCollect(...)` similar to `KMeansTest`? The main reason is that the test code using `executeAndCollect` is closer to what algorithm users will see. Basically the test code will get a list of values representing outputs, and then compare these outputs with the expected outputs (as a list of rows). ########## File path: flink-ml-core/src/main/java/org/apache/flink/ml/common/utils/ComparatorAdapter.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.utils; + +import org.apache.flink.api.common.typeutils.TypeComparator; + +import java.util.Comparator; + +/** Utility class to convert {@link TypeComparator} to a {@link Comparator}. */ +public class ComparatorAdapter { + + /** + * Adapts a {@link TypeComparator} to a {@link Comparator}. + * + * @param typeComparator The input typeComparator. + * @param <T> The class type of the input element. + * @return The converted comparator. + */ + public static <T> Comparator<T> getComparator(TypeComparator<T> typeComparator) { Review comment: This method is only called in one place for now and its body has just one line. It seems a bit overkill to have one class/file/method for this one line. ########## File path: flink-ml-lib/src/main/java/org/apache/flink/ml/common/param/HasWeightCol.java ########## @@ -0,0 +1,36 @@ +/* + * 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.StringParam; +import org.apache.flink.ml.param.WithParams; + +/** Interface for the shared weight column param. */ +public interface HasWeightCol<T> extends WithParams<T> { + Param<String> WEIGHT_COL = new StringParam("weightCol", "Weight column name.", null); Review comment: Would it be better to mention in the doc that `If this is not set or empty, we treat all instance weights as 1.0`? ########## File path: flink-ml-lib/src/main/java/org/apache/flink/ml/common/param/HasLabelCol.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.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 label column param. */ +public interface HasLabelCol<T> extends WithParams<T> { + Param<String> LABEL_COL = + new StringParam("labelCol", "Label column name.", null, ParamValidators.notNull()); Review comment: Could we set its default value to be `label`, similar to how we set the default value of `featuresCol` and `predictionCol`? ########## File path: flink-ml-lib/src/main/java/org/apache/flink/ml/common/param/HasTol.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.DoubleParam; +import org.apache.flink.ml.param.Param; +import org.apache.flink.ml.param.ParamValidators; +import org.apache.flink.ml.param.WithParams; + +/** Interface for the shared tolerance param. */ +public interface HasTol<T> extends WithParams<T> { + + Param<Double> TOL = + new DoubleParam( + "tol", + "Convergence tolerance for iterative algorithms.", + 0., Review comment: Should we default `tol` to some small non-zero number that is typically used to compare double values, e.g. `1e-6`? It is probably reasonable to consider the iteration has converged if the loss/accuracy differs by only a very small epsilon. Spark does this as well. ########## File path: flink-ml-lib/src/test/java/org/apache/flink/ml/classification/linear/LogisticRegressionTest.java ########## @@ -0,0 +1,277 @@ +/* + * 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.classification.linear; + +import org.apache.flink.api.common.restartstrategy.RestartStrategies; +import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.api.java.typeutils.RowTypeInfo; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.ml.util.ReadWriteUtils; +import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.functions.sink.SinkFunction; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.types.Row; + +import org.junit.Before; +import org.junit.Test; + +import java.nio.file.Files; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +/** Tests {@link LogisticRegression} and {@link LogisticRegressionModel}. */ +public class LogisticRegressionTest { + + private StreamExecutionEnvironment env; + + private StreamTableEnvironment tEnv; + + private static List<Row> trainData = Review comment: nits: `private static final` ########## File path: flink-ml-lib/src/main/java/org/apache/flink/ml/common/param/HasRawPredictionCol.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.Param; +import org.apache.flink.ml.param.StringParam; +import org.apache.flink.ml.param.WithParams; + +/** Interface for the shared raw prediction column param. */ +public interface HasRawPredictionCol<T> extends WithParams<T> { + + Param<String> RAW_PREDICTION_COL = Review comment: Should we set its default value to `rawPrediction` so that this is always provided by default? This information is probably useful to know during development/tuning phase. ########## File path: flink-ml-lib/src/test/java/org/apache/flink/ml/classification/linear/LogisticRegressionTest.java ########## @@ -0,0 +1,277 @@ +/* + * 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.classification.linear; + +import org.apache.flink.api.common.restartstrategy.RestartStrategies; +import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.api.java.typeutils.RowTypeInfo; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.ml.util.ReadWriteUtils; +import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.functions.sink.SinkFunction; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.types.Row; + +import org.junit.Before; +import org.junit.Test; + +import java.nio.file.Files; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Objects; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +/** Tests {@link LogisticRegression} and {@link LogisticRegressionModel}. */ +public class LogisticRegressionTest { + + private StreamExecutionEnvironment env; + + private StreamTableEnvironment tEnv; + + private static List<Row> trainData = + Arrays.asList( + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {1, 2, 3, 4}, -1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.), + Row.of(new double[] {10, 2, 3, 4}, 1., 1.)); + + private static double[] expectedCoefficient = new double[] {-0.67, 0.21, 0.32, 0.43}; + + private static final double TOLERANCE = 1e-7; + + private Table dataTable; + + @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); + Collections.shuffle(trainData); + dataTable = + tEnv.fromDataStream( + env.fromCollection( + trainData, + new RowTypeInfo( + new TypeInformation[] { + PrimitiveArrayTypeInfo.DOUBLE_PRIMITIVE_ARRAY_TYPE_INFO, + Types.DOUBLE, + Types.DOUBLE + }, + new String[] {"features", "label", "weight"}))); + } + + private void verifyPredictionResult( + Table output, String featuresCol, String predCol, String rawPred) throws Exception { Review comment: nits: `rawPred` -> `rawPredCol` -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
