lindong28 commented on code in PR #83: URL: https://github.com/apache/flink-ml/pull/83#discussion_r884764718
########## flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/OnlineLogisticRegressionParams.java: ########## @@ -0,0 +1,66 @@ +/* + * 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.logisticregression; + +import org.apache.flink.ml.common.param.HasBatchStrategy; +import org.apache.flink.ml.common.param.HasElasticNet; +import org.apache.flink.ml.common.param.HasGlobalBatchSize; +import org.apache.flink.ml.common.param.HasLabelCol; +import org.apache.flink.ml.common.param.HasReg; +import org.apache.flink.ml.common.param.HasWeightCol; +import org.apache.flink.ml.param.DoubleParam; +import org.apache.flink.ml.param.Param; +import org.apache.flink.ml.param.ParamValidators; + +/** + * Params of {@link OnlineLogisticRegression}. + * + * @param <T> The class type of this instance. + */ +public interface OnlineLogisticRegressionParams<T> + extends HasLabelCol<T>, + HasWeightCol<T>, + HasBatchStrategy<T>, + HasGlobalBatchSize<T>, + HasReg<T>, + HasElasticNet<T>, + OnlineLogisticRegressionModelParams<T> { + + Param<Double> ALPHA = + new DoubleParam("alpha", "The parameter alpha of ftrl.", 0.1, ParamValidators.gt(0.0)); + + Param<Double> BETA = + new DoubleParam("alpha", "The parameter beta of ftrl.", 0.1, ParamValidators.gt(0.0)); Review Comment: alpha -> beta. ########## flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/OnlineLogisticRegressionModel.java: ########## @@ -0,0 +1,202 @@ +/* + * 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.logisticregression; + +import org.apache.flink.api.common.state.ListState; +import org.apache.flink.api.common.state.ListStateDescriptor; +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.metrics.Gauge; +import org.apache.flink.ml.api.Model; +import org.apache.flink.ml.common.datastream.TableUtils; +import org.apache.flink.ml.linalg.DenseVector; +import org.apache.flink.ml.linalg.Vector; +import org.apache.flink.ml.param.Param; +import org.apache.flink.ml.util.ParamUtils; +import org.apache.flink.ml.util.ReadWriteUtils; +import org.apache.flink.runtime.state.StateInitializationContext; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.TwoInputStreamOperator; +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 org.apache.commons.lang3.ArrayUtils; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +import static org.apache.flink.ml.classification.logisticregression.LogisticRegressionModel.predictOneDataPoint; + +/** + * A Model which classifies data using the model data computed by {@link OnlineLogisticRegression}. + */ +public class OnlineLogisticRegressionModel + implements Model<OnlineLogisticRegressionModel>, + OnlineLogisticRegressionModelParams<OnlineLogisticRegressionModel> { + public static final String MODEL_DATA_VERSION_GAUGE_KEY = "modelDataVersion"; + private final Map<Param<?>, Object> paramMap = new HashMap<>(); + private Table modelDataTable; + + public OnlineLogisticRegressionModel() { + ParamUtils.initializeMapWithDefaultValues(paramMap, this); + } + + @Override + public Table[] transform(Table... inputs) { + Preconditions.checkArgument(inputs.length == 1); + + StreamTableEnvironment tEnv = + (StreamTableEnvironment) ((TableImpl) inputs[0]).getTableEnvironment(); + + RowTypeInfo inputTypeInfo = TableUtils.getRowTypeInfo(inputs[0].getResolvedSchema()); + RowTypeInfo outputTypeInfo = + new RowTypeInfo( + ArrayUtils.addAll( + inputTypeInfo.getFieldTypes(), + Types.DOUBLE, + TypeInformation.of(DenseVector.class), + Types.LONG), + ArrayUtils.addAll( + inputTypeInfo.getFieldNames(), + getPredictionCol(), + getRawPredictionCol(), + getModelVersionCol())); + + DataStream<Row> predictionResult = + tEnv.toDataStream(inputs[0]) + .connect( + LogisticRegressionModelData.getModelDataStream(modelDataTable) + .broadcast()) + .transform( + "PredictLabelOperator", + outputTypeInfo, + new PredictLabelOperator(inputTypeInfo, getFeaturesCol())); + + return new Table[] {tEnv.fromDataStream(predictionResult)}; + } + + /** A utility operator used for prediction. */ + private static class PredictLabelOperator extends AbstractStreamOperator<Row> + implements TwoInputStreamOperator<Row, LogisticRegressionModelData, Row> { + private final RowTypeInfo inputTypeInfo; + + private final String featuresCol; + private ListState<Row> bufferedPointsState; + private DenseVector coefficient; + private long modelDataVersion = 0L; + + public PredictLabelOperator(RowTypeInfo inputTypeInfo, String featuresCol) { + this.inputTypeInfo = inputTypeInfo; + this.featuresCol = featuresCol; + } + + @Override + public void initializeState(StateInitializationContext context) throws Exception { + super.initializeState(context); + bufferedPointsState = + context.getOperatorStateStore() + .getListState( + new ListStateDescriptor<>("bufferedPoints", inputTypeInfo)); + } + + @Override + public void open() throws Exception { + super.open(); + + getRuntimeContext() + .getMetricGroup() + .gauge( + MODEL_DATA_VERSION_GAUGE_KEY, + (Gauge<String>) () -> Long.toString(modelDataVersion)); + } + + @Override + public void processElement1(StreamRecord<Row> streamRecord) throws Exception { + processElement(streamRecord); + } + + @Override + public void processElement2(StreamRecord<LogisticRegressionModelData> streamRecord) + throws Exception { + LogisticRegressionModelData modelData = streamRecord.getValue(); + coefficient = modelData.coefficient; + modelDataVersion = modelData.modelVersion; + for (Row dataPoint : bufferedPointsState.get()) { + processElement(new StreamRecord<>(dataPoint)); + } + bufferedPointsState.clear(); + } + + public void processElement(StreamRecord<Row> streamRecord) throws Exception { + Row dataPoint = streamRecord.getValue(); + if (coefficient == null) { + bufferedPointsState.add(dataPoint); + return; + } + Vector features = (Vector) dataPoint.getField(featuresCol); + Row predictionResult = predictOneDataPoint(features, coefficient); + output.collect( + new StreamRecord<>( + Row.join( + dataPoint, + Row.of( + predictionResult.getField(0), + predictionResult.getField(1), + modelDataVersion)))); + } + } + + @Override + public void save(String path) throws IOException { + ReadWriteUtils.saveMetadata(this, path); + } + + public static OnlineLogisticRegressionModel load(StreamTableEnvironment tEnv, String path) + throws IOException { + OnlineLogisticRegressionModel model = ReadWriteUtils.loadStageParam(path); + Table modelDataTable = Review Comment: Given that `OnlineLogisticRegressionModel::save(...)` does not write model data to disk, it seems inconsistent to have `OnlineLogisticRegressionModel::load(...)` read model data from the given path. Should we follow the same behavior of `OnlineKMeansModel::load(...)` and only read metadata in this mehod? ########## flink-ml-lib/src/test/java/org/apache/flink/ml/classification/OnlineLogisticRegressionTest.java: ########## @@ -0,0 +1,687 @@ +/* + * 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; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.JobSubmissionResult; +import org.apache.flink.api.common.restartstrategy.RestartStrategies; +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.configuration.RestOptions; +import org.apache.flink.metrics.Gauge; +import org.apache.flink.ml.classification.logisticregression.LogisticRegression; +import org.apache.flink.ml.classification.logisticregression.LogisticRegressionModel; +import org.apache.flink.ml.classification.logisticregression.LogisticRegressionModelData; +import org.apache.flink.ml.classification.logisticregression.OnlineLogisticRegression; +import org.apache.flink.ml.classification.logisticregression.OnlineLogisticRegressionModel; +import org.apache.flink.ml.linalg.DenseVector; +import org.apache.flink.ml.linalg.SparseVector; +import org.apache.flink.ml.linalg.Vectors; +import org.apache.flink.ml.util.InMemorySinkFunction; +import org.apache.flink.ml.util.InMemorySourceFunction; +import org.apache.flink.ml.util.TestUtils; +import org.apache.flink.runtime.jobgraph.JobGraph; +import org.apache.flink.runtime.minicluster.MiniCluster; +import org.apache.flink.runtime.minicluster.MiniClusterConfiguration; +import org.apache.flink.runtime.testutils.InMemoryReporter; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.Table; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.types.Row; +import org.apache.flink.util.TestLogger; + +import org.apache.commons.collections.IteratorUtils; +import org.junit.After; +import org.junit.Assert; +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 java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import static org.apache.flink.ml.classification.logisticregression.OnlineLogisticRegressionModel.MODEL_DATA_VERSION_GAUGE_KEY; + +/** Tests {@link OnlineLogisticRegression} and {@link OnlineLogisticRegressionModel}. */ +public class OnlineLogisticRegressionTest extends TestLogger { + @Rule public final TemporaryFolder tempFolder = new TemporaryFolder(); + + private static final double[] ONE_ARRAY = new double[] {1.0, 1.0, 1.0}; + + private static final Row[] TRAIN_DENSE_ROWS_1 = + new Row[] { + Row.of(Vectors.dense(0.1, 2.), 0.), + Row.of(Vectors.dense(0.2, 2.), 0.), + Row.of(Vectors.dense(0.3, 2.), 0.), + Row.of(Vectors.dense(0.4, 2.), 0.), + Row.of(Vectors.dense(0.5, 2.), 0.), + Row.of(Vectors.dense(11., 12.), 1.), + Row.of(Vectors.dense(12., 11.), 1.), + Row.of(Vectors.dense(13., 12.), 1.), + Row.of(Vectors.dense(14., 12.), 1.), + Row.of(Vectors.dense(15., 12.), 1.) + }; + + private static final Row[] TRAIN_DENSE_ROWS_2 = + new Row[] { + Row.of(Vectors.dense(0.2, 3.), 0.), + Row.of(Vectors.dense(0.8, 1.), 0.), + Row.of(Vectors.dense(0.7, 1.), 0.), + Row.of(Vectors.dense(0.6, 2.), 0.), + Row.of(Vectors.dense(0.2, 2.), 0.), + Row.of(Vectors.dense(14., 17.), 1.), + Row.of(Vectors.dense(15., 10.), 1.), + Row.of(Vectors.dense(16., 16.), 1.), + Row.of(Vectors.dense(17., 10.), 1.), + Row.of(Vectors.dense(18., 13.), 1.) + }; + + private static final Row[] PREDICT_DENSE_ROWS = + new Row[] { + Row.of(Vectors.dense(0.8, 2.7), 0.0), Row.of(Vectors.dense(15.5, 11.2), 1.0) + }; + + private static final Row[] TRAIN_SPARSE_ROWS_1 = + new Row[] { + Row.of(Vectors.sparse(10, new int[] {1, 3, 4}, ONE_ARRAY), 0., 1.0), + Row.of(Vectors.sparse(10, new int[] {0, 2, 3}, ONE_ARRAY), 0., 1.4), + Row.of(Vectors.sparse(10, new int[] {0, 3, 4}, ONE_ARRAY), 0., 1.3), + Row.of(Vectors.sparse(10, new int[] {2, 3, 4}, ONE_ARRAY), 0., 1.4), + Row.of(Vectors.sparse(10, new int[] {1, 3, 4}, ONE_ARRAY), 0., 1.6), + Row.of(Vectors.sparse(10, new int[] {6, 7, 8}, ONE_ARRAY), 1., 1.8), + Row.of(Vectors.sparse(10, new int[] {6, 8, 9}, ONE_ARRAY), 1., 1.9), + Row.of(Vectors.sparse(10, new int[] {5, 8, 9}, ONE_ARRAY), 1., 1.0), + Row.of(Vectors.sparse(10, new int[] {5, 6, 7}, ONE_ARRAY), 1., 1.1) + }; + + private static final Row[] TRAIN_SPARSE_ROWS_2 = + new Row[] { + Row.of(Vectors.sparse(10, new int[] {1, 2, 4}, ONE_ARRAY), 0., 1.0), + Row.of(Vectors.sparse(10, new int[] {2, 3, 4}, ONE_ARRAY), 0., 1.3), + Row.of(Vectors.sparse(10, new int[] {0, 2, 4}, ONE_ARRAY), 0., 1.4), + Row.of(Vectors.sparse(10, new int[] {1, 3, 4}, ONE_ARRAY), 0., 1.0), + Row.of(Vectors.sparse(10, new int[] {6, 7, 9}, ONE_ARRAY), 1., 1.6), + Row.of(Vectors.sparse(10, new int[] {7, 8, 9}, ONE_ARRAY), 1., 1.8), + Row.of(Vectors.sparse(10, new int[] {5, 7, 9}, ONE_ARRAY), 1., 1.0), + Row.of(Vectors.sparse(10, new int[] {5, 6, 7}, ONE_ARRAY), 1., 1.5), + Row.of(Vectors.sparse(10, new int[] {5, 8, 9}, ONE_ARRAY), 1., 1.0) + }; + + private static final Row[] PREDICT_SPARSE_ROWS = + new Row[] { + Row.of(Vectors.sparse(10, new int[] {1, 3, 5}, ONE_ARRAY), 0.), + Row.of(Vectors.sparse(10, new int[] {5, 8, 9}, ONE_ARRAY), 1.) + }; + + private static final int defaultParallelism = 4; + private static final int numTaskManagers = 2; + private static final int numSlotsPerTaskManager = 2; + + private long currentModelDataVersion; + + private InMemorySourceFunction<Row> trainDenseSource; + private InMemorySourceFunction<Row> predictDenseSource; + private InMemorySourceFunction<Row> trainSparseSource; + private InMemorySourceFunction<Row> predictSparseSource; + private InMemorySinkFunction<Row> outputSink; + private InMemorySinkFunction<LogisticRegressionModelData> modelDataSink; + + // TODO: creates static mini cluster once for whole test class after dependency upgrades to + // Flink 1.15. + private InMemoryReporter reporter; + private MiniCluster miniCluster; + private StreamExecutionEnvironment env; + private StreamTableEnvironment tEnv; + + private Table offlineTrainDenseTable; + private Table onlineTrainDenseTable; + private Table onlinePredictDenseTable; + private Table onlineTrainSparseTable; + private Table onlinePredictSparseTable; + private Table initDenseModel; + private Table initSparseModel; + + @Before + public void before() throws Exception { + currentModelDataVersion = 0; + + trainDenseSource = new InMemorySourceFunction<>(); + predictDenseSource = new InMemorySourceFunction<>(); + trainSparseSource = new InMemorySourceFunction<>(); + predictSparseSource = new InMemorySourceFunction<>(); + outputSink = new InMemorySinkFunction<>(); + modelDataSink = new InMemorySinkFunction<>(); + + Configuration config = new Configuration(); + config.set(RestOptions.BIND_PORT, "18081-19091"); + config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true); + reporter = InMemoryReporter.create(); + reporter.addToConfiguration(config); + + miniCluster = + new MiniCluster( + new MiniClusterConfiguration.Builder() + .setConfiguration(config) + .setNumTaskManagers(numTaskManagers) + .setNumSlotsPerTaskManager(numSlotsPerTaskManager) + .build()); + miniCluster.start(); + + env = StreamExecutionEnvironment.getExecutionEnvironment(config); + env.setParallelism(defaultParallelism); + env.enableCheckpointing(100); + env.setRestartStrategy(RestartStrategies.noRestart()); + tEnv = StreamTableEnvironment.create(env); + + offlineTrainDenseTable = + tEnv.fromDataStream(env.fromElements(TRAIN_DENSE_ROWS_1)).as("features", "label"); + onlineTrainDenseTable = + tEnv.fromDataStream( + env.addSource( + trainDenseSource, + new RowTypeInfo( + new TypeInformation[] { + TypeInformation.of(DenseVector.class), Types.DOUBLE + }, + new String[] {"features", "label"}))); + + onlinePredictDenseTable = + tEnv.fromDataStream( + env.addSource( + predictDenseSource, + new RowTypeInfo( + new TypeInformation[] { + TypeInformation.of(DenseVector.class), Types.DOUBLE + }, + new String[] {"features", "label"}))); + + onlineTrainSparseTable = + tEnv.fromDataStream( + env.addSource( + trainSparseSource, + new RowTypeInfo( + new TypeInformation[] { + TypeInformation.of(SparseVector.class), + Types.DOUBLE, + Types.DOUBLE + }, + new String[] {"features", "label", "weight"}))); + + onlinePredictSparseTable = + tEnv.fromDataStream( + env.addSource( + predictSparseSource, + new RowTypeInfo( + new TypeInformation[] { + TypeInformation.of(SparseVector.class), Types.DOUBLE + }, + new String[] {"features", "label"}))); + + initDenseModel = + tEnv.fromDataStream( + env.fromElements( + Row.of( + new DenseVector( + new double[] { + 0.41233679404769874, -0.18088118293232122 + }), + 0L))); + initSparseModel = + tEnv.fromDataStream( + env.fromElements( + Row.of( + new DenseVector( + new double[] { + 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, + 0.01, 0.01 + }), + 0L))); + } + + @After + public void after() throws Exception { + miniCluster.close(); + } + + /** + * Performs transform() on the provided model with predictTable, and adds sinks for + * OnlineLogisticRegressionModel's transform output and model data. + */ + private void transformAndOutputData( + OnlineLogisticRegressionModel onlineModel, boolean isSparse) { + Table outputTable = + onlineModel + .transform(isSparse ? onlinePredictSparseTable : onlinePredictDenseTable)[ + 0]; + tEnv.toDataStream(outputTable).addSink(outputSink); + + Table modelDataTable = onlineModel.getModelData()[0]; + LogisticRegressionModelData.getModelDataStream(modelDataTable).addSink(modelDataSink); + } + + /** Blocks the thread until Model has set up init model data. */ + private void waitInitModelDataSetup(JobID jobID) throws InterruptedException { + while (reporter.findMetrics(jobID, MODEL_DATA_VERSION_GAUGE_KEY).size() + < defaultParallelism) { + Thread.sleep(100); + } + waitModelDataUpdate(jobID); + } + + /** Blocks the thread until the Model has received the next model-data-update event. */ + @SuppressWarnings("unchecked") + private void waitModelDataUpdate(JobID jobID) throws InterruptedException { + do { + long tmpModelDataVersion = + reporter.findMetrics(jobID, MODEL_DATA_VERSION_GAUGE_KEY).values().stream() + .map(x -> Long.parseLong(((Gauge<String>) x).getValue())) + .min(Long::compareTo) + .get(); + if (tmpModelDataVersion == currentModelDataVersion) { + Thread.sleep(100); + } else { + currentModelDataVersion = tmpModelDataVersion; + break; + } + } while (true); + } + + /** + * Inserts default predict data to the predict queue, fetches the prediction results, and + * asserts that the grouping result is as expected. + * + * @param expectedRawInfo A list containing sets of expected result RawInfo. + */ + private void predictAndAssert(List<DenseVector> expectedRawInfo, boolean isSparse) + throws Exception { + if (isSparse) { + predictSparseSource.addAll(PREDICT_SPARSE_ROWS); + } else { + predictDenseSource.addAll(PREDICT_DENSE_ROWS); + } + List<Row> rawResult = + outputSink.poll(isSparse ? PREDICT_SPARSE_ROWS.length : PREDICT_DENSE_ROWS.length); + List<DenseVector> resultDetail = new ArrayList<>(rawResult.size()); + for (Row row : rawResult) { + resultDetail.add(row.getFieldAs(3)); + } + resultDetail.sort(TestUtils::compare); + expectedRawInfo.sort(TestUtils::compare); + for (int i = 0; i < resultDetail.size(); ++i) { + double[] realData = resultDetail.get(i).values; + double[] expectedData = expectedRawInfo.get(i).values; + for (int j = 0; j < expectedData.length; ++j) { + Assert.assertEquals(realData[j], expectedData[j], 1.0e-5); + } + } + } + + private JobID submitJob(JobGraph jobGraph) + throws ExecutionException, InterruptedException, TimeoutException { + return miniCluster + .submitJob(jobGraph) + .thenApply(JobSubmissionResult::getJobID) + .get(1, TimeUnit.SECONDS); + } + + @Test + public void testParam() { + OnlineLogisticRegression onlineLogisticRegression = new OnlineLogisticRegression(); + Assert.assertEquals("features", onlineLogisticRegression.getFeaturesCol()); + Assert.assertEquals("count", onlineLogisticRegression.getBatchStrategy()); + Assert.assertEquals("label", onlineLogisticRegression.getLabelCol()); + Assert.assertEquals(0.0, onlineLogisticRegression.getReg(), 1.0e-5); + Assert.assertEquals(0.0, onlineLogisticRegression.getElasticNet(), 1.0e-5); + Assert.assertEquals(0.1, onlineLogisticRegression.getAlpha(), 1.0e-5); + Assert.assertEquals(0.1, onlineLogisticRegression.getBeta(), 1.0e-5); + Assert.assertEquals(32, onlineLogisticRegression.getGlobalBatchSize()); + + onlineLogisticRegression + .setFeaturesCol("test_feature") + .setLabelCol("test_label") + .setGlobalBatchSize(5) + .setReg(0.25) + .setElasticNet(0.25) Review Comment: Should we use different values for these parameters so that we can capture the bugs mentioned in the other comment? Same for the python tests. ########## flink-ml-lib/src/main/java/org/apache/flink/ml/classification/logisticregression/OnlineLogisticRegressionParams.java: ########## @@ -0,0 +1,66 @@ +/* + * 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.logisticregression; + +import org.apache.flink.ml.common.param.HasBatchStrategy; +import org.apache.flink.ml.common.param.HasElasticNet; +import org.apache.flink.ml.common.param.HasGlobalBatchSize; +import org.apache.flink.ml.common.param.HasLabelCol; +import org.apache.flink.ml.common.param.HasReg; +import org.apache.flink.ml.common.param.HasWeightCol; +import org.apache.flink.ml.param.DoubleParam; +import org.apache.flink.ml.param.Param; +import org.apache.flink.ml.param.ParamValidators; + +/** + * Params of {@link OnlineLogisticRegression}. + * + * @param <T> The class type of this instance. + */ +public interface OnlineLogisticRegressionParams<T> + extends HasLabelCol<T>, + HasWeightCol<T>, + HasBatchStrategy<T>, + HasGlobalBatchSize<T>, + HasReg<T>, + HasElasticNet<T>, + OnlineLogisticRegressionModelParams<T> { + + Param<Double> ALPHA = + new DoubleParam("alpha", "The parameter alpha of ftrl.", 0.1, ParamValidators.gt(0.0)); Review Comment: nits: Would it be more intuitive to say `The alpha parameter ...`? Same for `The beta parameter ...`. Same for the python parameter description. -- 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]
