Github user thvasilo commented on a diff in the pull request:
https://github.com/apache/flink/pull/2761#discussion_r89469371
--- Diff:
flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/ml/IncrementalLearningSkeleton.scala
---
@@ -0,0 +1,169 @@
+/*
+ * 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.streaming.scala.examples.ml
+
+import java.util.concurrent.TimeUnit
+
+import org.apache.flink.api.java.utils.ParameterTool
+import org.apache.flink.api.scala._
+import org.apache.flink.streaming.api.TimeCharacteristic
+import
org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
+import org.apache.flink.streaming.api.functions.source.SourceFunction
+import
org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
+import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
+import org.apache.flink.streaming.api.scala.function.AllWindowFunction
+import org.apache.flink.streaming.api.watermark.Watermark
+import org.apache.flink.streaming.api.windowing.time.Time
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow
+import org.apache.flink.util.Collector
+
+/**
+ * Skeleton for incremental machine learning algorithm consisting of a
+ * pre-computed model, which gets updated for the new inputs and new
input data
+ * for which the job provides predictions.
+ *
+ * <p>
+ * This may serve as a base of a number of algorithms, e.g. updating an
+ * incremental Alternating Least Squares model while also providing the
+ * predictions.
+ *
+ * <p>
+ * This example shows how to use:
+ * <ul>
+ * <li>Connected streams
+ * <li>CoFunctions
+ * <li>Tuple data types
+ * </ul>
+ */
+object IncrementalLearningSkeleton {
+
+ //
*************************************************************************
+ // PROGRAM
+ //
*************************************************************************
+
+ def main(args: Array[String]): Unit = {
+ // Checking input parameters
+ val params = ParameterTool.fromArgs(args)
+
+ // set up the execution environment
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+
+ // build new model on every second of new data
+ val trainingData = env.addSource(new FiniteTrainingDataSource)
+ val newData = env.addSource(new FiniteNewDataSource)
+
+ val model = trainingData
+ .assignTimestampsAndWatermarks(new LinearTimestamp)
+ .timeWindowAll(Time.of(5000, TimeUnit.MILLISECONDS))
+ .apply(new PartialModelBuilder)
+
+ // use partial model for newData
+ val prediction = newData.connect(model).map(
+ (_: Int) => 0,
--- End diff --
I think the way this example is implement does get the point of the example
across.
We are not trying to just generate a stream of 1s and 0s, the purpose is to
show that we can use the connected stream coming from the `newData` and the
`model` streams to read in a batch model which we enhance with the
`partialModel`, and then use the new data stream to continuously improve the
partial model.
I would recommend making this step more verbose as it is in the Java
version of the code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
