You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by di...@apache.org on 2021/04/01 00:01:41 UTC
[flink] 03/05: [FLINK-21976] Remove Flink ML related examples from
flink-examples
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2519c1097434fd67081f82a26d13ad93bf31800e
Author: Dong Lin <li...@gmail.com>
AuthorDate: Wed Mar 31 12:10:07 2021 +0800
[FLINK-21976] Remove Flink ML related examples from flink-examples
This closes #15405.
---
.../flink/examples/java/ml/LinearRegression.java | 257 ---------------------
.../java/ml/util/LinearRegressionData.java | 69 ------
.../ml/util/LinearRegressionDataGenerator.java | 115 ---------
.../flink/examples/scala/ml/LinearRegression.scala | 159 -------------
flink-examples/flink-examples-streaming/pom.xml | 25 --
.../examples/ml/IncrementalLearningSkeleton.java | 212 -----------------
.../ml/util/IncrementalLearningSkeletonData.java | 33 ---
.../examples/ml/IncrementalLearningSkeleton.scala | 184 ---------------
.../streaming/test/StreamingExamplesITCase.java | 9 -
.../scala/examples/StreamingExamplesITCase.scala | 9 -
10 files changed, 1072 deletions(-)
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/LinearRegression.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/LinearRegression.java
deleted file mode 100644
index 4f2f528..0000000
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/LinearRegression.java
+++ /dev/null
@@ -1,257 +0,0 @@
-/*
- * 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.examples.java.ml;
-
-import org.apache.flink.api.common.functions.MapFunction;
-import org.apache.flink.api.common.functions.ReduceFunction;
-import org.apache.flink.api.common.functions.RichMapFunction;
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.operators.IterativeDataSet;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.utils.ParameterTool;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.examples.java.ml.util.LinearRegressionData;
-
-import java.io.Serializable;
-import java.util.Collection;
-
-/**
- * This example implements a basic Linear Regression to solve the y = theta0 + theta1*x problem
- * using batch gradient descent algorithm.
- *
- * <p>Linear Regression with BGD(batch gradient descent) algorithm is an iterative clustering
- * algorithm and works as follows:<br>
- * Giving a data set and target set, the BGD try to find out the best parameters for the data set to
- * fit the target set. In each iteration, the algorithm computes the gradient of the cost function
- * and use it to update all the parameters. The algorithm terminates after a fixed number of
- * iterations (as in this implementation) With enough iteration, the algorithm can minimize the cost
- * function and find the best parameters This is the Wikipedia entry for the <a href =
- * "http://en.wikipedia.org/wiki/Linear_regression">Linear regression</a> and <a href =
- * "http://en.wikipedia.org/wiki/Gradient_descent">Gradient descent algorithm</a>.
- *
- * <p>This implementation works on one-dimensional data. And find the two-dimensional theta.<br>
- * It find the best Theta parameter to fit the target.
- *
- * <p>Input files are plain text files and must be formatted as follows:
- *
- * <ul>
- * <li>Data points are represented as two double values separated by a blank character. The first
- * one represent the X(the training data) and the second represent the Y(target). Data points
- * are separated by newline characters.<br>
- * For example <code>"-0.02 -0.04\n5.3 10.6\n"</code> gives two data points (x=-0.02, y=-0.04)
- * and (x=5.3, y=10.6).
- * </ul>
- *
- * <p>This example shows how to use:
- *
- * <ul>
- * <li>Bulk iterations
- * <li>Broadcast variables in bulk iterations
- * <li>Custom Java objects (PoJos)
- * </ul>
- */
-@SuppressWarnings("serial")
-public class LinearRegression {
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
- public static void main(String[] args) throws Exception {
-
- final ParameterTool params = ParameterTool.fromArgs(args);
-
- // set up execution environment
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- final int iterations = params.getInt("iterations", 10);
-
- // make parameters available in the web interface
- env.getConfig().setGlobalJobParameters(params);
-
- // get input x data from elements
- DataSet<Data> data;
- if (params.has("input")) {
- // read data from CSV file
- data =
- env.readCsvFile(params.get("input"))
- .fieldDelimiter(" ")
- .includeFields(true, true)
- .pojoType(Data.class);
- } else {
- System.out.println("Executing LinearRegression example with default input data set.");
- System.out.println("Use --input to specify file input.");
- data = LinearRegressionData.getDefaultDataDataSet(env);
- }
-
- // get the parameters from elements
- DataSet<Params> parameters = LinearRegressionData.getDefaultParamsDataSet(env);
-
- // set number of bulk iterations for SGD linear Regression
- IterativeDataSet<Params> loop = parameters.iterate(iterations);
-
- DataSet<Params> newParameters =
- data
- // compute a single step using every sample
- .map(new SubUpdate())
- .withBroadcastSet(loop, "parameters")
- // sum up all the steps
- .reduce(new UpdateAccumulator())
- // average the steps and update all parameters
- .map(new Update());
-
- // feed new parameters back into next iteration
- DataSet<Params> result = loop.closeWith(newParameters);
-
- // emit result
- if (params.has("output")) {
- result.writeAsText(params.get("output"));
- // execute program
- env.execute("Linear Regression example");
- } else {
- System.out.println("Printing result to stdout. Use --output to specify output path.");
- result.print();
- }
- }
-
- // *************************************************************************
- // DATA TYPES
- // *************************************************************************
-
- /** A simple data sample, x means the input, and y means the target. */
- public static class Data implements Serializable {
- public double x, y;
-
- public Data() {}
-
- public Data(double x, double y) {
- this.x = x;
- this.y = y;
- }
-
- @Override
- public String toString() {
- return "(" + x + "|" + y + ")";
- }
- }
-
- /** A set of parameters -- theta0, theta1. */
- public static class Params implements Serializable {
-
- private double theta0, theta1;
-
- public Params() {}
-
- public Params(double x0, double x1) {
- this.theta0 = x0;
- this.theta1 = x1;
- }
-
- @Override
- public String toString() {
- return theta0 + " " + theta1;
- }
-
- public double getTheta0() {
- return theta0;
- }
-
- public double getTheta1() {
- return theta1;
- }
-
- public void setTheta0(double theta0) {
- this.theta0 = theta0;
- }
-
- public void setTheta1(double theta1) {
- this.theta1 = theta1;
- }
-
- public Params div(Integer a) {
- this.theta0 = theta0 / a;
- this.theta1 = theta1 / a;
- return this;
- }
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- /** Compute a single BGD type update for every parameters. */
- public static class SubUpdate extends RichMapFunction<Data, Tuple2<Params, Integer>> {
-
- private Collection<Params> parameters;
-
- private Params parameter;
-
- private int count = 1;
-
- /** Reads the parameters from a broadcast variable into a collection. */
- @Override
- public void open(Configuration parameters) throws Exception {
- this.parameters = getRuntimeContext().getBroadcastVariable("parameters");
- }
-
- @Override
- public Tuple2<Params, Integer> map(Data in) throws Exception {
-
- for (Params p : parameters) {
- this.parameter = p;
- }
-
- double theta0 =
- parameter.theta0
- - 0.01 * ((parameter.theta0 + (parameter.theta1 * in.x)) - in.y);
- double theta1 =
- parameter.theta1
- - 0.01
- * (((parameter.theta0 + (parameter.theta1 * in.x)) - in.y)
- * in.x);
-
- return new Tuple2<Params, Integer>(new Params(theta0, theta1), count);
- }
- }
-
- /** Accumulator all the update. */
- public static class UpdateAccumulator implements ReduceFunction<Tuple2<Params, Integer>> {
-
- @Override
- public Tuple2<Params, Integer> reduce(
- Tuple2<Params, Integer> val1, Tuple2<Params, Integer> val2) {
-
- double newTheta0 = val1.f0.theta0 + val2.f0.theta0;
- double newTheta1 = val1.f0.theta1 + val2.f0.theta1;
- Params result = new Params(newTheta0, newTheta1);
- return new Tuple2<Params, Integer>(result, val1.f1 + val2.f1);
- }
- }
-
- /** Compute the final update by average them. */
- public static class Update implements MapFunction<Tuple2<Params, Integer>, Params> {
-
- @Override
- public Params map(Tuple2<Params, Integer> arg0) throws Exception {
-
- return arg0.f0.div(arg0.f1);
- }
- }
-}
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/util/LinearRegressionData.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/util/LinearRegressionData.java
deleted file mode 100644
index 5cb1339..0000000
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/util/LinearRegressionData.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * 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.examples.java.ml.util;
-
-import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.examples.java.ml.LinearRegression.Data;
-import org.apache.flink.examples.java.ml.LinearRegression.Params;
-
-import java.util.LinkedList;
-import java.util.List;
-
-/**
- * Provides the default data sets used for the Linear Regression example program. The default data
- * sets are used, if no parameters are given to the program.
- */
-public class LinearRegressionData {
-
- // We have the data as object arrays so that we can also generate Scala Data
- // Sources from it.
- public static final Object[][] PARAMS = new Object[][] {new Object[] {0.0, 0.0}};
-
- public static final Object[][] DATA =
- new Object[][] {
- new Object[] {0.5, 1.0}, new Object[] {1.0, 2.0},
- new Object[] {2.0, 4.0}, new Object[] {3.0, 6.0},
- new Object[] {4.0, 8.0}, new Object[] {5.0, 10.0},
- new Object[] {6.0, 12.0}, new Object[] {7.0, 14.0},
- new Object[] {8.0, 16.0}, new Object[] {9.0, 18.0},
- new Object[] {10.0, 20.0}, new Object[] {-0.08, -0.16},
- new Object[] {0.13, 0.26}, new Object[] {-1.17, -2.35},
- new Object[] {1.72, 3.45}, new Object[] {1.70, 3.41},
- new Object[] {1.20, 2.41}, new Object[] {-0.59, -1.18},
- new Object[] {0.28, 0.57}, new Object[] {1.65, 3.30},
- new Object[] {-0.55, -1.08}
- };
-
- public static DataSet<Params> getDefaultParamsDataSet(ExecutionEnvironment env) {
- List<Params> paramsList = new LinkedList<>();
- for (Object[] params : PARAMS) {
- paramsList.add(new Params((Double) params[0], (Double) params[1]));
- }
- return env.fromCollection(paramsList);
- }
-
- public static DataSet<Data> getDefaultDataDataSet(ExecutionEnvironment env) {
- List<Data> dataList = new LinkedList<>();
- for (Object[] data : DATA) {
- dataList.add(new Data((Double) data[0], (Double) data[1]));
- }
- return env.fromCollection(dataList);
- }
-}
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/util/LinearRegressionDataGenerator.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/util/LinearRegressionDataGenerator.java
deleted file mode 100644
index 52a912f..0000000
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/util/LinearRegressionDataGenerator.java
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * 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.examples.java.ml.util;
-
-import java.io.BufferedWriter;
-import java.io.File;
-import java.io.FileWriter;
-import java.io.IOException;
-import java.text.DecimalFormat;
-import java.util.Locale;
-import java.util.Random;
-
-/**
- * Generates data for the {@link org.apache.flink.examples.java.ml.LinearRegression} example
- * program.
- */
-public class LinearRegressionDataGenerator {
-
- static {
- Locale.setDefault(Locale.US);
- }
-
- private static final String POINTS_FILE = "data";
- private static final long DEFAULT_SEED = 4650285087650871364L;
- private static final int DIMENSIONALITY = 1;
- private static final DecimalFormat FORMAT = new DecimalFormat("#0.00");
- private static final char DELIMITER = ' ';
-
- /**
- * Main method to generate data for the {@link
- * org.apache.flink.examples.java.ml.LinearRegression} example program.
- *
- * <p>The generator creates to files:
- *
- * <ul>
- * <li><code>{tmp.dir}/data</code> for the data points
- * </ul>
- *
- * @param args
- * <ol>
- * <li>Int: Number of data points
- * <li><b>Optional</b> Long: Random seed
- * </ol>
- */
- public static void main(String[] args) throws IOException {
-
- // check parameter count
- if (args.length < 1) {
- System.out.println("LinearRegressionDataGenerator <numberOfDataPoints> [<seed>]");
- System.exit(1);
- }
-
- // parse parameters
- final int numDataPoints = Integer.parseInt(args[0]);
- final long firstSeed = args.length > 1 ? Long.parseLong(args[4]) : DEFAULT_SEED;
- final Random random = new Random(firstSeed);
- final String tmpDir = System.getProperty("java.io.tmpdir");
-
- // write the points out
- BufferedWriter pointsOut = null;
- try {
- pointsOut = new BufferedWriter(new FileWriter(new File(tmpDir + "/" + POINTS_FILE)));
- StringBuilder buffer = new StringBuilder();
-
- // DIMENSIONALITY + 1 means that the number of x(dimensionality) and target y
- double[] point = new double[DIMENSIONALITY + 1];
-
- for (int i = 1; i <= numDataPoints; i++) {
- point[0] = random.nextGaussian();
- point[1] = 2 * point[0] + 0.01 * random.nextGaussian();
- writePoint(point, buffer, pointsOut);
- }
-
- } finally {
- if (pointsOut != null) {
- pointsOut.close();
- }
- }
-
- System.out.println(
- "Wrote " + numDataPoints + " data points to " + tmpDir + "/" + POINTS_FILE);
- }
-
- private static void writePoint(double[] data, StringBuilder buffer, BufferedWriter out)
- throws IOException {
- buffer.setLength(0);
-
- // write coordinates
- for (int j = 0; j < data.length; j++) {
- buffer.append(FORMAT.format(data[j]));
- if (j < data.length - 1) {
- buffer.append(DELIMITER);
- }
- }
-
- out.write(buffer.toString());
- out.newLine();
- }
-}
diff --git a/flink-examples/flink-examples-batch/src/main/scala/org/apache/flink/examples/scala/ml/LinearRegression.scala b/flink-examples/flink-examples-batch/src/main/scala/org/apache/flink/examples/scala/ml/LinearRegression.scala
deleted file mode 100644
index 4663db6..0000000
--- a/flink-examples/flink-examples-batch/src/main/scala/org/apache/flink/examples/scala/ml/LinearRegression.scala
+++ /dev/null
@@ -1,159 +0,0 @@
-/*
- * 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.examples.scala.ml
-
-import org.apache.flink.api.common.functions._
-import org.apache.flink.api.java.utils.ParameterTool
-import org.apache.flink.api.scala._
-import org.apache.flink.configuration.Configuration
-import org.apache.flink.examples.java.ml.util.LinearRegressionData
-
-import scala.collection.JavaConverters._
-
-/**
- * This example implements a basic Linear Regression to solve the y = theta0 + theta1*x problem
- * using batch gradient descent algorithm.
- *
- * Linear Regression with BGD(batch gradient descent) algorithm is an iterative algorithm and
- * works as follows:
- *
- * Giving a data set and target set, the BGD try to find out the best parameters for the data set
- * to fit the target set.
- * In each iteration, the algorithm computes the gradient of the cost function and use it to
- * update all the parameters.
- * The algorithm terminates after a fixed number of iterations (as in this implementation).
- * With enough iteration, the algorithm can minimize the cost function and find the best parameters
- * This is the Wikipedia entry for the
- * [[http://en.wikipedia.org/wiki/Linear_regression Linear regression]] and
- * [[http://en.wikipedia.org/wiki/Gradient_descent Gradient descent algorithm]].
- *
- * This implementation works on one-dimensional data and finds the best two-dimensional theta to
- * fit the target.
- *
- * Input files are plain text files and must be formatted as follows:
- *
- * - Data points are represented as two double values separated by a blank character. The first
- * one represent the X(the training data) and the second represent the Y(target). Data points are
- * separated by newline characters.
- * For example `"-0.02 -0.04\n5.3 10.6\n"`gives two data points
- * (x=-0.02, y=-0.04) and (x=5.3, y=10.6).
- *
- * This example shows how to use:
- *
- * - Bulk iterations
- * - Broadcast variables in bulk iterations
- */
-object LinearRegression {
-
- def main(args: Array[String]) {
-
- val params: ParameterTool = ParameterTool.fromArgs(args)
-
- // set up execution environment
- val env = ExecutionEnvironment.getExecutionEnvironment
-
- // make parameters available in the web interface
- env.getConfig.setGlobalJobParameters(params)
-
- val parameters = env.fromCollection(LinearRegressionData.PARAMS map {
- case Array(x, y) => Params(x.asInstanceOf[Double], y.asInstanceOf[Double])
- })
-
- val data =
- if (params.has("input")) {
- env.readCsvFile[(Double, Double)](
- params.get("input"),
- fieldDelimiter = " ",
- includedFields = Array(0, 1))
- .map { t => new Data(t._1, t._2) }
- } else {
- println("Executing LinearRegression example with default input data set.")
- println("Use --input to specify file input.")
- val data = LinearRegressionData.DATA map {
- case Array(x, y) => Data(x.asInstanceOf[Double], y.asInstanceOf[Double])
- }
- env.fromCollection(data)
- }
-
- val numIterations = params.getInt("iterations", 10)
-
- val result = parameters.iterate(numIterations) { currentParameters =>
- val newParameters = data
- .map(new SubUpdate).withBroadcastSet(currentParameters, "parameters")
- .reduce { (p1, p2) =>
- val result = p1._1 + p2._1
- (result, p1._2 + p2._2)
- }
- .map { x => x._1.div(x._2) }
- newParameters
- }
-
- if (params.has("output")) {
- result.writeAsText(params.get("output"))
- env.execute("Scala Linear Regression example")
- } else {
- println("Printing result to stdout. Use --output to specify output path.")
- result.print()
- }
- }
-
- /**
- * A simple data sample, x means the input, and y means the target.
- */
- case class Data(var x: Double, var y: Double)
-
- /**
- * A set of parameters -- theta0, theta1.
- */
- case class Params(theta0: Double, theta1: Double) {
- def div(a: Int): Params = {
- Params(theta0 / a, theta1 / a)
- }
-
- def + (other: Params) = {
- Params(theta0 + other.theta0, theta1 + other.theta1)
- }
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- /**
- * Compute a single BGD type update for every parameters.
- */
- class SubUpdate extends RichMapFunction[Data, (Params, Int)] {
-
- private var parameter: Params = null
-
- /** Reads the parameters from a broadcast variable into a collection. */
- override def open(parameters: Configuration) {
- val parameters = getRuntimeContext.getBroadcastVariable[Params]("parameters").asScala
- parameter = parameters.head
- }
-
- def map(in: Data): (Params, Int) = {
- val theta0 =
- parameter.theta0 - 0.01 * ((parameter.theta0 + (parameter.theta1 * in.x)) - in.y)
- val theta1 =
- parameter.theta1 - 0.01 * (((parameter.theta0 + (parameter.theta1 * in.x)) - in.y) * in.x)
- (Params(theta0, theta1), 1)
- }
- }
-}
diff --git a/flink-examples/flink-examples-streaming/pom.xml b/flink-examples/flink-examples-streaming/pom.xml
index 214cff5..c487a94 100644
--- a/flink-examples/flink-examples-streaming/pom.xml
+++ b/flink-examples/flink-examples-streaming/pom.xml
@@ -212,30 +212,6 @@ under the License.
</configuration>
</execution>
- <!-- IncrementalLearning -->
- <execution>
- <id>IncrementalLearning</id>
- <phase>package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
- <configuration>
- <classifier>IncrementalLearning</classifier>
-
- <archive>
- <manifestEntries>
- <program-class>org.apache.flink.streaming.examples.ml.IncrementalLearningSkeleton</program-class>
- </manifestEntries>
- </archive>
-
- <includes>
- <include>org/apache/flink/streaming/examples/ml/*.class</include>
- <include>META-INF/LICENSE</include>
- <include>META-INF/NOTICE</include>
- </includes>
- </configuration>
- </execution>
-
<!-- WindowJoin -->
<execution>
<id>WindowJoin</id>
@@ -557,7 +533,6 @@ under the License.
<id>rename</id>
<configuration>
<target>
- <copy file="${project.basedir}/target/flink-examples-streaming_${scala.binary.version}-${project.version}-IncrementalLearning.jar" tofile="${project.basedir}/target/IncrementalLearning.jar" />
<copy file="${project.basedir}/target/flink-examples-streaming_${scala.binary.version}-${project.version}-Iteration.jar" tofile="${project.basedir}/target/Iteration.jar" />
<copy file="${project.basedir}/target/flink-examples-streaming_${scala.binary.version}-${project.version}-SessionWindowing.jar" tofile="${project.basedir}/target/SessionWindowing.jar" />
<copy file="${project.basedir}/target/flink-examples-streaming_${scala.binary.version}-${project.version}-TopSpeedWindowing.jar" tofile="${project.basedir}/target/TopSpeedWindowing.jar" />
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/IncrementalLearningSkeleton.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/IncrementalLearningSkeleton.java
deleted file mode 100644
index 4f55f61..0000000
--- a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/IncrementalLearningSkeleton.java
+++ /dev/null
@@ -1,212 +0,0 @@
-/*
- * 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.examples.ml;
-
-import org.apache.flink.api.java.utils.ParameterTool;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
-import org.apache.flink.streaming.api.functions.co.CoMapFunction;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
-import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
-import org.apache.flink.streaming.api.watermark.Watermark;
-import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
-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>
- */
-public class IncrementalLearningSkeleton {
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
- public static void main(String[] args) throws Exception {
-
- // Checking input parameters
- final ParameterTool params = ParameterTool.fromArgs(args);
-
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
- DataStream<Integer> trainingData = env.addSource(new FiniteTrainingDataSource());
- DataStream<Integer> newData = env.addSource(new FiniteNewDataSource());
-
- // build new model on every second of new data
- DataStream<Double[]> model =
- trainingData
- .assignTimestampsAndWatermarks(new LinearTimestamp())
- .windowAll(TumblingEventTimeWindows.of(Time.milliseconds(5000)))
- .apply(new PartialModelBuilder());
-
- // use partial model for newData
- DataStream<Integer> prediction = newData.connect(model).map(new Predictor());
-
- // emit result
- if (params.has("output")) {
- prediction.writeAsText(params.get("output"));
- } else {
- System.out.println("Printing result to stdout. Use --output to specify output path.");
- prediction.print();
- }
-
- // execute program
- env.execute("Streaming Incremental Learning");
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- /**
- * Feeds new data for newData. By default it is implemented as constantly emitting the Integer 1
- * in a loop.
- */
- public static class FiniteNewDataSource implements SourceFunction<Integer> {
- private static final long serialVersionUID = 1L;
- private int counter;
-
- @Override
- public void run(SourceContext<Integer> ctx) throws Exception {
- Thread.sleep(15);
- while (counter < 50) {
- ctx.collect(getNewData());
- }
- }
-
- @Override
- public void cancel() {
- // No cleanup needed
- }
-
- private Integer getNewData() throws InterruptedException {
- Thread.sleep(5);
- counter++;
- return 1;
- }
- }
-
- /**
- * Feeds new training data for the partial model builder. By default it is implemented as
- * constantly emitting the Integer 1 in a loop.
- */
- public static class FiniteTrainingDataSource implements SourceFunction<Integer> {
- private static final long serialVersionUID = 1L;
- private int counter = 0;
-
- @Override
- public void run(SourceContext<Integer> collector) throws Exception {
- while (counter < 8200) {
- collector.collect(getTrainingData());
- }
- }
-
- @Override
- public void cancel() {
- // No cleanup needed
- }
-
- private Integer getTrainingData() throws InterruptedException {
- counter++;
- return 1;
- }
- }
-
- private static class LinearTimestamp implements AssignerWithPunctuatedWatermarks<Integer> {
- private static final long serialVersionUID = 1L;
-
- private long counter = 0L;
-
- @Override
- public long extractTimestamp(Integer element, long previousElementTimestamp) {
- return counter += 10L;
- }
-
- @Override
- public Watermark checkAndGetNextWatermark(Integer lastElement, long extractedTimestamp) {
- return new Watermark(counter - 1);
- }
- }
-
- /** Builds up-to-date partial models on new training data. */
- public static class PartialModelBuilder
- implements AllWindowFunction<Integer, Double[], TimeWindow> {
- private static final long serialVersionUID = 1L;
-
- protected Double[] buildPartialModel(Iterable<Integer> values) {
- return new Double[] {1.};
- }
-
- @Override
- public void apply(TimeWindow window, Iterable<Integer> values, Collector<Double[]> out)
- throws Exception {
- out.collect(buildPartialModel(values));
- }
- }
-
- /**
- * Creates newData using the model produced in batch-processing and the up-to-date partial
- * model.
- *
- * <p>By default emits the Integer 0 for every newData and the Integer 1 for every model update.
- */
- public static class Predictor implements CoMapFunction<Integer, Double[], Integer> {
- private static final long serialVersionUID = 1L;
-
- Double[] batchModel = null;
- Double[] partialModel = null;
-
- @Override
- public Integer map1(Integer value) {
- // Return newData
- return predict(value);
- }
-
- @Override
- public Integer map2(Double[] value) {
- // Update model
- partialModel = value;
- batchModel = getBatchModel();
- return 1;
- }
-
- // pulls model built with batch-job on the old training data
- protected Double[] getBatchModel() {
- return new Double[] {0.};
- }
-
- // performs newData using the two models
- protected Integer predict(Integer inTuple) {
- return 0;
- }
- }
-}
diff --git a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/util/IncrementalLearningSkeletonData.java b/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/util/IncrementalLearningSkeletonData.java
deleted file mode 100644
index eaf5bdb..0000000
--- a/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/ml/util/IncrementalLearningSkeletonData.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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.examples.ml.util;
-
-/** Data for IncrementalLearningSkeletonITCase. */
-public class IncrementalLearningSkeletonData {
-
- public static final String RESULTS =
- "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n"
- + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "1\n" + "0\n" + "0\n" + "0\n" + "0\n"
- + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n"
- + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n"
- + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n"
- + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n"
- + "0\n" + "0\n" + "0\n" + "0\n" + "0\n" + "0\n";
-
- private IncrementalLearningSkeletonData() {}
-}
diff --git a/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/ml/IncrementalLearningSkeleton.scala b/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/ml/IncrementalLearningSkeleton.scala
deleted file mode 100644
index 2f1e168..0000000
--- a/flink-examples/flink-examples-streaming/src/main/scala/org/apache/flink/streaming/scala/examples/ml/IncrementalLearningSkeleton.scala
+++ /dev/null
@@ -1,184 +0,0 @@
-/*
- * 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 org.apache.flink.api.java.utils.ParameterTool
-import org.apache.flink.api.scala._
-import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
-import org.apache.flink.streaming.api.functions.co.CoMapFunction
-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.function.AllWindowFunction
-import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
-import org.apache.flink.streaming.api.watermark.Watermark
-import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
-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.
- *
- * 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.
- *
- * This example shows how to use:
- *
- * - Connected streams
- * - CoFunctions
- * - Tuple data types
- *
- */
-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
-
- // build new model on every second of new data
- val trainingData: DataStream[Integer] = env.addSource(new FiniteTrainingDataSource)
- val newData: DataStream[Integer] = env.addSource(new FiniteNewDataSource)
-
- val model: DataStream[Array[java.lang.Double]] = trainingData
- .assignTimestampsAndWatermarks(new LinearTimestamp)
- .windowAll(TumblingEventTimeWindows.of(Time.milliseconds(5000)))
- .apply(new PartialModelBuilder)
-
- // use partial model for newData
- val prediction: DataStream[Integer] = newData.connect(model).map(new Predictor)
-
- // emit result
- if (params.has("output")) {
- prediction.writeAsText(params.get("output"))
- } else {
- println("Printing result to stdout. Use --output to specify output path.")
- prediction.print()
- }
-
- // execute program
- env.execute("Streaming Incremental Learning")
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- /**
- * Feeds new data for newData. By default it is implemented as constantly
- * emitting the Integer 1 in a loop.
- */
- private class FiniteNewDataSource extends SourceFunction[Integer] {
- override def run(ctx: SourceContext[Integer]) = {
- Thread.sleep(15)
- (0 until 50).foreach{ _ =>
- Thread.sleep(5)
- ctx.collect(1)
- }
- }
-
- override def cancel() = {
- // No cleanup needed
- }
- }
-
- /**
- * Feeds new training data for the partial model builder. By default it is
- * implemented as constantly emitting the Integer 1 in a loop.
- */
- private class FiniteTrainingDataSource extends SourceFunction[Integer] {
- override def run(ctx: SourceContext[Integer]) =
- (0 until 8200).foreach( _ => ctx.collect(1) )
-
- override def cancel() = {
- // No cleanup needed
- }
- }
-
- private class LinearTimestamp extends AssignerWithPunctuatedWatermarks[Integer] {
- var counter = 0L
-
- override def extractTimestamp(element: Integer, previousElementTimestamp: Long): Long = {
- counter += 10L
- counter
- }
-
- override def checkAndGetNextWatermark(lastElement: Integer, extractedTimestamp: Long) = {
- new Watermark(counter - 1)
- }
- }
-
- /**
- * Builds up-to-date partial models on new training data.
- */
- private class PartialModelBuilder
- extends AllWindowFunction[Integer, Array[java.lang.Double], TimeWindow] {
-
- protected def buildPartialModel(values: Iterable[Integer]): Array[java.lang.Double] =
- Array[java.lang.Double](1)
-
- override def apply(window: TimeWindow,
- values: Iterable[Integer],
- out: Collector[Array[java.lang.Double]]): Unit = {
- out.collect(buildPartialModel(values))
- }
- }
-
- /**
- * Creates newData using the model produced in batch-processing and the
- * up-to-date partial model.
- *
- * By default emits the Integer 0 for every newData and the Integer 1
- * for every model update.
- *
- */
- private class Predictor extends CoMapFunction[Integer, Array[java.lang.Double], Integer] {
-
- var batchModel: Array[java.lang.Double] = null
- var partialModel: Array[java.lang.Double] = null
-
- override def map1(value: Integer): Integer = {
- // Return newData
- predict(value)
- }
-
- override def map2(value: Array[java.lang.Double]): Integer = {
- // Update model
- partialModel = value
- batchModel = getBatchModel()
- 1
- }
-
- // pulls model built with batch-job on the old training data
- protected def getBatchModel(): Array[java.lang.Double] = Array[java.lang.Double](0)
-
- // performs newData using the two models
- protected def predict(inTuple: Int): Int = 0
- }
-
-}
diff --git a/flink-examples/flink-examples-streaming/src/test/java/org/apache/flink/streaming/test/StreamingExamplesITCase.java b/flink-examples/flink-examples-streaming/src/test/java/org/apache/flink/streaming/test/StreamingExamplesITCase.java
index d8f8b56..7598f98 100644
--- a/flink-examples/flink-examples-streaming/src/test/java/org/apache/flink/streaming/test/StreamingExamplesITCase.java
+++ b/flink-examples/flink-examples-streaming/src/test/java/org/apache/flink/streaming/test/StreamingExamplesITCase.java
@@ -30,7 +30,6 @@ import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.examples.iteration.util.IterateExampleData;
-import org.apache.flink.streaming.examples.ml.util.IncrementalLearningSkeletonData;
import org.apache.flink.streaming.examples.twitter.util.TwitterExampleData;
import org.apache.flink.streaming.examples.windowing.util.SessionWindowingData;
import org.apache.flink.streaming.test.examples.join.WindowJoinData;
@@ -107,14 +106,6 @@ public class StreamingExamplesITCase extends AbstractTestBase {
}
@Test
- public void testIncrementalLearningSkeleton() throws Exception {
- final String resultPath = getTempDirPath("result");
- org.apache.flink.streaming.examples.ml.IncrementalLearningSkeleton.main(
- new String[] {"--output", resultPath});
- compareResultsByLinesInMemory(IncrementalLearningSkeletonData.RESULTS, resultPath);
- }
-
- @Test
public void testTwitterStream() throws Exception {
final String resultPath = getTempDirPath("result");
org.apache.flink.streaming.examples.twitter.TwitterExample.main(
diff --git a/flink-examples/flink-examples-streaming/src/test/scala/org/apache/flink/streaming/scala/examples/StreamingExamplesITCase.scala b/flink-examples/flink-examples-streaming/src/test/scala/org/apache/flink/streaming/scala/examples/StreamingExamplesITCase.scala
index 7407294..de00b95 100644
--- a/flink-examples/flink-examples-streaming/src/test/scala/org/apache/flink/streaming/scala/examples/StreamingExamplesITCase.scala
+++ b/flink-examples/flink-examples-streaming/src/test/scala/org/apache/flink/streaming/scala/examples/StreamingExamplesITCase.scala
@@ -25,13 +25,11 @@ import org.apache.flink.core.fs.FileSystem.WriteMode
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.examples.iteration.util.IterateExampleData
-import org.apache.flink.streaming.examples.ml.util.IncrementalLearningSkeletonData
import org.apache.flink.streaming.examples.twitter.util.TwitterExampleData
import org.apache.flink.streaming.examples.windowing.util.SessionWindowingData
import org.apache.flink.streaming.scala.examples.iteration.IterateExample
import org.apache.flink.streaming.scala.examples.join.WindowJoin
import org.apache.flink.streaming.scala.examples.join.WindowJoin.{Grade, Salary}
-import org.apache.flink.streaming.scala.examples.ml.IncrementalLearningSkeleton
import org.apache.flink.streaming.scala.examples.twitter.TwitterExample
import org.apache.flink.streaming.scala.examples.windowing.{SessionWindowing, WindowWordCount}
import org.apache.flink.streaming.scala.examples.wordcount.WordCount
@@ -96,13 +94,6 @@ class StreamingExamplesITCase extends AbstractTestBase {
}
@Test
- def testIncrementalLearningSkeleton(): Unit = {
- val resultPath = getTempDirPath("result")
- IncrementalLearningSkeleton.main(Array("--output", resultPath))
- TestBaseUtils.compareResultsByLinesInMemory(IncrementalLearningSkeletonData.RESULTS, resultPath)
- }
-
- @Test
def testTwitterExample(): Unit = {
val resultPath = getTempDirPath("result")
TwitterExample.main(Array("--output", resultPath))