This is an automated email from the ASF dual-hosted git repository. lindong pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink-ml.git
commit f9327830b4da25940716714fc88eafbb56eed4b2 Author: Boris Tsirkin <[email protected]> AuthorDate: Wed May 18 21:48:46 2022 +0200 [FLINK-27798] Migrated to Flink 1.15.0. This closes #95. --- docs/README.md | 5 +- docs/content/docs/try-flink-ml/quick-start.md | 14 +- flink-ml-benchmark/README.md | 2 +- flink-ml-benchmark/pom.xml | 25 ++- .../flink/ml/benchmark/DataGeneratorTest.java | 3 + flink-ml-core/pom.xml | 97 ++++------ .../operator/AbstractBroadcastWrapperOperator.java | 6 +- flink-ml-dist/pom.xml | 8 +- flink-ml-dist/src/main/assemblies/bin.xml | 2 +- flink-ml-iteration/pom.xml | 73 ++------ .../datacache/nonkeyed/DataCacheSnapshot.java | 4 +- .../flink/iteration/operator/OperatorUtils.java | 5 +- .../flink/iteration/operator/HeadOperatorTest.java | 5 +- .../iteration/operator/ReplayOperatorTest.java | 2 +- .../MultipleInputAllRoundWrapperOperatorTest.java | 7 +- .../OneInputAllRoundWrapperOperatorTest.java | 3 +- .../TwoInputAllRoundWrapperOperatorTest.java | 5 +- .../MultipleInputPerRoundWrapperOperatorTest.java | 7 +- .../OneInputPerRoundWrapperOperatorTest.java | 3 +- .../TwoInputPerRoundWrapperOperatorTest.java | 5 +- flink-ml-lib/pom.xml | 18 +- .../BinaryClassificationEvaluator.java | 4 +- .../flink/ml/clustering/OnlineKMeansTest.java | 61 +++--- flink-ml-python/dev/dev-requirements.txt | 2 +- .../ml/lib/tests/test_ml_lib_completeness.py | 2 +- flink-ml-python/pyflink/ml/tests/test_utils.py | 2 +- flink-ml-python/setup.py | 2 +- flink-ml-tests/pom.xml | 8 +- flink-ml-uber/pom.xml | 18 +- pom.xml | 204 ++++++++++++++++++++- 30 files changed, 381 insertions(+), 221 deletions(-) diff --git a/docs/README.md b/docs/README.md index a18fb18..6e7faac 100644 --- a/docs/README.md +++ b/docs/README.md @@ -88,7 +88,7 @@ the page: Flink uses [shortcodes](https://gohugo.io/content-management/shortcodes/) to add custom functionality to its documentation markdown. For example: - {{< artifact flink-ml-core withScalaVersion >}} + {{< artifact flink-ml-core withTestScope >}} This will be replaced by the maven artifact for flink-ml-core that users should copy into their pom.xml file. It will render out to: @@ -96,8 +96,9 @@ should copy into their pom.xml file. It will render out to: ```xml <dependency> <groupdId>org.apache.flink</groupId> - <artifactId>flink-ml-core_2.12</artifactId> + <artifactId>flink-ml-core</artifactId> <version><!-- current flink version --></version> + <scope>test</scope> </dependency> ``` diff --git a/docs/content/docs/try-flink-ml/quick-start.md b/docs/content/docs/try-flink-ml/quick-start.md index beaab27..0ef5a19 100644 --- a/docs/content/docs/try-flink-ml/quick-start.md +++ b/docs/content/docs/try-flink-ml/quick-start.md @@ -35,11 +35,11 @@ Learning Model and use it to provide prediction service. In order to use Flink ML in a Maven project, add the following dependencies to `pom.xml`. -{{< artifact flink-ml-core withScalaVersion >}} +{{< artifact flink-ml-core >}} -{{< artifact flink-ml-iteration withScalaVersion >}} +{{< artifact flink-ml-iteration >}} -{{< artifact flink-ml-lib withScalaVersion >}} +{{< artifact flink-ml-lib >}} The example code provided in this document requires additional dependencies on the Flink Table API. In order to execute the example code successfully, please @@ -48,16 +48,16 @@ make sure the following dependencies also exist in `pom.xml`. ```xml <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-clients_2.12</artifactId> - <version>1.14.0</version> + <artifactId>flink-clients</artifactId> + <version>1.15.0</version> </dependency> ``` ```xml <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-table-planner_2.12</artifactId> - <version>1.14.0</version> + <artifactId>flink-table-planner-loader</artifactId> + <version>1.15.0</version> </dependency> ``` diff --git a/flink-ml-benchmark/README.md b/flink-ml-benchmark/README.md index 5edb3c7..69e4939 100644 --- a/flink-ml-benchmark/README.md +++ b/flink-ml-benchmark/README.md @@ -7,7 +7,7 @@ stages in a Linux/MacOS environment. ### Install Flink -Please make sure Flink 1.14 or higher version has been installed in your local +Please make sure Flink 1.15 or higher version has been installed in your local environment. You can refer to the [local installation](https://nightlies.apache.org/flink/flink-docs-master/docs/try-flink/local_installation/) instruction on Flink's document website for how to achieve this. diff --git a/flink-ml-benchmark/pom.xml b/flink-ml-benchmark/pom.xml index 5b704e8..ce30dbd 100644 --- a/flink-ml-benchmark/pom.xml +++ b/flink-ml-benchmark/pom.xml @@ -26,27 +26,27 @@ under the License. <version>2.1-SNAPSHOT</version> </parent> - <artifactId>flink-ml-benchmark_${scala.binary.version}</artifactId> + <artifactId>flink-ml-benchmark</artifactId> <name>Flink ML : Benchmark</name> <dependencies> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-ml-iteration_${scala.binary.version}</artifactId> + <artifactId>flink-ml-iteration</artifactId> <version>${project.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-ml-core_${scala.binary.version}</artifactId> + <artifactId>flink-ml-core</artifactId> <version>${project.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-ml-lib_${scala.binary.version}</artifactId> + <artifactId>flink-ml-lib</artifactId> <version>${project.version}</version> <scope>provided</scope> </dependency> @@ -60,21 +60,21 @@ under the License. <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId> + <artifactId>flink-table-api-java-bridge</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-clients_${scala.binary.version}</artifactId> + <artifactId>flink-clients</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-streaming-java_${scala.binary.version}</artifactId> + <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>test</scope> <type>test-jar</type> @@ -89,14 +89,19 @@ under the License. <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-table-planner_${scala.binary.version}</artifactId> - <version>${flink.version}</version> + <artifactId>flink-table-runtime</artifactId> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-planner-loader</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-test-utils_${scala.binary.version}</artifactId> + <artifactId>flink-test-utils</artifactId> <version>${flink.version}</version> <scope>test</scope> </dependency> diff --git a/flink-ml-benchmark/src/test/java/org/apache/flink/ml/benchmark/DataGeneratorTest.java b/flink-ml-benchmark/src/test/java/org/apache/flink/ml/benchmark/DataGeneratorTest.java index 66ecc86..937d25b 100644 --- a/flink-ml-benchmark/src/test/java/org/apache/flink/ml/benchmark/DataGeneratorTest.java +++ b/flink-ml-benchmark/src/test/java/org/apache/flink/ml/benchmark/DataGeneratorTest.java @@ -31,6 +31,7 @@ import org.apache.flink.util.CloseableIterator; import org.junit.Test; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; /** Tests data generators. */ @@ -52,6 +53,7 @@ public class DataGeneratorTest { Row row = it.next(); assertEquals(row.getArity(), 1); DenseVector vector = (DenseVector) row.getField(generator.getColNames()[0][0]); + assertNotNull(vector); assertEquals(vector.size(), generator.getVectorDim()); count++; } @@ -107,6 +109,7 @@ public class DataGeneratorTest { Row row = it.next(); count++; DenseVector features = (DenseVector) row.getField(featuresCol); + assertNotNull(features); for (double value : features.values) { assertTrue(value >= 0); assertTrue(value <= generator.getFeatureArity() - 1); diff --git a/flink-ml-core/pom.xml b/flink-ml-core/pom.xml index 56dbe4a..9bad8a0 100644 --- a/flink-ml-core/pom.xml +++ b/flink-ml-core/pom.xml @@ -28,13 +28,13 @@ under the License. <version>2.1-SNAPSHOT</version> </parent> - <artifactId>flink-ml-core_${scala.binary.version}</artifactId> + <artifactId>flink-ml-core</artifactId> <name>Flink ML : Core</name> <dependencies> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-ml-iteration_${scala.binary.version}</artifactId> + <artifactId>flink-ml-iteration</artifactId> <version>${project.version}</version> <scope>provided</scope> </dependency> @@ -48,63 +48,66 @@ under the License. <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-streaming-java_${scala.binary.version}</artifactId> + <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-table-planner_${scala.binary.version}</artifactId> - <version>${flink.version}</version> - <scope>test</scope> + <artifactId>flink-table-runtime</artifactId> + <scope>provided</scope> + </dependency> + + <dependency> + <groupId>dev.ludovic.netlib</groupId> + <artifactId>blas</artifactId> + <version>2.2.1</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-table-runtime_${scala.binary.version}</artifactId> + <artifactId>flink-table-api-java-bridge</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-streaming-java_${scala.binary.version}</artifactId> - <version>${flink.version}</version> - <scope>test</scope> - <type>test-jar</type> + <artifactId>flink-connector-files</artifactId> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-test-utils_${scala.binary.version}</artifactId> - <version>${flink.version}</version> + <artifactId>flink-table-planner-loader</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-shaded-jackson</artifactId> - <scope>provided</scope> + <artifactId>flink-streaming-java</artifactId> + <version>${flink.version}</version> + <scope>test</scope> + <type>test-jar</type> </dependency> <dependency> - <groupId>dev.ludovic.netlib</groupId> - <artifactId>blas</artifactId> - <version>2.2.0</version> + <groupId>org.apache.flink</groupId> + <artifactId>flink-test-utils</artifactId> + <version>${flink.version}</version> + <scope>test</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId> - <version>${flink.version}</version> + <artifactId>flink-shaded-jackson</artifactId> <scope>provided</scope> </dependency> <dependency> <groupId>org.mockito</groupId> <artifactId>mockito-core</artifactId> - <version>2.21.0</version> + <version>2.28.2</version> <type>jar</type> <scope>test</scope> </dependency> @@ -114,57 +117,31 @@ under the License. <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <scope>test</scope> - <version>${hadoop.version}</version> - <exclusions> - <exclusion> - <groupId>log4j</groupId> - <artifactId>log4j</artifactId> - </exclusion> - </exclusions> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <scope>test</scope> - <version>${hadoop.version}</version> - <exclusions> - <exclusion> - <groupId>log4j</groupId> - <artifactId>log4j</artifactId> - </exclusion> - <exclusion> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-log4j12</artifactId> - </exclusion> - <exclusion> - <!-- This dependency is no longer shipped with the JDK since Java 9.--> - <groupId>jdk.tools</groupId> - <artifactId>jdk.tools</artifactId> - </exclusion> - </exclusions> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-minicluster</artifactId> <scope>test</scope> - <version>${hadoop.version}</version> - <exclusions> - <exclusion> - <groupId>log4j</groupId> - <artifactId>log4j</artifactId> - </exclusion> - <exclusion> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-log4j12</artifactId> - </exclusion> - <exclusion> - <!-- This dependency is no longer shipped with the JDK since Java 9.--> - <groupId>jdk.tools</groupId> - <artifactId>jdk.tools</artifactId> - </exclusion> - </exclusions> + </dependency> + <!-- enforcing jackson versions --> + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-databind</artifactId> + </dependency> + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-core</artifactId> + </dependency> + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-annotations</artifactId> </dependency> </dependencies> </project> diff --git a/flink-ml-core/src/main/java/org/apache/flink/ml/common/broadcast/operator/AbstractBroadcastWrapperOperator.java b/flink-ml-core/src/main/java/org/apache/flink/ml/common/broadcast/operator/AbstractBroadcastWrapperOperator.java index 0dc4cfc..67c2a2d 100644 --- a/flink-ml-core/src/main/java/org/apache/flink/ml/common/broadcast/operator/AbstractBroadcastWrapperOperator.java +++ b/flink-ml-core/src/main/java/org/apache/flink/ml/common/broadcast/operator/AbstractBroadcastWrapperOperator.java @@ -46,7 +46,7 @@ import org.apache.flink.runtime.state.StateInitializationContext; import org.apache.flink.runtime.state.StatePartitionStreamProvider; import org.apache.flink.runtime.state.StateSnapshotContext; import org.apache.flink.runtime.util.NonClosingInputStreamDecorator; -import org.apache.flink.runtime.util.NonClosingOutpusStreamDecorator; +import org.apache.flink.runtime.util.NonClosingOutputStreamDecorator; import org.apache.flink.streaming.api.graph.StreamConfig; import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator; import org.apache.flink.streaming.api.operators.InternalTimeServiceManager; @@ -137,7 +137,7 @@ public abstract class AbstractBroadcastWrapperOperator<T, S extends StreamOperat * path of the file used to stored the cached records. It could be local file system or remote * file system. */ - private Path basePath; + private final Path basePath; /** DataCacheWriter for each input. */ @SuppressWarnings("rawtypes") @@ -561,7 +561,7 @@ public abstract class AbstractBroadcastWrapperOperator<T, S extends StreamOperat stateSnapshotContext.getRawOperatorStateOutput(); checkpointOutputStream.startNewPartition(); try (DataOutputStream dos = - new DataOutputStream(new NonClosingOutpusStreamDecorator(checkpointOutputStream))) { + new DataOutputStream(new NonClosingOutputStreamDecorator(checkpointOutputStream))) { dos.writeInt(numInputs); } for (int i = 0; i < numInputs; i++) { diff --git a/flink-ml-dist/pom.xml b/flink-ml-dist/pom.xml index 8cabfa1..2cc828b 100644 --- a/flink-ml-dist/pom.xml +++ b/flink-ml-dist/pom.xml @@ -26,7 +26,7 @@ under the License. <version>2.1-SNAPSHOT</version> </parent> - <artifactId>flink-ml-dist_${scala.binary.version}</artifactId> + <artifactId>flink-ml-dist</artifactId> <name>Flink ML : Dist</name> <packaging>jar</packaging> @@ -36,7 +36,7 @@ under the License. <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-ml-uber_${scala.binary.version}</artifactId> + <artifactId>flink-ml-uber</artifactId> <version>${project.version}</version> </dependency> @@ -45,11 +45,11 @@ under the License. <dependency> <groupId>org.apache.flink</groupId> <artifactId>statefun-flink-core</artifactId> - <version>3.1.0</version> + <version>${statefun.version}</version> <exclusions> <exclusion> <groupId>org.apache.flink</groupId> - <artifactId>flink-streaming-java_2.12</artifactId> + <artifactId>flink-streaming-java</artifactId> </exclusion> </exclusions> <scope>compile</scope> diff --git a/flink-ml-dist/src/main/assemblies/bin.xml b/flink-ml-dist/src/main/assemblies/bin.xml index de9bc46..3ca8d41 100644 --- a/flink-ml-dist/src/main/assemblies/bin.xml +++ b/flink-ml-dist/src/main/assemblies/bin.xml @@ -35,7 +35,7 @@ under the License. <includes> <include>org.apache.flink:statefun-flink-core</include> - <include>org.apache.flink:flink-ml-uber_${scala.binary.version}</include> + <include>org.apache.flink:flink-ml-uber</include> </includes> </dependencySet> </dependencySets> diff --git a/flink-ml-iteration/pom.xml b/flink-ml-iteration/pom.xml index 8ab408f..4e01461 100644 --- a/flink-ml-iteration/pom.xml +++ b/flink-ml-iteration/pom.xml @@ -21,6 +21,9 @@ under the License. xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> <modelVersion>4.0.0</modelVersion> + <properties> + <statefun-flink-core.version>3.2.0</statefun-flink-core.version> + </properties> <parent> <groupId>org.apache.flink</groupId> @@ -28,7 +31,7 @@ under the License. <version>2.1-SNAPSHOT</version> </parent> - <artifactId>flink-ml-iteration_${scala.binary.version}</artifactId> + <artifactId>flink-ml-iteration</artifactId> <name>Flink ML : Iteration</name> <dependencies> @@ -56,7 +59,7 @@ under the License. <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-streaming-java_${scala.binary.version}</artifactId> + <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> @@ -64,7 +67,7 @@ under the License. <!-- We have special treatment with the rocksdb state. --> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId> + <artifactId>flink-statebackend-rocksdb</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> @@ -73,25 +76,29 @@ under the License. <dependency> <groupId>org.apache.flink</groupId> <artifactId>statefun-flink-core</artifactId> - <version>3.1.0</version> + <version>${statefun.version}</version> <exclusions> <exclusion> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_2.12</artifactId> </exclusion> + <exclusion> + <groupId>org.apache.flink</groupId> + <artifactId>flink-metrics-dropwizard</artifactId> + </exclusion> </exclusions> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-clients_${scala.binary.version}</artifactId> + <artifactId>flink-clients</artifactId> <version>${flink.version}</version> <scope>test</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-streaming-java_${scala.binary.version}</artifactId> + <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>test</scope> <type>test-jar</type> @@ -110,13 +117,19 @@ under the License. <artifactId>flink-test-utils-junit</artifactId> <version>${flink.version}</version> <scope>test</scope> + <exclusions> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-annotations</artifactId> + </exclusion> + </exclusions> </dependency> <!-- We depends on StreamTaskMailboxTestHarnessBuilder and it requires this dependency --> <dependency> <groupId>org.mockito</groupId> <artifactId>mockito-core</artifactId> - <version>2.21.0</version> + <version>2.28.2</version> <type>jar</type> <scope>test</scope> </dependency> @@ -132,35 +145,12 @@ under the License. <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <scope>test</scope> - <version>${hadoop.version}</version> - <exclusions> - <exclusion> - <groupId>log4j</groupId> - <artifactId>log4j</artifactId> - </exclusion> - </exclusions> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <scope>test</scope> - <version>${hadoop.version}</version> - <exclusions> - <exclusion> - <groupId>log4j</groupId> - <artifactId>log4j</artifactId> - </exclusion> - <exclusion> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-log4j12</artifactId> - </exclusion> - <exclusion> - <!-- This dependency is no longer shipped with the JDK since Java 9.--> - <groupId>jdk.tools</groupId> - <artifactId>jdk.tools</artifactId> - </exclusion> - </exclusions> </dependency> <dependency> @@ -168,13 +158,6 @@ under the License. <artifactId>hadoop-hdfs</artifactId> <scope>test</scope> <type>test-jar</type> - <version>${hadoop.version}</version> - <exclusions> - <exclusion> - <groupId>log4j</groupId> - <artifactId>log4j</artifactId> - </exclusion> - </exclusions> </dependency> <dependency> @@ -182,22 +165,6 @@ under the License. <artifactId>hadoop-common</artifactId> <scope>test</scope> <type>test-jar</type> - <version>${hadoop.version}</version> - <exclusions> - <exclusion> - <groupId>log4j</groupId> - <artifactId>log4j</artifactId> - </exclusion> - <exclusion> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-log4j12</artifactId> - </exclusion> - <exclusion> - <!-- This dependency is no longer shipped with the JDK since Java 9.--> - <groupId>jdk.tools</groupId> - <artifactId>jdk.tools</artifactId> - </exclusion> - </exclusions> </dependency> </dependencies> </project> diff --git a/flink-ml-iteration/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/DataCacheSnapshot.java b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/DataCacheSnapshot.java index baaa36c..a00a440 100644 --- a/flink-ml-iteration/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/DataCacheSnapshot.java +++ b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/DataCacheSnapshot.java @@ -26,7 +26,7 @@ import org.apache.flink.core.fs.FileSystem; import org.apache.flink.core.fs.Path; import org.apache.flink.core.memory.DataInputViewStreamWrapper; import org.apache.flink.runtime.util.NonClosingInputStreamDecorator; -import org.apache.flink.runtime.util.NonClosingOutpusStreamDecorator; +import org.apache.flink.runtime.util.NonClosingOutputStreamDecorator; import org.apache.flink.statefun.flink.core.feedback.FeedbackConsumer; import org.apache.flink.util.IOUtils; import org.apache.flink.util.function.SupplierWithException; @@ -81,7 +81,7 @@ public class DataCacheSnapshot { public void writeTo(OutputStream checkpointOutputStream) throws IOException { try (DataOutputStream dos = - new DataOutputStream(new NonClosingOutpusStreamDecorator(checkpointOutputStream))) { + new DataOutputStream(new NonClosingOutputStreamDecorator(checkpointOutputStream))) { dos.writeInt(CURRENT_VERSION); dos.writeBoolean(readerPosition != null); if (readerPosition != null) { diff --git a/flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/OperatorUtils.java b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/OperatorUtils.java index 25d200b..8e36a7b 100644 --- a/flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/OperatorUtils.java +++ b/flink-ml-iteration/src/main/java/org/apache/flink/iteration/operator/OperatorUtils.java @@ -37,6 +37,7 @@ import org.apache.flink.util.function.SupplierWithException; import org.apache.flink.util.function.ThrowingConsumer; import java.io.IOException; +import java.nio.file.Paths; import java.util.Arrays; import java.util.Random; import java.util.UUID; @@ -110,9 +111,9 @@ public class OperatorUtils { String pathStr = configuration.get(IterationOptions.DATA_CACHE_PATH); if (pathStr == null) { Random random = new Random(); - pathStr = "file://" + localSpillPaths[random.nextInt(localSpillPaths.length)]; + final String localSpillPath = localSpillPaths[random.nextInt(localSpillPaths.length)]; + pathStr = Paths.get(localSpillPath).toUri().toString(); } - return new Path(pathStr); } diff --git a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/HeadOperatorTest.java b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/HeadOperatorTest.java index c1fe602..1777893 100644 --- a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/HeadOperatorTest.java +++ b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/HeadOperatorTest.java @@ -33,6 +33,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointType; import org.apache.flink.runtime.checkpoint.TaskStateSnapshot; import org.apache.flink.runtime.io.network.api.CheckpointBarrier; import org.apache.flink.runtime.io.network.api.EndOfData; +import org.apache.flink.runtime.io.network.api.StopMode; import org.apache.flink.runtime.jobgraph.OperatorID; import org.apache.flink.runtime.operators.coordination.OperatorEvent; import org.apache.flink.runtime.operators.coordination.OperatorEventGateway; @@ -187,7 +188,7 @@ public class HeadOperatorTest extends TestLogger { }); // Mark the input as finished. - harness.processEvent(EndOfData.INSTANCE); + harness.processEvent(new EndOfData(StopMode.DRAIN)); // There should be no exception taskExecuteResult.get(); @@ -703,7 +704,7 @@ public class HeadOperatorTest extends TestLogger { iterationId, IterationRecord.newEpochWatermark(Integer.MAX_VALUE + 1, "tail"), null); - harness.processEvent(EndOfData.INSTANCE); + harness.processEvent(new EndOfData(StopMode.DRAIN)); harness.finishProcessing(); return null; diff --git a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/ReplayOperatorTest.java b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/ReplayOperatorTest.java index 636633f..49bd045 100644 --- a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/ReplayOperatorTest.java +++ b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/ReplayOperatorTest.java @@ -278,7 +278,7 @@ public class ReplayOperatorTest extends TestLogger { .getConfiguration() .set( IterationOptions.DATA_CACHE_PATH, - "file://" + tempFolder.newFolder().getAbsolutePath()); + tempFolder.newFolder().toPath().toAbsolutePath().toUri().toString()); harness.getStreamTask().restore(); return runnable.apply(harness); } diff --git a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/allround/MultipleInputAllRoundWrapperOperatorTest.java b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/allround/MultipleInputAllRoundWrapperOperatorTest.java index 1eb5734..e07b5ae 100644 --- a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/allround/MultipleInputAllRoundWrapperOperatorTest.java +++ b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/allround/MultipleInputAllRoundWrapperOperatorTest.java @@ -29,6 +29,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder; import org.apache.flink.runtime.checkpoint.CheckpointOptions; import org.apache.flink.runtime.checkpoint.CheckpointType; import org.apache.flink.runtime.io.network.api.EndOfData; +import org.apache.flink.runtime.io.network.api.StopMode; import org.apache.flink.runtime.jobgraph.OperatorID; import org.apache.flink.runtime.state.CheckpointStorageLocationReference; import org.apache.flink.runtime.state.StateInitializationContext; @@ -134,9 +135,9 @@ public class MultipleInputAllRoundWrapperOperatorTest extends TestLogger { harness.getStreamTask().notifyCheckpointAbortAsync(6, 5); harness.processAll(); - harness.processEvent(EndOfData.INSTANCE, 0); - harness.processEvent(EndOfData.INSTANCE, 1); - harness.processEvent(EndOfData.INSTANCE, 2); + harness.processEvent(new EndOfData(StopMode.DRAIN), 0); + harness.processEvent(new EndOfData(StopMode.DRAIN), 1); + harness.processEvent(new EndOfData(StopMode.DRAIN), 2); harness.endInput(); harness.finishProcessing(); diff --git a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/allround/OneInputAllRoundWrapperOperatorTest.java b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/allround/OneInputAllRoundWrapperOperatorTest.java index fd75d65..7b9d402 100644 --- a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/allround/OneInputAllRoundWrapperOperatorTest.java +++ b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/allround/OneInputAllRoundWrapperOperatorTest.java @@ -31,6 +31,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointOptions; import org.apache.flink.runtime.checkpoint.CheckpointType; import org.apache.flink.runtime.checkpoint.TaskStateSnapshot; import org.apache.flink.runtime.io.network.api.EndOfData; +import org.apache.flink.runtime.io.network.api.StopMode; import org.apache.flink.runtime.jobgraph.OperatorID; import org.apache.flink.runtime.state.CheckpointStorageLocationReference; import org.apache.flink.runtime.state.StateInitializationContext; @@ -121,7 +122,7 @@ public class OneInputAllRoundWrapperOperatorTest extends TestLogger { harness.getStreamTask().notifyCheckpointAbortAsync(6, 5); harness.processAll(); - harness.processEvent(EndOfData.INSTANCE, 0); + harness.processEvent(new EndOfData(StopMode.DRAIN), 0); harness.endInput(); harness.finishProcessing(); diff --git a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/allround/TwoInputAllRoundWrapperOperatorTest.java b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/allround/TwoInputAllRoundWrapperOperatorTest.java index 5930e8b..b8aed71 100644 --- a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/allround/TwoInputAllRoundWrapperOperatorTest.java +++ b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/allround/TwoInputAllRoundWrapperOperatorTest.java @@ -29,6 +29,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder; import org.apache.flink.runtime.checkpoint.CheckpointOptions; import org.apache.flink.runtime.checkpoint.CheckpointType; import org.apache.flink.runtime.io.network.api.EndOfData; +import org.apache.flink.runtime.io.network.api.StopMode; import org.apache.flink.runtime.jobgraph.OperatorID; import org.apache.flink.runtime.state.CheckpointStorageLocationReference; import org.apache.flink.runtime.state.StateInitializationContext; @@ -124,8 +125,8 @@ public class TwoInputAllRoundWrapperOperatorTest extends TestLogger { harness.getStreamTask().notifyCheckpointAbortAsync(6, 5); harness.processAll(); - harness.processEvent(EndOfData.INSTANCE, 0); - harness.processEvent(EndOfData.INSTANCE, 1); + harness.processEvent(new EndOfData(StopMode.DRAIN), 0); + harness.processEvent(new EndOfData(StopMode.DRAIN), 1); harness.endInput(); harness.finishProcessing(); diff --git a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/MultipleInputPerRoundWrapperOperatorTest.java b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/MultipleInputPerRoundWrapperOperatorTest.java index 59685a9..c58ded8 100644 --- a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/MultipleInputPerRoundWrapperOperatorTest.java +++ b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/MultipleInputPerRoundWrapperOperatorTest.java @@ -31,6 +31,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder; import org.apache.flink.runtime.checkpoint.CheckpointOptions; import org.apache.flink.runtime.checkpoint.CheckpointType; import org.apache.flink.runtime.io.network.api.EndOfData; +import org.apache.flink.runtime.io.network.api.StopMode; import org.apache.flink.runtime.jobgraph.OperatorID; import org.apache.flink.runtime.state.CheckpointStorageLocationReference; import org.apache.flink.runtime.state.StateInitializationContext; @@ -141,9 +142,9 @@ public class MultipleInputPerRoundWrapperOperatorTest extends TestLogger { OperatorUtils.getUniqueSenderId(operatorId, 0)))), new ArrayList<>(harness.getOutput())); - harness.processEvent(EndOfData.INSTANCE, 0); - harness.processEvent(EndOfData.INSTANCE, 1); - harness.processEvent(EndOfData.INSTANCE, 2); + harness.processEvent(new EndOfData(StopMode.DRAIN), 0); + harness.processEvent(new EndOfData(StopMode.DRAIN), 1); + harness.processEvent(new EndOfData(StopMode.DRAIN), 2); harness.endInput(); harness.finishProcessing(); diff --git a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/OneInputPerRoundWrapperOperatorTest.java b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/OneInputPerRoundWrapperOperatorTest.java index 43460a7..384ef91 100644 --- a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/OneInputPerRoundWrapperOperatorTest.java +++ b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/OneInputPerRoundWrapperOperatorTest.java @@ -35,6 +35,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointOptions; import org.apache.flink.runtime.checkpoint.CheckpointType; import org.apache.flink.runtime.checkpoint.TaskStateSnapshot; import org.apache.flink.runtime.io.network.api.EndOfData; +import org.apache.flink.runtime.io.network.api.StopMode; import org.apache.flink.runtime.jobgraph.OperatorID; import org.apache.flink.runtime.state.CheckpointStorageLocationReference; import org.apache.flink.runtime.state.OperatorStateCheckpointOutputStream; @@ -142,7 +143,7 @@ public class OneInputPerRoundWrapperOperatorTest extends TestLogger { OperatorUtils.getUniqueSenderId(operatorId, 0)))), new ArrayList<>(harness.getOutput())); - harness.processEvent(EndOfData.INSTANCE, 0); + harness.processEvent(new EndOfData(StopMode.DRAIN), 0); harness.endInput(); harness.finishProcessing(); diff --git a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/TwoInputPerRoundWrapperOperatorTest.java b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/TwoInputPerRoundWrapperOperatorTest.java index 6e08e57..26a2797 100644 --- a/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/TwoInputPerRoundWrapperOperatorTest.java +++ b/flink-ml-iteration/src/test/java/org/apache/flink/iteration/operator/perround/TwoInputPerRoundWrapperOperatorTest.java @@ -30,6 +30,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointMetricsBuilder; import org.apache.flink.runtime.checkpoint.CheckpointOptions; import org.apache.flink.runtime.checkpoint.CheckpointType; import org.apache.flink.runtime.io.network.api.EndOfData; +import org.apache.flink.runtime.io.network.api.StopMode; import org.apache.flink.runtime.jobgraph.OperatorID; import org.apache.flink.runtime.state.CheckpointStorageLocationReference; import org.apache.flink.runtime.state.StateInitializationContext; @@ -132,8 +133,8 @@ public class TwoInputPerRoundWrapperOperatorTest extends TestLogger { OperatorUtils.getUniqueSenderId(operatorId, 0)))), new ArrayList<>(harness.getOutput())); - harness.processEvent(EndOfData.INSTANCE, 0); - harness.processEvent(EndOfData.INSTANCE, 1); + harness.processEvent(new EndOfData(StopMode.DRAIN), 0); + harness.processEvent(new EndOfData(StopMode.DRAIN), 1); harness.endInput(); harness.finishProcessing(); diff --git a/flink-ml-lib/pom.xml b/flink-ml-lib/pom.xml index 5ab227d..6633e2f 100644 --- a/flink-ml-lib/pom.xml +++ b/flink-ml-lib/pom.xml @@ -26,20 +26,20 @@ under the License. <version>2.1-SNAPSHOT</version> </parent> - <artifactId>flink-ml-lib_${scala.binary.version}</artifactId> + <artifactId>flink-ml-lib</artifactId> <name>Flink ML : Lib</name> <dependencies> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-ml-core_${scala.binary.version}</artifactId> + <artifactId>flink-ml-core</artifactId> <version>${project.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-ml-iteration_${scala.binary.version}</artifactId> + <artifactId>flink-ml-iteration</artifactId> <version>${project.version}</version> <scope>provided</scope> </dependency> @@ -53,35 +53,33 @@ under the License. <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId> + <artifactId>flink-table-api-java-bridge</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-table-planner_${scala.binary.version}</artifactId> - <version>${flink.version}</version> + <artifactId>flink-table-planner-loader</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-table-runtime_${scala.binary.version}</artifactId> - <version>${flink.version}</version> + <artifactId>flink-table-runtime</artifactId> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-test-utils_${scala.binary.version}</artifactId> + <artifactId>flink-test-utils</artifactId> <version>${flink.version}</version> <scope>test</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-streaming-java_${scala.binary.version}</artifactId> + <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>test</scope> <type>test-jar</type> diff --git a/flink-ml-lib/src/main/java/org/apache/flink/ml/evaluation/binaryclassification/BinaryClassificationEvaluator.java b/flink-ml-lib/src/main/java/org/apache/flink/ml/evaluation/binaryclassification/BinaryClassificationEvaluator.java index 86cf75a..49cf4ae 100644 --- a/flink-ml-lib/src/main/java/org/apache/flink/ml/evaluation/binaryclassification/BinaryClassificationEvaluator.java +++ b/flink-ml-lib/src/main/java/org/apache/flink/ml/evaluation/binaryclassification/BinaryClassificationEvaluator.java @@ -27,10 +27,10 @@ import org.apache.flink.api.common.functions.RichMapPartitionFunction; 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.tuple.Tuple3; import org.apache.flink.api.java.tuple.Tuple4; import org.apache.flink.api.java.typeutils.RowTypeInfo; -import org.apache.flink.api.scala.typeutils.Types; import org.apache.flink.iteration.operator.OperatorStateUtils; import org.apache.flink.ml.api.AlgoOperator; import org.apache.flink.ml.common.broadcast.BroadcastUtils; @@ -200,7 +200,7 @@ public class BinaryClassificationEvaluator final String[] metricsNames = getMetricsNames(); TypeInformation<?>[] metricTypes = new TypeInformation[metricsNames.length]; - Arrays.fill(metricTypes, Types.DOUBLE()); + Arrays.fill(metricTypes, Types.DOUBLE); RowTypeInfo outputTypeInfo = new RowTypeInfo(metricTypes, metricsNames); DataStream<Row> evalResult = diff --git a/flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/OnlineKMeansTest.java b/flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/OnlineKMeansTest.java index 047e8ea..cd4af68 100644 --- a/flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/OnlineKMeansTest.java +++ b/flink-ml-lib/src/test/java/org/apache/flink/ml/clustering/OnlineKMeansTest.java @@ -18,6 +18,8 @@ package org.apache.flink.ml.clustering; +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.configuration.Configuration; @@ -34,6 +36,7 @@ import org.apache.flink.ml.linalg.Vectors; import org.apache.flink.ml.linalg.typeinfo.DenseVectorTypeInfo; import org.apache.flink.ml.util.InMemorySinkFunction; import org.apache.flink.ml.util.InMemorySourceFunction; +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; @@ -58,6 +61,9 @@ import java.util.Comparator; import java.util.HashSet; import java.util.List; import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import static org.apache.flink.ml.clustering.KMeansTest.groupFeaturesByPrediction; import static org.junit.Assert.assertEquals; @@ -195,24 +201,24 @@ public class OnlineKMeansTest extends TestLogger { } /** Blocks the thread until Model has set up init model data. */ - private void waitInitModelDataSetup() throws InterruptedException { - while (reporter.findMetrics(OnlineKMeansModel.MODEL_DATA_VERSION_GAUGE_KEY).size() + private void waitInitModelDataSetup(JobID jobID) throws InterruptedException { + while (reporter.findMetrics(jobID, OnlineKMeansModel.MODEL_DATA_VERSION_GAUGE_KEY).size() < defaultParallelism) { Thread.sleep(100); } - waitModelDataUpdate(); + waitModelDataUpdate(jobID); } /** Blocks the thread until the Model has received the next model-data-update event. */ @SuppressWarnings("unchecked") - private void waitModelDataUpdate() throws InterruptedException { + private void waitModelDataUpdate(JobID jobID) throws InterruptedException { do { int tmpModelDataVersion = - reporter.findMetrics(OnlineKMeansModel.MODEL_DATA_VERSION_GAUGE_KEY).values() - .stream() + reporter.findMetrics(jobID, OnlineKMeansModel.MODEL_DATA_VERSION_GAUGE_KEY) + .values().stream() .map(x -> Integer.parseInt(((Gauge<String>) x).getValue())) .min(Integer::compareTo) - .get(); + .orElse(0); if (tmpModelDataVersion == currentModelDataVersion) { Thread.sleep(100); } else { @@ -240,6 +246,14 @@ public class OnlineKMeansTest extends TestLogger { Assert.assertTrue(CollectionUtils.isEqualCollection(expectedGroups, actualGroups)); } + private JobID submitJob(JobGraph jobGraph) + throws ExecutionException, InterruptedException, TimeoutException { + return miniCluster + .submitJob(jobGraph) + .thenApply(JobSubmissionResult::getJobID) + .get(1, TimeUnit.SECONDS); + } + @Test public void testParam() { OnlineKMeans onlineKMeans = new OnlineKMeans(); @@ -279,16 +293,16 @@ public class OnlineKMeansTest extends TestLogger { OnlineKMeansModel onlineModel = onlineKMeans.fit(onlineTrainTable); transformAndOutputData(onlineModel); - miniCluster.submitJob(env.getStreamGraph().getJobGraph()); - waitInitModelDataSetup(); + final JobID jobID = submitJob(env.getStreamGraph().getJobGraph()); + waitInitModelDataSetup(jobID); trainSource.addAll(trainData1); - waitModelDataUpdate(); + waitModelDataUpdate(jobID); predictAndAssert( expectedGroups1, onlineKMeans.getFeaturesCol(), onlineKMeans.getPredictionCol()); trainSource.addAll(trainData2); - waitModelDataUpdate(); + waitModelDataUpdate(jobID); predictAndAssert( expectedGroups2, onlineKMeans.getFeaturesCol(), onlineKMeans.getPredictionCol()); } @@ -308,13 +322,13 @@ public class OnlineKMeansTest extends TestLogger { OnlineKMeansModel onlineModel = onlineKMeans.fit(onlineTrainTable); transformAndOutputData(onlineModel); - miniCluster.submitJob(env.getStreamGraph().getJobGraph()); - waitInitModelDataSetup(); + final JobID jobID = submitJob(env.getStreamGraph().getJobGraph()); + waitInitModelDataSetup(jobID); predictAndAssert( expectedGroups1, onlineKMeans.getFeaturesCol(), onlineKMeans.getPredictionCol()); trainSource.addAll(trainData2); - waitModelDataUpdate(); + waitModelDataUpdate(jobID); predictAndAssert( expectedGroups2, onlineKMeans.getFeaturesCol(), onlineKMeans.getPredictionCol()); } @@ -334,7 +348,7 @@ public class OnlineKMeansTest extends TestLogger { OnlineKMeansModel onlineModel = onlineKMeans.fit(onlineTrainTable); transformAndOutputData(onlineModel); - miniCluster.submitJob(env.getStreamGraph().getJobGraph()); + submitJob(env.getStreamGraph().getJobGraph()); modelDataSink.poll(); trainSource.addAll(trainData2); @@ -395,7 +409,8 @@ public class OnlineKMeansTest extends TestLogger { String savePath = tempFolder.newFolder().getAbsolutePath(); onlineKMeans.save(savePath); - miniCluster.executeJobBlocking(env.getStreamGraph().getJobGraph()); + final JobGraph jobGraph = env.getStreamGraph().getJobGraph(); + miniCluster.executeJobBlocking(jobGraph); OnlineKMeans loadedOnlineKMeans = OnlineKMeans.load(tEnv, savePath); OnlineKMeansModel onlineModel = loadedOnlineKMeans.fit(onlineTrainTable); @@ -407,13 +422,13 @@ public class OnlineKMeansTest extends TestLogger { transformAndOutputData(loadedOnlineModel); - miniCluster.submitJob(env.getStreamGraph().getJobGraph()); - waitInitModelDataSetup(); + final JobID jobID = submitJob(env.getStreamGraph().getJobGraph()); + waitInitModelDataSetup(jobID); predictAndAssert( expectedGroups1, onlineKMeans.getFeaturesCol(), onlineKMeans.getPredictionCol()); trainSource.addAll(trainData2); - waitModelDataUpdate(); + waitModelDataUpdate(jobID); predictAndAssert( expectedGroups2, onlineKMeans.getFeaturesCol(), onlineKMeans.getPredictionCol()); } @@ -430,7 +445,7 @@ public class OnlineKMeansTest extends TestLogger { OnlineKMeansModel onlineModel = onlineKMeans.fit(onlineTrainTable); transformAndOutputData(onlineModel); - miniCluster.submitJob(env.getStreamGraph().getJobGraph()); + submitJob(env.getStreamGraph().getJobGraph()); modelDataSink.poll(); trainSource.addAll(trainData1); @@ -479,15 +494,15 @@ public class OnlineKMeansTest extends TestLogger { .setPredictionCol("prediction"); transformAndOutputData(onlineModel); - miniCluster.submitJob(env.getStreamGraph().getJobGraph()); + final JobID jobID = submitJob(env.getStreamGraph().getJobGraph()); modelDataSource.addAll(modelData1); - waitInitModelDataSetup(); + waitInitModelDataSetup(jobID); predictAndAssert( expectedGroups1, onlineModel.getFeaturesCol(), onlineModel.getPredictionCol()); modelDataSource.addAll(modelData2); - waitModelDataUpdate(); + waitModelDataUpdate(jobID); predictAndAssert( expectedGroups2, onlineModel.getFeaturesCol(), onlineModel.getPredictionCol()); } diff --git a/flink-ml-python/dev/dev-requirements.txt b/flink-ml-python/dev/dev-requirements.txt index 30024d9..1f30ffb 100755 --- a/flink-ml-python/dev/dev-requirements.txt +++ b/flink-ml-python/dev/dev-requirements.txt @@ -14,7 +14,7 @@ # limitations under the License. setuptools>=18.0 wheel -apache-flink==1.14.4 +apache-flink==1.15.0 pandas>=1.0,<1.2.0 jsonpickle==2.0.0 cloudpickle==1.2.2 diff --git a/flink-ml-python/pyflink/ml/lib/tests/test_ml_lib_completeness.py b/flink-ml-python/pyflink/ml/lib/tests/test_ml_lib_completeness.py index 045a274..ff17298 100644 --- a/flink-ml-python/pyflink/ml/lib/tests/test_ml_lib_completeness.py +++ b/flink-ml-python/pyflink/ml/lib/tests/test_ml_lib_completeness.py @@ -49,7 +49,7 @@ class MLLibTest(PyFlinkMLTestCase): this_directory, "../../../../../flink-ml-lib")) ml_lib_jar = glob.glob(os.path.join( - FLINK_ML_LIB_SOURCE_PATH, "target", "flink-ml-lib_*-SNAPSHOT.jar"))[0] + FLINK_ML_LIB_SOURCE_PATH, "target", "flink-ml-lib-*SNAPSHOT.jar"))[0] StageAnalyzer = get_gateway().jvm.org.apache.flink.ml.util.StageAnalyzer return sorted([stage for stage in StageAnalyzer.analyzeLibJars(ml_lib_jar) diff --git a/flink-ml-python/pyflink/ml/tests/test_utils.py b/flink-ml-python/pyflink/ml/tests/test_utils.py index 2841b03..7cea8e7 100644 --- a/flink-ml-python/pyflink/ml/tests/test_utils.py +++ b/flink-ml-python/pyflink/ml/tests/test_utils.py @@ -67,6 +67,6 @@ class PyFlinkMLTestCase(unittest.TestCase): this_directory, "../../../../flink-ml-lib")) ml_test_jar = glob.glob(os.path.join( - FLINK_ML_LIB_SOURCE_PATH, "target", "flink-ml-lib_*-tests.jar"))[0] + FLINK_ML_LIB_SOURCE_PATH, "target", "flink-ml-lib-*-tests.jar"))[0] self.env.add_classpaths("file://{0}".format(ml_test_jar)) diff --git a/flink-ml-python/setup.py b/flink-ml-python/setup.py index 14f7314..55b617e 100644 --- a/flink-ml-python/setup.py +++ b/flink-ml-python/setup.py @@ -103,7 +103,7 @@ try: author='Apache Software Foundation', author_email='[email protected]', python_requires='>=3.6', - install_requires=['apache-flink==1.14.4', 'pandas>=1.0,<1.2.0', 'jsonpickle==2.0.0', + install_requires=['apache-flink==1.15.0', 'pandas>=1.0,<1.2.0', 'jsonpickle==2.0.0', 'cloudpickle==1.2.2', 'numpy>=1.14.3,<1.20'], tests_require=['pytest==4.4.1'], description='Apache Flink ML Python API', diff --git a/flink-ml-tests/pom.xml b/flink-ml-tests/pom.xml index 692c705..2401e60 100644 --- a/flink-ml-tests/pom.xml +++ b/flink-ml-tests/pom.xml @@ -28,28 +28,28 @@ under the License. <version>2.1-SNAPSHOT</version> </parent> - <artifactId>flink-ml-tests_${scala.binary.version}</artifactId> + <artifactId>flink-ml-tests</artifactId> <name>Flink ML : Tests</name> <dependencies> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-ml-iteration_${scala.binary.version}</artifactId> + <artifactId>flink-ml-iteration</artifactId> <version>${project.version}</version> <scope>test</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-ml-core_${scala.binary.version}</artifactId> + <artifactId>flink-ml-core</artifactId> <version>${project.version}</version> <scope>test</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-streaming-java_${scala.binary.version}</artifactId> + <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>test</scope> </dependency> diff --git a/flink-ml-uber/pom.xml b/flink-ml-uber/pom.xml index 95277b2..3680135 100644 --- a/flink-ml-uber/pom.xml +++ b/flink-ml-uber/pom.xml @@ -26,7 +26,7 @@ under the License. <version>2.1-SNAPSHOT</version> </parent> - <artifactId>flink-ml-uber_${scala.binary.version}</artifactId> + <artifactId>flink-ml-uber</artifactId> <name>Flink ML : Uber</name> <description> This module contains the basic modules for writing Flink ML programs. @@ -37,25 +37,25 @@ under the License. <dependencies> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-ml-core_${scala.binary.version}</artifactId> + <artifactId>flink-ml-core</artifactId> <version>${project.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-ml-iteration_${scala.binary.version}</artifactId> + <artifactId>flink-ml-iteration</artifactId> <version>${project.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-ml-lib_${scala.binary.version}</artifactId> + <artifactId>flink-ml-lib</artifactId> <version>${project.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> - <artifactId>flink-ml-benchmark_${scala.binary.version}</artifactId> + <artifactId>flink-ml-benchmark</artifactId> <version>${project.version}</version> </dependency> </dependencies> @@ -75,10 +75,10 @@ under the License. <configuration> <artifactSet> <includes combine.children="append"> - <include>org.apache.flink:flink-ml-core_${scala.binary.version}</include> - <include>org.apache.flink:flink-ml-iteration_${scala.binary.version}</include> - <include>org.apache.flink:flink-ml-lib_${scala.binary.version}</include> - <include>org.apache.flink:flink-ml-benchmark_${scala.binary.version}</include> + <include>org.apache.flink:flink-ml-core</include> + <include>org.apache.flink:flink-ml-iteration</include> + <include>org.apache.flink:flink-ml-lib</include> + <include>org.apache.flink:flink-ml-benchmark</include> <include>dev.ludovic.netlib:blas</include> </includes> </artifactSet> diff --git a/pom.xml b/pom.xml index a6b7128..116b205 100644 --- a/pom.xml +++ b/pom.xml @@ -62,19 +62,20 @@ under the License. </modules> <properties> - <flink.shaded.version>12.0</flink.shaded.version> - <scala.binary.version>2.12</scala.binary.version> - <jackson.version>2.10.1</jackson.version> + <flink.shaded.version>15.0</flink.shaded.version> + <jackson.version>2.12.4</jackson.version> + <jackson-databind.version>2.12.4</jackson-databind.version> <target.java.version>1.8</target.java.version> <spotless.version>2.4.2</spotless.version> - <slf4j.version>1.7.15</slf4j.version> - <log4j.version>2.17.1</log4j.version> - <junit.version>4.13.1</junit.version> + <slf4j.version>1.7.36</slf4j.version> + <statefun.version>3.2.0</statefun.version> + <log4j.version>2.17.2</log4j.version> + <junit.version>4.13.2</junit.version> <flink.forkCount>1C</flink.forkCount> <flink.reuseForks>true</flink.reuseForks> - <flink.version>1.14.4</flink.version> - <zookeeper.version>3.4.14</zookeeper.version> - <hadoop.version>2.8.5</hadoop.version> + <flink.version>1.15.0</flink.version> + <zookeeper.version>3.6.3</zookeeper.version> + <hadoop.version>2.10.1</hadoop.version> <!-- Can be set to any value to reproduce a specific build. --> <test.randomization.seed/> @@ -140,6 +141,11 @@ under the License. <artifactId>flink-shaded-zookeeper-3</artifactId> <version>${zookeeper.version}-${flink.shaded.version}</version> </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-files</artifactId> + <version>${flink.version}</version> + </dependency> <dependency> <groupId>org.apache.zookeeper</groupId> @@ -210,6 +216,186 @@ under the License. <version>${flink.version}</version> <scope>test</scope> </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-runtime</artifactId> + <version>${flink.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-table-planner-loader</artifactId> + <version>${flink.version}</version> + <scope>test</scope> + </dependency> + <!-- hdfs is required for the data cache test --> + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-hdfs</artifactId> + <scope>test</scope> + <version>${hadoop.version}</version> + <exclusions> + <exclusion> + <groupId>log4j</groupId> + <artifactId>log4j</artifactId> + </exclusion> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-annotations</artifactId> + </exclusion> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-core</artifactId> + </exclusion> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-databind</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-common</artifactId> + <scope>test</scope> + <version>${hadoop.version}</version> + <exclusions> + <exclusion> + <groupId>log4j</groupId> + <artifactId>log4j</artifactId> + </exclusion> + <exclusion> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-log4j12</artifactId> + </exclusion> + <exclusion> + <!-- This dependency is no longer shipped with the JDK since Java 9.--> + <groupId>jdk.tools</groupId> + <artifactId>jdk.tools</artifactId> + </exclusion> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-annotations</artifactId> + </exclusion> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-core</artifactId> + </exclusion> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-databind</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-hdfs</artifactId> + <scope>test</scope> + <type>test-jar</type> + <version>${hadoop.version}</version> + <exclusions> + <exclusion> + <groupId>log4j</groupId> + <artifactId>log4j</artifactId> + </exclusion> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-annotations</artifactId> + </exclusion> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-core</artifactId> + </exclusion> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-databind</artifactId> + </exclusion> + </exclusions> + </dependency> + + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-common</artifactId> + <scope>test</scope> + <type>test-jar</type> + <version>${hadoop.version}</version> + <exclusions> + <exclusion> + <groupId>log4j</groupId> + <artifactId>log4j</artifactId> + </exclusion> + <exclusion> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-log4j12</artifactId> + </exclusion> + <exclusion> + <!-- This dependency is no longer shipped with the JDK since Java 9.--> + <groupId>jdk.tools</groupId> + <artifactId>jdk.tools</artifactId> + </exclusion> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-annotations</artifactId> + </exclusion> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-core</artifactId> + </exclusion> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-databind</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>org.apache.hadoop</groupId> + <artifactId>hadoop-minicluster</artifactId> + <scope>test</scope> + <version>${hadoop.version}</version> + <exclusions> + <exclusion> + <groupId>log4j</groupId> + <artifactId>log4j</artifactId> + </exclusion> + <exclusion> + <groupId>org.slf4j</groupId> + <artifactId>slf4j-log4j12</artifactId> + </exclusion> + <exclusion> + <!-- This dependency is no longer shipped with the JDK since Java 9.--> + <groupId>jdk.tools</groupId> + <artifactId>jdk.tools</artifactId> + </exclusion> + + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-annotations</artifactId> + </exclusion> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-core</artifactId> + </exclusion> + <exclusion> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-databind</artifactId> + </exclusion> + </exclusions> + </dependency> + + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-databind</artifactId> + <version>${jackson-databind.version}</version> + </dependency> + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-core</artifactId> + <version>${jackson.version}</version> + </dependency> + <dependency> + <groupId>com.fasterxml.jackson.core</groupId> + <artifactId>jackson-annotations</artifactId> + <version>${jackson.version}</version> + </dependency> </dependencies> </dependencyManagement>
