You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2022/05/26 12:57:49 UTC
[flink-ml] branch master updated (b1aa1d7 -> c8fef60)
This is an automated email from the ASF dual-hosted git repository.
lindong pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink-ml.git
from b1aa1d7 [FLINK-27294][python][ml] Add evaluation algorithm support for BinaryClassificationEvaluator in ML Python API
new f932783 [FLINK-27798] Migrated to Flink 1.15.0.
new c8fef60 [FLINK-27798] Fix python test for Flink 1.15 migration.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
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 +-
.../pyflink/ml/core/tests/test_pipeline.py | 23 ++-
.../ml/lib/feature/tests/test_vectorassembler.py | 22 ++-
.../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 ++++++++++++++++++++-
32 files changed, 410 insertions(+), 237 deletions(-)
[flink-ml] 01/02: [FLINK-27798] Migrated to Flink 1.15.0.
Posted by li...@apache.org.
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 <ma...@dotbg.name>
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='dev@flink.apache.org',
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>
[flink-ml] 02/02: [FLINK-27798] Fix python test for Flink 1.15 migration.
Posted by li...@apache.org.
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 c8fef6054f83c1dc1dc188606b61bd5a92c60c57
Author: yunfengzhou-hub <yu...@outlook.com>
AuthorDate: Thu May 26 20:50:20 2022 +0800
[FLINK-27798] Fix python test for Flink 1.15 migration.
This closes #104.
---
.../pyflink/ml/core/tests/test_pipeline.py | 23 ++++++++++++----------
.../ml/lib/feature/tests/test_vectorassembler.py | 22 +++++++++++++++------
2 files changed, 29 insertions(+), 16 deletions(-)
diff --git a/flink-ml-python/pyflink/ml/core/tests/test_pipeline.py b/flink-ml-python/pyflink/ml/core/tests/test_pipeline.py
index ac5e662..754c9ee 100644
--- a/flink-ml-python/pyflink/ml/core/tests/test_pipeline.py
+++ b/flink-ml-python/pyflink/ml/core/tests/test_pipeline.py
@@ -18,8 +18,6 @@
import os
from typing import Dict, Any, List
-import pandas as pd
-from pandas._testing import assert_frame_equal
from pyflink.table import Table, StreamTableEnvironment
from pyflink.ml.core.api import Model
@@ -39,8 +37,9 @@ class PipelineTest(PyFlinkMLTestCase):
model = PipelineModel([model_a, model_b, model_c])
output_table = model.transform(input_table)[0]
- assert_frame_equal(output_table.to_pandas(),
- pd.DataFrame([[31], [32], [33]], columns=['a']))
+ predicted_results = [result[0] for result in
+ self.t_env.to_data_stream(output_table).execute_and_collect()]
+ self.assertEqual(predicted_results, [31, 32, 33])
# Saves and loads the PipelineModel.
path = os.path.join(self.temp_dir, "test_pipeline_model")
@@ -48,8 +47,9 @@ class PipelineTest(PyFlinkMLTestCase):
loaded_model = PipelineModel.load(self.t_env, path)
output_table2 = loaded_model.transform(input_table)[0]
- assert_frame_equal(output_table2.to_pandas(),
- pd.DataFrame([[31], [32], [33]], columns=['a']))
+ predicted_results = [result[0] for result in
+ self.t_env.to_data_stream(output_table2).execute_and_collect()]
+ self.assertEqual(predicted_results, [31, 32, 33])
def test_pipeline(self):
input_table = self.t_env.from_elements([(1,), (2,), (3,)], ['a'])
@@ -60,8 +60,9 @@ class PipelineTest(PyFlinkMLTestCase):
model = estimator.fit(input_table)
output_table = model.transform(input_table)[0]
- assert_frame_equal(output_table.to_pandas(),
- pd.DataFrame([[21], [22], [23]], columns=['a']))
+ predicted_results = [result[0] for result in
+ self.t_env.to_data_stream(output_table).execute_and_collect()]
+ self.assertEqual(predicted_results, [21, 22, 23])
# Saves and loads the PipelineModel.
path = os.path.join(self.temp_dir, "test_pipeline")
@@ -70,8 +71,10 @@ class PipelineTest(PyFlinkMLTestCase):
model = loaded_estimator.fit(input_table)
output_table = model.transform(input_table)[0]
- assert_frame_equal(output_table.to_pandas(),
- pd.DataFrame([[21], [22], [23]], columns=['a']))
+
+ predicted_results = [result[0] for result in
+ self.t_env.to_data_stream(output_table).execute_and_collect()]
+ self.assertEqual(predicted_results, [21, 22, 23])
class Add10Model(Model):
diff --git a/flink-ml-python/pyflink/ml/lib/feature/tests/test_vectorassembler.py b/flink-ml-python/pyflink/ml/lib/feature/tests/test_vectorassembler.py
index c72e6dd..5362fcf 100644
--- a/flink-ml-python/pyflink/ml/lib/feature/tests/test_vectorassembler.py
+++ b/flink-ml-python/pyflink/ml/lib/feature/tests/test_vectorassembler.py
@@ -15,6 +15,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
+import os
from pyflink.common import Types
@@ -26,6 +27,7 @@ from pyflink.ml.tests.test_utils import PyFlinkMLTestCase
class VectorAssemblerTest(PyFlinkMLTestCase):
def setUp(self):
super(VectorAssemblerTest, self).setUp()
+ # TODO: Add test for handling invalid values after FLINK-27797 is resolved.
self.input_data_table = self.t_env.from_data_stream(
self.env.from_collection([
(0,
@@ -37,7 +39,6 @@ class VectorAssemblerTest(PyFlinkMLTestCase):
1.0,
Vectors.sparse(5, [1, 2, 3, 4],
[1.0, 2.0, 3.0, 4.0])),
- (2, None, None, None),
],
type_info=Types.ROW_NAMED(
['id', 'vec', 'num', 'sparse_vec'],
@@ -60,13 +61,22 @@ class VectorAssemblerTest(PyFlinkMLTestCase):
self.assertEqual('skip', vector_assembler.handle_invalid)
self.assertEqual('assembled_vec', vector_assembler.output_col)
- def test_keep_invalid(self):
+ def test_save_load_transform(self):
vector_assembler = VectorAssembler() \
.set_input_cols('vec', 'num', 'sparse_vec') \
.set_output_col('assembled_vec') \
.set_handle_invalid('keep')
- output = vector_assembler.transform(self.input_data_table)[0]
- self.assertEqual(
- ['id', 'vec', 'num', 'sparse_vec', 'assembled_vec'],
- output.get_schema().get_field_names())
+ path = os.path.join(self.temp_dir, 'test_save_load_transform_vector_assembler')
+ vector_assembler.save(path)
+ vector_assembler = VectorAssembler.load(self.t_env, path)
+
+ output_table = vector_assembler.transform(self.input_data_table)[0]
+ actual_outputs = [(result[0], result[4]) for result in
+ self.t_env.to_data_stream(output_table).execute_and_collect()]
+
+ for actual_output in actual_outputs:
+ if actual_output[0] == 0:
+ self.assertEqual(self.expected_output_data_1, actual_output[1])
+ else:
+ self.assertEqual(self.expected_output_data_2, actual_output[1])