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
The following commit(s) were added to refs/heads/master by this push:
new 1a7337a [Flink 26443] Add benchmark framework
1a7337a is described below
commit 1a7337ac49da531c29b9d210b5b34174d6afa623
Author: yunfengzhou-hub <[email protected]>
AuthorDate: Wed Apr 6 22:38:00 2022 +0800
[Flink 26443] Add benchmark framework
This closes #71.
---
README.md | 6 +
flink-ml-benchmark/README.md | 196 +++++++++++++++++++++
flink-ml-benchmark/pom.xml | 105 +++++++++++
.../org/apache/flink/ml/benchmark/Benchmark.java | 126 +++++++++++++
.../apache/flink/ml/benchmark/BenchmarkResult.java | 58 ++++++
.../apache/flink/ml/benchmark/BenchmarkUtils.java | 183 +++++++++++++++++++
.../ml/benchmark/datagenerator/DataGenerator.java | 22 +--
.../datagenerator/InputDataGenerator.java | 53 ++++++
.../clustering/KMeansModelDataGenerator.java | 79 +++++++++
.../common/DenseVectorArrayGenerator.java | 116 ++++++++++++
.../datagenerator/common/DenseVectorGenerator.java | 102 +++++++++++
.../datagenerator/param/HasArraySize.java | 26 ++-
.../datagenerator/param/HasVectorDim.java | 26 ++-
.../src/main/resources/benchmark-conf.json | 49 ++++++
.../apache/flink/ml/benchmark/BenchmarkTest.java | 105 +++++++++++
.../main/java/org/apache/flink/ml/api/Stage.java | 2 +
.../param/{LongArrayParam.java => ArrayParam.java} | 24 ++-
.../apache/flink/ml/param/DoubleArrayParam.java | 2 +-
.../org/apache/flink/ml/param/FloatArrayParam.java | 2 +-
.../java/org/apache/flink/ml/param/FloatParam.java | 10 ++
.../org/apache/flink/ml/param/IntArrayParam.java | 2 +-
.../org/apache/flink/ml/param/LongArrayParam.java | 2 +-
.../java/org/apache/flink/ml/param/LongParam.java | 10 ++
.../main/java/org/apache/flink/ml/param/Param.java | 21 ++-
.../apache/flink/ml/param/StringArrayParam.java | 2 +-
.../org/apache/flink/ml/util/ReadWriteUtils.java | 119 ++++++++-----
.../java/org/apache/flink/ml/api/StageTest.java | 54 +++++-
flink-ml-dist/pom.xml | 94 ++++++++++
flink-ml-dist/src/main/assemblies/bin.xml | 80 +++++++++
flink-ml-dist/src/main/flink-ml-bin/README.txt | 32 ++++
.../main/flink-ml-bin/bin/flink-ml-benchmark.sh | 42 +++++
flink-ml-uber/pom.xml | 7 +
pom.xml | 4 +
33 files changed, 1657 insertions(+), 104 deletions(-)
diff --git a/README.md b/README.md
index 27633bc..d935484 100644
--- a/README.md
+++ b/README.md
@@ -14,6 +14,12 @@ Then you will find a JAR file that contains your
application, plus any libraries
that you may have added as dependencies to the application:
`target/<artifact-id>-<version>.jar`.
+## <a name="benchmark"></a>Benchmark
+
+Flink ML provides functionalities to benchmark its machine learning algorithms.
+For detailed information, please check the [Benchmark
+Getting Started](./flink-ml-benchmark/README.md).
+
## <a name="documentation"></a>Documentation
The documentation of Flink ML is located on the website:
diff --git a/flink-ml-benchmark/README.md b/flink-ml-benchmark/README.md
new file mode 100644
index 0000000..ac57722
--- /dev/null
+++ b/flink-ml-benchmark/README.md
@@ -0,0 +1,196 @@
+# Flink ML Benchmark Getting Started
+
+This document provides instructions on how to run benchmarks on Flink ML's
+stages in a Linux/MacOS environment.
+
+## Prerequisites
+
+### Install Flink
+
+Please make sure Flink 1.14 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.
+
+### Set Up Flink Environment Variables
+
+After having installed Flink, please register `$FLINK_HOME` as an environment
+variable into your local environment.
+
+```bash
+cd ${path_to_flink}
+export FLINK_HOME=`pwd`
+```
+
+[//]: # (TODO: Add instructions to download binary distribution when release is
+ available)
+### Build Flink ML library
+
+In order to use Flink ML's CLI you need to have the latest binary distribution
+of Flink ML. You can acquire the distribution by building Flink ML's source
code
+locally with the following command.
+
+```bash
+cd ${path_to_flink_ml}
+mvn clean package -DskipTests
+cd ./flink-ml-dist/target/flink-ml-*-bin/flink-ml*/
+```
+
+### Add Flink ML binaries to Flink
+
+You need to copy Flink ML's binary distribution files to Flink's folder for
+proper initialization. Please run the following command from Flink ML's binary
+distribution's folder.
+
+```bash
+cp ./lib/*.jar $FLINK_HOME/lib/
+```
+
+## Run Benchmark Example
+
+Please start a Flink standalone session in your local environment with the
+following command.
+
+```bash
+$FLINK_HOME/bin/start-cluster.sh
+```
+
+You should be able to navigate to the web UI at
+[localhost:8081](http://localhost:8081/) to view the Flink dashboard and see
+that the cluster is up and running.
+
+Then in Flink ML's binary distribution's folder, execute the following command
+to run the default benchmarks.
+
+```bash
+./bin/flink-ml-benchmark.sh ./conf/benchmark-conf.json --output-file
results.json
+```
+
+You will notice that some Flink jobs are submitted to your Flink cluster, and
+the following information is printed out in your terminal. This means that you
+have successfully executed the default benchmarks.
+
+```
+Found benchmarks [KMeansModel-1, KMeans-1]
+...
+Benchmarks execution completed.
+Benchmark results summary:
+[ {
+ "name" : "KMeansModel-1",
+ "totalTimeMs" : 782.0,
+ "inputRecordNum" : 10000,
+ "inputThroughput" : 12787.723785166241,
+ "outputRecordNum" : 10000,
+ "outputThroughput" : 12787.723785166241
+}, {
+ "name" : "KMeans-1",
+ "totalTimeMs" : 7022.0,
+ "inputRecordNum" : 10000,
+ "inputThroughput" : 1424.0956992309884,
+ "outputRecordNum" : 1,
+ "outputThroughput" : 0.14240956992309883
+} ]
+Benchmark results saved as json in results.json.
+```
+
+The command above would save the results into `results.json` as below.
+
+```json
+[ {
+ "name" : "KMeansModel-1",
+ "totalTimeMs" : 782.0,
+ "inputRecordNum" : 10000,
+ "inputThroughput" : 12787.723785166241,
+ "outputRecordNum" : 10000,
+ "outputThroughput" : 12787.723785166241
+}, {
+ "name" : "KMeans-1",
+ "totalTimeMs" : 7022.0,
+ "inputRecordNum" : 10000,
+ "inputThroughput" : 1424.0956992309884,
+ "outputRecordNum" : 1,
+ "outputThroughput" : 0.14240956992309883
+} ]
+```
+
+## Customize Benchmark Configuration
+
+`flink-ml-benchmark.sh` parses benchmarks to be executed according to the input
+configuration file, like `./conf/benchmark-conf.json`. It can also parse your
+custom configuration file so long as it contains a JSON object in the following
+format.
+
+- The file should contain the following as the metadata of the JSON object.
+ - `"version"`: The version of the json format. Currently its value must be 1.
+- Keys in the JSON object, except `"version"`, are regarded as the names of the
+ benchmarks.
+- The value of each benchmark name should be a JSON object containing the
+ following keys.
+ - `"stage"`: The stage to be benchmarked.
+ - `"inputs"`: The input data of the stage to be benchmarked.
+ - `"modelData"`(Optional): The model data of the stage to be benchmarked, if
+ the stage is a `Model` and needs to have its model data explicitly set.
+- The value of `"stage"`, `"inputs"` or `"modelData"` should be a JSON object
+ containing the following keys.
+ - `"className"`: The full classpath of a `WithParams` subclass. For
`"stage"`,
+ the class should be a subclass of `Stage`. For `"inputs"` or `"modelData"`,
+ the class should be a subclass of `DataGenerator`.
+ - `"paramMap"`: A JSON object containing the parameters related to the
+ specific `Stage` or `DataGenerator`.
+
+Combining the format requirements above, the example configuration in
+`./conf/benchmark-conf.json` is as follows. This configuration contains two
+benchmarks. The first benchmark name is "KMeans-1", and is executed on `KMeans`
+stage. The stage is benchmarked against 10000 randomly generated vectors.
+
+```json
+{
+ "version": 1,
+ "KMeans-1": {
+ "stage": {
+ "className": "org.apache.flink.ml.clustering.kmeans.KMeans",
+ "paramMap": {
+ "featuresCol": "features",
+ "predictionCol": "prediction"
+ }
+ },
+ "inputData": {
+ "className":
"org.apache.flink.ml.benchmark.datagenerator.common.DenseVectorGenerator",
+ "paramMap": {
+ "seed": 2,
+ "colNames": ["features"],
+ "numValues": 10000,
+ "vectorDim": 10
+ }
+ }
+ },
+ "KMeansModel-1": {
+ "stage": {
+ "className": "org.apache.flink.ml.clustering.kmeans.KMeansModel",
+ "paramMap": {
+ "featuresCol": "features",
+ "k": 2,
+ "distanceMeasure": "euclidean",
+ "predictionCol": "prediction"
+ }
+ },
+ "modelData": {
+ "className":
"org.apache.flink.ml.benchmark.datagenerator.clustering.KMeansModelDataGenerator",
+ "paramMap": {
+ "seed": 1,
+ "arraySize": 2,
+ "vectorDim": 10
+ }
+ },
+ "inputData": {
+ "className":
"org.apache.flink.ml.benchmark.datagenerator.common.DenseVectorGenerator",
+ "paramMap": {
+ "seed": 2,
+ "colNames": ["features"],
+ "numValues": 10000,
+ "vectorDim": 10
+ }
+ }
+ }
+}
+```
diff --git a/flink-ml-benchmark/pom.xml b/flink-ml-benchmark/pom.xml
new file mode 100644
index 0000000..5b704e8
--- /dev/null
+++ b/flink-ml-benchmark/pom.xml
@@ -0,0 +1,105 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+ http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ 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>
+
+ <parent>
+ <artifactId>flink-ml-parent</artifactId>
+ <groupId>org.apache.flink</groupId>
+ <version>2.1-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>flink-ml-benchmark_${scala.binary.version}</artifactId>
+ <name>Flink ML : Benchmark</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-ml-iteration_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-ml-core_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-ml-lib_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-java</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+
<artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-clients_${scala.binary.version}</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>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils-junit</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+
<artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils_${scala.binary.version}</artifactId>
+ <version>${flink.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+</project>
diff --git
a/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/Benchmark.java
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/Benchmark.java
new file mode 100644
index 0000000..f870dc6
--- /dev/null
+++
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/Benchmark.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.benchmark;
+
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+/** Entry class for benchmark execution. */
+public class Benchmark {
+ private static final Logger LOG = LoggerFactory.getLogger(Benchmark.class);
+
+ static final String VERSION_KEY = "version";
+
+ static final Option HELP_OPTION =
+ Option.builder("h")
+ .longOpt("help")
+ .desc("Show the help message for the command line
interface.")
+ .build();
+
+ static final Option OUTPUT_FILE_OPTION =
+ Option.builder()
+ .longOpt("output-file")
+ .desc("The output file name to save benchmark results.")
+ .hasArg()
+ .build();
+
+ static final Options OPTIONS =
+ new Options().addOption(HELP_OPTION).addOption(OUTPUT_FILE_OPTION);
+
+ public static void printHelp() {
+ HelpFormatter formatter = new HelpFormatter();
+ formatter.setLeftPadding(5);
+ formatter.setWidth(80);
+
+ System.out.println("./flink-ml-benchmark.sh <config-file-path>
[OPTIONS]");
+ System.out.println();
+ formatter.setSyntaxPrefix("The following options are available:");
+ formatter.printHelp(" ", OPTIONS);
+
+ System.out.println();
+ }
+
+ @SuppressWarnings("unchecked")
+ public static void executeBenchmarks(CommandLine commandLine) throws
Exception {
+ String configFile = commandLine.getArgs()[0];
+ Map<String, ?> benchmarks = BenchmarkUtils.parseJsonFile(configFile);
+ System.out.println("Found benchmarks " + benchmarks.keySet());
+
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ List<BenchmarkResult> results = new ArrayList<>();
+ for (Map.Entry<String, ?> benchmark : benchmarks.entrySet()) {
+ LOG.info("Running benchmark " + benchmark.getKey() + ".");
+
+ BenchmarkResult result =
+ BenchmarkUtils.runBenchmark(
+ tEnv, benchmark.getKey(), (Map<String, ?>)
benchmark.getValue());
+
+ results.add(result);
+ LOG.info(BenchmarkUtils.getResultsMapAsJson(result));
+ }
+
+ String benchmarkResultsJson =
+ BenchmarkUtils.getResultsMapAsJson(results.toArray(new
BenchmarkResult[0]));
+ System.out.println("Benchmarks execution completed.");
+ System.out.println("Benchmark results summary:");
+ System.out.println(benchmarkResultsJson);
+
+ if (commandLine.hasOption(OUTPUT_FILE_OPTION.getLongOpt())) {
+ String saveFile =
commandLine.getOptionValue(OUTPUT_FILE_OPTION.getLongOpt());
+ ReadWriteUtils.saveToFile(saveFile, benchmarkResultsJson, true);
+ System.out.println("Benchmark results saved as json in " +
saveFile + ".");
+ }
+ }
+
+ public static void printInvalidError(String[] args) {
+ System.out.println("Invalid command line arguments " +
Arrays.toString(args));
+ System.out.println();
+ System.out.println("Specify the help option (-h or --help) to get help
on the command.");
+ }
+
+ public static void main(String[] args) throws Exception {
+ CommandLineParser parser = new DefaultParser();
+ CommandLine commandLine = parser.parse(OPTIONS, args);
+
+ if (commandLine.hasOption(HELP_OPTION.getLongOpt())) {
+ printHelp();
+ } else if (commandLine.getArgs().length == 1) {
+ executeBenchmarks(commandLine);
+ } else {
+ printInvalidError(args);
+ }
+ }
+}
diff --git
a/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/BenchmarkResult.java
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/BenchmarkResult.java
new file mode 100644
index 0000000..4efdec0
--- /dev/null
+++
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/BenchmarkResult.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.benchmark;
+
+import org.apache.flink.util.Preconditions;
+
+/** The result of executing a benchmark. */
+public class BenchmarkResult {
+ /** The benchmark name. */
+ public final String name;
+
+ /** The total execution time of the benchmark in milliseconds. */
+ public final Double totalTimeMs;
+
+ /** The total number of input records. */
+ public final Long inputRecordNum;
+
+ /** The average input throughput in number of records per second. */
+ public final Double inputThroughput;
+
+ /** The total number of output records. */
+ public final Long outputRecordNum;
+
+ /** The average output throughput in number of records per second. */
+ public final Double outputThroughput;
+
+ public BenchmarkResult(
+ String name,
+ Double totalTimeMs,
+ Long inputRecordNum,
+ Double inputThroughput,
+ Long outputRecordNum,
+ Double outputThroughput) {
+ Preconditions.checkNotNull(name);
+ this.name = name;
+ this.totalTimeMs = totalTimeMs;
+ this.inputRecordNum = inputRecordNum;
+ this.inputThroughput = inputThroughput;
+ this.outputRecordNum = outputRecordNum;
+ this.outputThroughput = outputThroughput;
+ }
+}
diff --git
a/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/BenchmarkUtils.java
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/BenchmarkUtils.java
new file mode 100644
index 0000000..f21b43b
--- /dev/null
+++
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/BenchmarkUtils.java
@@ -0,0 +1,183 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.benchmark;
+
+import org.apache.flink.api.common.JobExecutionResult;
+import org.apache.flink.api.common.accumulators.LongCounter;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.api.AlgoOperator;
+import org.apache.flink.ml.api.Estimator;
+import org.apache.flink.ml.api.Model;
+import org.apache.flink.ml.api.Stage;
+import org.apache.flink.ml.benchmark.datagenerator.DataGenerator;
+import org.apache.flink.ml.benchmark.datagenerator.InputDataGenerator;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.util.Preconditions;
+
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/** Utility methods for benchmarks. */
+public class BenchmarkUtils {
+ /** Loads benchmark configuration maps from the provided json file. */
+ @SuppressWarnings("unchecked")
+ public static Map<String, ?> parseJsonFile(String path) throws IOException
{
+ InputStream inputStream = new FileInputStream(path);
+ Map<String, ?> jsonMap =
ReadWriteUtils.OBJECT_MAPPER.readValue(inputStream, Map.class);
+ Preconditions.checkArgument(
+ jsonMap.containsKey(Benchmark.VERSION_KEY)
+ && jsonMap.get(Benchmark.VERSION_KEY).equals(1));
+
+ return jsonMap.entrySet().stream()
+ .filter(x -> !x.getKey().equals(Benchmark.VERSION_KEY))
+ .collect(Collectors.toMap(Map.Entry::getKey,
Map.Entry::getValue));
+ }
+
+ /**
+ * Instantiates a benchmark from its parameter map and executes the
benchmark in the provided
+ * environment.
+ *
+ * @return Results of the executed benchmark.
+ */
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ public static BenchmarkResult runBenchmark(
+ StreamTableEnvironment tEnv, String name, Map<String, ?> params)
throws Exception {
+ Stage stage = ReadWriteUtils.instantiateWithParams((Map<String, ?>)
params.get("stage"));
+ InputDataGenerator inputDataGenerator =
+ ReadWriteUtils.instantiateWithParams((Map<String, ?>)
params.get("inputData"));
+ DataGenerator modelDataGenerator = null;
+ if (params.containsKey("modelData")) {
+ modelDataGenerator =
+ ReadWriteUtils.instantiateWithParams((Map<String, ?>)
params.get("modelData"));
+ }
+
+ return runBenchmark(tEnv, name, stage, inputDataGenerator,
modelDataGenerator);
+ }
+
+ /**
+ * Executes a benchmark from a stage with its inputDataGenerator and
modelDataGenerator in the
+ * provided environment.
+ *
+ * @return Results of the executed benchmark.
+ */
+ private static BenchmarkResult runBenchmark(
+ StreamTableEnvironment tEnv,
+ String name,
+ Stage<?> stage,
+ InputDataGenerator<?> inputDataGenerator,
+ DataGenerator<?> modelDataGenerator)
+ throws Exception {
+ StreamExecutionEnvironment env =
TableUtils.getExecutionEnvironment(tEnv);
+
+ Table[] inputTables = inputDataGenerator.getData(tEnv);
+ if (modelDataGenerator != null) {
+ ((Model<?>) stage).setModelData(modelDataGenerator.getData(tEnv));
+ }
+
+ Table[] outputTables;
+ if (stage instanceof Estimator) {
+ outputTables = ((Estimator<?, ?>)
stage).fit(inputTables).getModelData();
+ } else if (stage instanceof AlgoOperator) {
+ outputTables = ((AlgoOperator<?>) stage).transform(inputTables);
+ } else {
+ throw new IllegalArgumentException("Unsupported Stage class " +
stage.getClass());
+ }
+
+ for (Table table : outputTables) {
+ tEnv.toDataStream(table).addSink(new
CountingAndDiscardingSink<>());
+ }
+
+ JobExecutionResult executionResult = env.execute("Flink ML Benchmark
Job " + name);
+
+ double totalTimeMs = (double)
executionResult.getNetRuntime(TimeUnit.MILLISECONDS);
+ long inputRecordNum = inputDataGenerator.getNumValues();
+ double inputThroughput = inputRecordNum * 1000.0 / totalTimeMs;
+ long outputRecordNum =
+
executionResult.getAccumulatorResult(CountingAndDiscardingSink.COUNTER_NAME);
+ double outputThroughput = outputRecordNum * 1000.0 / totalTimeMs;
+
+ return new BenchmarkResult(
+ name,
+ totalTimeMs,
+ inputRecordNum,
+ inputThroughput,
+ outputRecordNum,
+ outputThroughput);
+ }
+
+ /** Converts the benchmark results to a json string as a map. */
+ public static String getResultsMapAsJson(BenchmarkResult... results)
+ throws JsonProcessingException {
+ List<Map<String, ?>> resultsMap = new ArrayList<>();
+ for (BenchmarkResult result : results) {
+ Map<String, Object> map = new LinkedHashMap<>();
+ map.put("name", result.name);
+ map.put("totalTimeMs", result.totalTimeMs);
+ map.put("inputRecordNum", result.inputRecordNum);
+ map.put("inputThroughput", result.inputThroughput);
+ map.put("outputRecordNum", result.outputRecordNum);
+ map.put("outputThroughput", result.outputThroughput);
+ resultsMap.add(map);
+ }
+ return ReadWriteUtils.OBJECT_MAPPER
+ .writerWithDefaultPrettyPrinter()
+ .writeValueAsString(resultsMap);
+ }
+
+ /**
+ * A stream sink that counts the number of all elements. The counting
result is stored in an
+ * {@link org.apache.flink.api.common.accumulators.Accumulator} specified
by {@link
+ * #COUNTER_NAME} and can be acquired by {@link
+ *
org.apache.flink.api.common.JobExecutionResult#getAccumulatorResult(String)}.
+ *
+ * @param <T> The type of elements received by the sink.
+ */
+ private static class CountingAndDiscardingSink<T> extends
RichSinkFunction<T> {
+ public static final String COUNTER_NAME = "numElements";
+
+ private static final long serialVersionUID = 1L;
+
+ private final LongCounter numElementsCounter = new LongCounter();
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ getRuntimeContext().addAccumulator(COUNTER_NAME,
numElementsCounter);
+ }
+
+ @Override
+ public void invoke(T value, Context context) {
+ numElementsCounter.add(1L);
+ }
+ }
+}
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatParam.java
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/DataGenerator.java
similarity index 60%
copy from flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatParam.java
copy to
flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/DataGenerator.java
index 0de890c..846a36f 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatParam.java
+++
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/DataGenerator.java
@@ -16,17 +16,17 @@
* limitations under the License.
*/
-package org.apache.flink.ml.param;
+package org.apache.flink.ml.benchmark.datagenerator;
-/** Class for the float parameter. */
-public class FloatParam extends Param<Float> {
+import org.apache.flink.ml.common.param.HasSeed;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
- public FloatParam(
- String name, String description, Float defaultValue,
ParamValidator<Float> validator) {
- super(name, Float.class, description, defaultValue, validator);
- }
-
- public FloatParam(String name, String description, Float defaultValue) {
- this(name, description, defaultValue, ParamValidators.alwaysTrue());
- }
+/** Interface for generating data as table arrays. */
+public interface DataGenerator<T extends DataGenerator<T>> extends HasSeed<T> {
+ /**
+ * Gets an array of Tables containing the data generated in the provided
stream table
+ * environment.
+ */
+ Table[] getData(StreamTableEnvironment tEnv);
}
diff --git
a/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/InputDataGenerator.java
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/InputDataGenerator.java
new file mode 100644
index 0000000..e13c1a8
--- /dev/null
+++
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/InputDataGenerator.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.benchmark.datagenerator;
+
+import org.apache.flink.ml.param.LongParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.StringArrayParam;
+
+/** Interface for generating data as input table arrays. */
+public interface InputDataGenerator<T extends InputDataGenerator<T>> extends
DataGenerator<T> {
+ Param<Long> NUM_VALUES =
+ new LongParam(
+ "numValues", "Number of data to be generated.", 10L,
ParamValidators.gt(0));
+
+ Param<String[]> COL_NAMES =
+ new StringArrayParam(
+ "colNames",
+ "An array of common-separated strings representing field
names of data tables.",
+ null);
+
+ default long getNumValues() {
+ return get(NUM_VALUES);
+ }
+
+ default T setNumValues(long value) {
+ return set(NUM_VALUES, value);
+ }
+
+ default String[] getColNames() {
+ return get(COL_NAMES);
+ }
+
+ default T setColNames(String... value) {
+ return set(COL_NAMES, value);
+ }
+}
diff --git
a/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/clustering/KMeansModelDataGenerator.java
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/clustering/KMeansModelDataGenerator.java
new file mode 100644
index 0000000..38ef523
--- /dev/null
+++
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/clustering/KMeansModelDataGenerator.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.benchmark.datagenerator.clustering;
+
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.ml.benchmark.datagenerator.DataGenerator;
+import org.apache.flink.ml.benchmark.datagenerator.InputDataGenerator;
+import
org.apache.flink.ml.benchmark.datagenerator.common.DenseVectorArrayGenerator;
+import org.apache.flink.ml.benchmark.datagenerator.param.HasArraySize;
+import org.apache.flink.ml.benchmark.datagenerator.param.HasVectorDim;
+import org.apache.flink.ml.clustering.kmeans.KMeansModelData;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.ml.util.ReadWriteUtils;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A DataGenerator which creates a table containing one {@link
+ * org.apache.flink.ml.clustering.kmeans.KMeansModel} instance.
+ */
+public class KMeansModelDataGenerator
+ implements DataGenerator<KMeansModelDataGenerator>,
+ HasVectorDim<KMeansModelDataGenerator>,
+ HasArraySize<KMeansModelDataGenerator> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public KMeansModelDataGenerator() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] getData(StreamTableEnvironment tEnv) {
+ InputDataGenerator<?> vectorArrayGenerator = new
DenseVectorArrayGenerator();
+ ReadWriteUtils.updateExistingParams(vectorArrayGenerator, paramMap);
+ vectorArrayGenerator.setNumValues(1);
+
+ Table vectorArrayTable = vectorArrayGenerator.getData(tEnv)[0];
+ DataStream<KMeansModelData> modelDataStream =
+ tEnv.toDataStream(vectorArrayTable, DenseVector[].class)
+ .map(new GenerateKMeansModelDataFunction());
+
+ return new Table[] {tEnv.fromDataStream(modelDataStream)};
+ }
+
+ private static class GenerateKMeansModelDataFunction
+ implements MapFunction<DenseVector[], KMeansModelData> {
+ @Override
+ public KMeansModelData map(DenseVector[] vectors) {
+ return new KMeansModelData(vectors, new
DenseVector(vectors.length));
+ }
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+}
diff --git
a/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/common/DenseVectorArrayGenerator.java
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/common/DenseVectorArrayGenerator.java
new file mode 100644
index 0000000..91c079d
--- /dev/null
+++
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/common/DenseVectorArrayGenerator.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.benchmark.datagenerator.common;
+
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.benchmark.datagenerator.InputDataGenerator;
+import org.apache.flink.ml.benchmark.datagenerator.param.HasArraySize;
+import org.apache.flink.ml.benchmark.datagenerator.param.HasVectorDim;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.util.NumberSequenceIterator;
+import org.apache.flink.util.Preconditions;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+
+/** A DataGenerator which creates a table of DenseVector array. */
+public class DenseVectorArrayGenerator
+ implements InputDataGenerator<DenseVectorArrayGenerator>,
+ HasArraySize<DenseVectorArrayGenerator>,
+ HasVectorDim<DenseVectorArrayGenerator> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public DenseVectorArrayGenerator() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] getData(StreamTableEnvironment tEnv) {
+ StreamExecutionEnvironment env =
TableUtils.getExecutionEnvironment(tEnv);
+
+ DataStream<DenseVector[]> dataStream =
+ env.fromParallelCollection(
+ new NumberSequenceIterator(1L, getNumValues()),
+ BasicTypeInfo.LONG_TYPE_INFO)
+ .map(
+ new
GenerateRandomContinuousVectorArrayFunction(
+ getSeed(), getVectorDim(),
getArraySize()));
+
+ Schema schema = Schema.newBuilder().column("f0",
DataTypes.of(DenseVector[].class)).build();
+ Table dataTable = tEnv.fromDataStream(dataStream, schema);
+ if (getColNames() != null) {
+ Preconditions.checkState(getColNames().length > 0);
+ dataTable = dataTable.as(getColNames()[0]);
+ }
+
+ return new Table[] {dataTable};
+ }
+
+ private static class GenerateRandomContinuousVectorArrayFunction
+ extends RichMapFunction<Long, DenseVector[]> {
+ private final int vectorDim;
+ private final long initSeed;
+ private final int arraySize;
+ private Random random;
+
+ private GenerateRandomContinuousVectorArrayFunction(
+ long initSeed, int vectorDim, int arraySize) {
+ this.vectorDim = vectorDim;
+ this.initSeed = initSeed;
+ this.arraySize = arraySize;
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ int index = getRuntimeContext().getIndexOfThisSubtask();
+ random = new Random(Tuple2.of(initSeed, index).hashCode());
+ }
+
+ @Override
+ public DenseVector[] map(Long value) {
+ DenseVector[] result = new DenseVector[arraySize];
+ for (int i = 0; i < arraySize; i++) {
+ result[i] = new DenseVector(vectorDim);
+ for (int j = 0; j < vectorDim; j++) {
+ result[i].values[j] = random.nextDouble();
+ }
+ }
+ return result;
+ }
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+}
diff --git
a/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/common/DenseVectorGenerator.java
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/common/DenseVectorGenerator.java
new file mode 100644
index 0000000..f05cf75
--- /dev/null
+++
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/common/DenseVectorGenerator.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.benchmark.datagenerator.common;
+
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.ml.benchmark.datagenerator.InputDataGenerator;
+import org.apache.flink.ml.benchmark.datagenerator.param.HasVectorDim;
+import org.apache.flink.ml.common.datastream.TableUtils;
+import org.apache.flink.ml.linalg.DenseVector;
+import org.apache.flink.ml.linalg.Vectors;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.util.ParamUtils;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.Table;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.util.NumberSequenceIterator;
+import org.apache.flink.util.Preconditions;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+
+/** A DataGenerator which creates a table of DenseVector. */
+public class DenseVectorGenerator
+ implements InputDataGenerator<DenseVectorGenerator>,
HasVectorDim<DenseVectorGenerator> {
+ private final Map<Param<?>, Object> paramMap = new HashMap<>();
+
+ public DenseVectorGenerator() {
+ ParamUtils.initializeMapWithDefaultValues(paramMap, this);
+ }
+
+ @Override
+ public Table[] getData(StreamTableEnvironment tEnv) {
+ StreamExecutionEnvironment env =
TableUtils.getExecutionEnvironment(tEnv);
+
+ DataStream<DenseVector> dataStream =
+ env.fromParallelCollection(
+ new NumberSequenceIterator(1L, getNumValues()),
+ BasicTypeInfo.LONG_TYPE_INFO)
+ .map(new RandomDenseVectorGenerator(getSeed(),
getVectorDim()));
+
+ Table dataTable = tEnv.fromDataStream(dataStream);
+ if (getColNames() != null) {
+ Preconditions.checkState(getColNames().length > 0);
+ dataTable = dataTable.as(getColNames()[0]);
+ }
+
+ return new Table[] {dataTable};
+ }
+
+ private static class RandomDenseVectorGenerator extends
RichMapFunction<Long, DenseVector> {
+ private final int vectorDim;
+ private final long initSeed;
+ private Random random;
+
+ private RandomDenseVectorGenerator(long initSeed, int vectorDim) {
+ this.vectorDim = vectorDim;
+ this.initSeed = initSeed;
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ super.open(parameters);
+ int index = getRuntimeContext().getIndexOfThisSubtask();
+ random = new Random(Tuple2.of(initSeed, index).hashCode());
+ }
+
+ @Override
+ public DenseVector map(Long value) {
+ double[] values = new double[vectorDim];
+ for (int i = 0; i < vectorDim; i++) {
+ values[i] = random.nextDouble();
+ }
+ return Vectors.dense(values);
+ }
+ }
+
+ @Override
+ public Map<Param<?>, Object> getParamMap() {
+ return paramMap;
+ }
+}
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatParam.java
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/param/HasArraySize.java
similarity index 53%
copy from flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatParam.java
copy to
flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/param/HasArraySize.java
index 0de890c..4c2d48e 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatParam.java
+++
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/param/HasArraySize.java
@@ -16,17 +16,27 @@
* limitations under the License.
*/
-package org.apache.flink.ml.param;
+package org.apache.flink.ml.benchmark.datagenerator.param;
-/** Class for the float parameter. */
-public class FloatParam extends Param<Float> {
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.WithParams;
- public FloatParam(
- String name, String description, Float defaultValue,
ParamValidator<Float> validator) {
- super(name, Float.class, description, defaultValue, validator);
+/** Interface for the benchmark array size param. */
+public interface HasArraySize<T> extends WithParams<T> {
+ Param<Integer> ARRAY_SIZE =
+ new IntParam(
+ "arraySize",
+ "Number of elements in the generated array.",
+ 1,
+ ParamValidators.gt(0));
+
+ default int getArraySize() {
+ return get(ARRAY_SIZE);
}
- public FloatParam(String name, String description, Float defaultValue) {
- this(name, description, defaultValue, ParamValidators.alwaysTrue());
+ default T setArraySize(int value) {
+ return set(ARRAY_SIZE, value);
}
}
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatParam.java
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/param/HasVectorDim.java
similarity index 53%
copy from flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatParam.java
copy to
flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/param/HasVectorDim.java
index 0de890c..824f821 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatParam.java
+++
b/flink-ml-benchmark/src/main/java/org/apache/flink/ml/benchmark/datagenerator/param/HasVectorDim.java
@@ -16,17 +16,27 @@
* limitations under the License.
*/
-package org.apache.flink.ml.param;
+package org.apache.flink.ml.benchmark.datagenerator.param;
-/** Class for the float parameter. */
-public class FloatParam extends Param<Float> {
+import org.apache.flink.ml.param.IntParam;
+import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.ParamValidators;
+import org.apache.flink.ml.param.WithParams;
- public FloatParam(
- String name, String description, Float defaultValue,
ParamValidator<Float> validator) {
- super(name, Float.class, description, defaultValue, validator);
+/** Interface for the benchmark vector dimension param. */
+public interface HasVectorDim<T> extends WithParams<T> {
+ Param<Integer> VECTOR_DIM =
+ new IntParam(
+ "vectorDim",
+ "Dimension of vector-typed data to be generated.",
+ 1,
+ ParamValidators.gt(0));
+
+ default int getVectorDim() {
+ return get(VECTOR_DIM);
}
- public FloatParam(String name, String description, Float defaultValue) {
- this(name, description, defaultValue, ParamValidators.alwaysTrue());
+ default T setVectorDim(int value) {
+ return set(VECTOR_DIM, value);
}
}
diff --git a/flink-ml-benchmark/src/main/resources/benchmark-conf.json
b/flink-ml-benchmark/src/main/resources/benchmark-conf.json
new file mode 100644
index 0000000..280453e
--- /dev/null
+++ b/flink-ml-benchmark/src/main/resources/benchmark-conf.json
@@ -0,0 +1,49 @@
+{
+ "version": 1,
+ "KMeans-1": {
+ "stage": {
+ "className": "org.apache.flink.ml.clustering.kmeans.KMeans",
+ "paramMap": {
+ "featuresCol": "features",
+ "predictionCol": "prediction"
+ }
+ },
+ "inputData": {
+ "className":
"org.apache.flink.ml.benchmark.datagenerator.common.DenseVectorGenerator",
+ "paramMap": {
+ "seed": 2,
+ "colNames": ["features"],
+ "numValues": 10000,
+ "vectorDim": 10
+ }
+ }
+ },
+ "KMeansModel-1": {
+ "stage": {
+ "className": "org.apache.flink.ml.clustering.kmeans.KMeansModel",
+ "paramMap": {
+ "featuresCol": "features",
+ "k": 2,
+ "distanceMeasure": "euclidean",
+ "predictionCol": "prediction"
+ }
+ },
+ "modelData": {
+ "className":
"org.apache.flink.ml.benchmark.datagenerator.clustering.KMeansModelDataGenerator",
+ "paramMap": {
+ "seed": 1,
+ "arraySize": 2,
+ "vectorDim": 10
+ }
+ },
+ "inputData": {
+ "className":
"org.apache.flink.ml.benchmark.datagenerator.common.DenseVectorGenerator",
+ "paramMap": {
+ "seed": 2,
+ "colNames": ["features"],
+ "numValues": 10000,
+ "vectorDim": 10
+ }
+ }
+ }
+}
diff --git
a/flink-ml-benchmark/src/test/java/org/apache/flink/ml/benchmark/BenchmarkTest.java
b/flink-ml-benchmark/src/test/java/org/apache/flink/ml/benchmark/BenchmarkTest.java
new file mode 100644
index 0000000..c3ca0cf
--- /dev/null
+++
b/flink-ml-benchmark/src/test/java/org/apache/flink/ml/benchmark/BenchmarkTest.java
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.ml.benchmark;
+
+import org.apache.flink.ml.benchmark.datagenerator.common.DenseVectorGenerator;
+import org.apache.flink.ml.clustering.kmeans.KMeans;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+
+import org.apache.commons.io.FileUtils;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/** Tests benchmarks. */
+public class BenchmarkTest extends AbstractTestBase {
+ @Rule public final TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Test
+ public void testParseJsonFile() throws Exception {
+ File configFile = new File(tempFolder.newFolder().getAbsolutePath() +
"/test-conf.json");
+ InputStream inputStream =
+
this.getClass().getClassLoader().getResourceAsStream("benchmark-conf.json");
+ FileUtils.copyInputStreamToFile(inputStream, configFile);
+
+ Map<String, ?> benchmarks =
BenchmarkUtils.parseJsonFile(configFile.getAbsolutePath());
+ assertEquals(benchmarks.size(), 2);
+ assertTrue(benchmarks.containsKey("KMeans-1"));
+ assertTrue(benchmarks.containsKey("KMeansModel-1"));
+ }
+
+ @Test
+ public void testRunBenchmark() throws Exception {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+
+ Map<String, Object> params = new HashMap<>();
+
+ Map<String, Object> stageParams = new HashMap<>();
+ stageParams.put("className", KMeans.class.getCanonicalName());
+ stageParams.put(
+ "paramMap",
+ new HashMap<String, Object>() {
+ {
+ put("k", 5);
+ put("featuresCol", "test_feature");
+ }
+ });
+ params.put("stage", stageParams);
+
+ Map<String, Object> inputDataParams = new HashMap<>();
+ inputDataParams.put("className",
DenseVectorGenerator.class.getCanonicalName());
+ inputDataParams.put(
+ "paramMap",
+ new HashMap<String, Object>() {
+ {
+ put("colNames", new String[] {"test_feature"});
+ put("numValues", 1000L);
+ put("vectorDim", 10);
+ }
+ });
+ params.put("inputData", inputDataParams);
+
+ long estimatedTime = System.currentTimeMillis();
+ BenchmarkResult result = BenchmarkUtils.runBenchmark(tEnv,
"testBenchmarkName", params);
+ estimatedTime = System.currentTimeMillis() - estimatedTime;
+
+ assertEquals("testBenchmarkName", result.name);
+ assertTrue(result.totalTimeMs > 0);
+ assertTrue(result.totalTimeMs <= estimatedTime);
+ assertEquals(1000L, (long) result.inputRecordNum);
+ assertEquals(1L, (long) result.outputRecordNum);
+ assertEquals(
+ result.inputRecordNum * 1000.0 / result.totalTimeMs,
result.inputThroughput, 1e-5);
+ assertEquals(
+ result.outputRecordNum * 1000.0 / result.totalTimeMs,
+ result.outputThroughput,
+ 1e-5);
+ }
+}
diff --git a/flink-ml-core/src/main/java/org/apache/flink/ml/api/Stage.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/api/Stage.java
index 439b956..20a0bf7 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/api/Stage.java
+++ b/flink-ml-core/src/main/java/org/apache/flink/ml/api/Stage.java
@@ -31,6 +31,8 @@ import java.io.Serializable;
*
* <p>Each stage is with parameters, and requires a public empty constructor
for restoration.
*
+ * <p>NOTE: every Stage subclass should have a no-arg constructor.
+ *
* <p>NOTE: every Stage subclass should implement a static method with
signature {@code static T
* load(StreamTableEnvironment tEnv, String path)}, where {@code T} refers to
the concrete subclass.
* This static method should instantiate a new stage instance based on the
data read from the given
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/param/LongArrayParam.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/param/ArrayParam.java
similarity index 58%
copy from
flink-ml-core/src/main/java/org/apache/flink/ml/param/LongArrayParam.java
copy to flink-ml-core/src/main/java/org/apache/flink/ml/param/ArrayParam.java
index 5e4fc47..adc3b80 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/param/LongArrayParam.java
+++ b/flink-ml-core/src/main/java/org/apache/flink/ml/param/ArrayParam.java
@@ -18,18 +18,26 @@
package org.apache.flink.ml.param;
-/** Class for the long array parameter. */
-public class LongArrayParam extends Param<Long[]> {
+import org.apache.flink.ml.util.ReadWriteUtils;
- public LongArrayParam(
+import java.io.IOException;
+
+/** Class for array-type parameters. */
+public class ArrayParam<T> extends Param<T[]> {
+
+ public ArrayParam(
String name,
+ Class<T[]> clazz,
String description,
- Long[] defaultValue,
- ParamValidator<Long[]> validator) {
- super(name, Long[].class, description, defaultValue, validator);
+ T[] defaultValue,
+ ParamValidator<T[]> validator) {
+ super(name, clazz, description, defaultValue, validator);
}
- public LongArrayParam(String name, String description, Long[]
defaultValue) {
- this(name, description, defaultValue, ParamValidators.alwaysTrue());
+ @Override
+ public T[] jsonDecode(Object json) throws IOException {
+ // Converts Object Mapper's default ArrayList<T> to T[].
+ String jsonStr = ReadWriteUtils.OBJECT_MAPPER.writeValueAsString(json);
+ return ReadWriteUtils.OBJECT_MAPPER.readValue(jsonStr, clazz);
}
}
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/param/DoubleArrayParam.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/param/DoubleArrayParam.java
index b86dd00..570629f 100644
---
a/flink-ml-core/src/main/java/org/apache/flink/ml/param/DoubleArrayParam.java
+++
b/flink-ml-core/src/main/java/org/apache/flink/ml/param/DoubleArrayParam.java
@@ -19,7 +19,7 @@
package org.apache.flink.ml.param;
/** Class for the double array parameter. */
-public class DoubleArrayParam extends Param<Double[]> {
+public class DoubleArrayParam extends ArrayParam<Double> {
public DoubleArrayParam(
String name,
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatArrayParam.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatArrayParam.java
index 4224557..af3704f 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatArrayParam.java
+++ b/flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatArrayParam.java
@@ -19,7 +19,7 @@
package org.apache.flink.ml.param;
/** Class for the float array parameter. */
-public class FloatArrayParam extends Param<Float[]> {
+public class FloatArrayParam extends ArrayParam<Float> {
public FloatArrayParam(
String name,
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatParam.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatParam.java
index 0de890c..f86ecfe 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatParam.java
+++ b/flink-ml-core/src/main/java/org/apache/flink/ml/param/FloatParam.java
@@ -18,6 +18,8 @@
package org.apache.flink.ml.param;
+import java.io.IOException;
+
/** Class for the float parameter. */
public class FloatParam extends Param<Float> {
@@ -29,4 +31,12 @@ public class FloatParam extends Param<Float> {
public FloatParam(String name, String description, Float defaultValue) {
this(name, description, defaultValue, ParamValidators.alwaysTrue());
}
+
+ @Override
+ public Float jsonDecode(Object json) throws IOException {
+ if (json instanceof Double) {
+ return ((Double) json).floatValue();
+ }
+ return (Float) json;
+ }
}
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/param/IntArrayParam.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/param/IntArrayParam.java
index 4f7c630..6ad7e58 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/param/IntArrayParam.java
+++ b/flink-ml-core/src/main/java/org/apache/flink/ml/param/IntArrayParam.java
@@ -19,7 +19,7 @@
package org.apache.flink.ml.param;
/** Class for the integer array parameter. */
-public class IntArrayParam extends Param<Integer[]> {
+public class IntArrayParam extends ArrayParam<Integer> {
public IntArrayParam(
String name,
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/param/LongArrayParam.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/param/LongArrayParam.java
index 5e4fc47..adff46f 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/param/LongArrayParam.java
+++ b/flink-ml-core/src/main/java/org/apache/flink/ml/param/LongArrayParam.java
@@ -19,7 +19,7 @@
package org.apache.flink.ml.param;
/** Class for the long array parameter. */
-public class LongArrayParam extends Param<Long[]> {
+public class LongArrayParam extends ArrayParam<Long> {
public LongArrayParam(
String name,
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/param/LongParam.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/param/LongParam.java
index 3fd7dd8..410229f 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/param/LongParam.java
+++ b/flink-ml-core/src/main/java/org/apache/flink/ml/param/LongParam.java
@@ -18,6 +18,8 @@
package org.apache.flink.ml.param;
+import java.io.IOException;
+
/** Class for the long parameter. */
public class LongParam extends Param<Long> {
@@ -29,4 +31,12 @@ public class LongParam extends Param<Long> {
public LongParam(String name, String description, Long defaultValue) {
this(name, description, defaultValue, ParamValidators.alwaysTrue());
}
+
+ @Override
+ public Long jsonDecode(Object json) throws IOException {
+ if (json instanceof Integer) {
+ return ((Integer) json).longValue();
+ }
+ return (Long) json;
+ }
}
diff --git a/flink-ml-core/src/main/java/org/apache/flink/ml/param/Param.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/param/Param.java
index b7a1aef..f5035bf 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/param/Param.java
+++ b/flink-ml-core/src/main/java/org/apache/flink/ml/param/Param.java
@@ -19,7 +19,6 @@
package org.apache.flink.ml.param;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.ml.util.ReadWriteUtils;
import java.io.IOException;
import java.io.Serializable;
@@ -58,24 +57,24 @@ public class Param<T> implements Serializable {
}
/**
- * Encodes the given object into a json-formatted string.
+ * Encodes the given java object into a json-supported object.
*
- * @param value An object of class type T.
- * @return A json-formatted string.
+ * @param value A java object of class type T.
+ * @return A json-supported object.
*/
- public String jsonEncode(T value) throws IOException {
- return ReadWriteUtils.OBJECT_MAPPER.writeValueAsString(value);
+ public Object jsonEncode(T value) throws IOException {
+ return value;
}
/**
- * Decodes the given string into an object of class type T.
+ * Decodes the given json-supported object into an object of class type T.
*
- * @param json A json-formatted string.
- * @return An object of class type T.
+ * @param json A json-supported object.
+ * @return A java object of class type T.
*/
@SuppressWarnings("unchecked")
- public T jsonDecode(String json) throws IOException {
- return ReadWriteUtils.OBJECT_MAPPER.readValue(json, clazz);
+ public T jsonDecode(Object json) throws IOException {
+ return (T) json;
}
@Override
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/param/StringArrayParam.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/param/StringArrayParam.java
index 5062463..7c6b736 100644
---
a/flink-ml-core/src/main/java/org/apache/flink/ml/param/StringArrayParam.java
+++
b/flink-ml-core/src/main/java/org/apache/flink/ml/param/StringArrayParam.java
@@ -19,7 +19,7 @@
package org.apache.flink.ml.param;
/** Class for the string array parameter. */
-public class StringArrayParam extends Param<String[]> {
+public class StringArrayParam extends ArrayParam<String> {
public StringArrayParam(
String name,
diff --git
a/flink-ml-core/src/main/java/org/apache/flink/ml/util/ReadWriteUtils.java
b/flink-ml-core/src/main/java/org/apache/flink/ml/util/ReadWriteUtils.java
index 54aefbe..ed7b227 100644
--- a/flink-ml-core/src/main/java/org/apache/flink/ml/util/ReadWriteUtils.java
+++ b/flink-ml-core/src/main/java/org/apache/flink/ml/util/ReadWriteUtils.java
@@ -33,6 +33,7 @@ import org.apache.flink.ml.builder.GraphModel;
import org.apache.flink.ml.builder.GraphNode;
import org.apache.flink.ml.common.datastream.TableUtils;
import org.apache.flink.ml.param.Param;
+import org.apache.flink.ml.param.WithParams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import
org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.BasePathBucketAssigner;
@@ -66,17 +67,17 @@ public class ReadWriteUtils {
// A helper method that calls encodes the given parameter value to a json
string. We can not
// call param.jsonEncode(value) directly because Param::jsonEncode(...)
needs the actual type
// of the value.
- private static <T> String jsonEncodeHelper(Param<T> param, Object value)
throws IOException {
+ private static <T> Object jsonEncodeHelper(Param<T> param, Object value)
throws IOException {
return param.jsonEncode((T) value);
}
// Converts Map<Param<?>, Object> to Map<String, String> which maps the
parameter name to the
// string-encoded parameter value.
- private static Map<String, String> jsonEncode(Map<Param<?>, Object>
paramMap)
+ private static Map<String, Object> jsonEncode(Map<Param<?>, Object>
paramMap)
throws IOException {
- Map<String, String> result = new HashMap<>(paramMap.size());
+ Map<String, Object> result = new HashMap<>(paramMap.size());
for (Map.Entry<Param<?>, Object> entry : paramMap.entrySet()) {
- String json = jsonEncodeHelper(entry.getKey(), entry.getValue());
+ Object json = jsonEncodeHelper(entry.getKey(), entry.getValue());
result.put(entry.getKey().name, json);
}
return result;
@@ -94,9 +95,6 @@ public class ReadWriteUtils {
*/
public static void saveMetadata(Stage<?> stage, String path, Map<String,
?> extraMetadata)
throws IOException {
- // Creates parent directories if not already created.
- FileSystem fs = mkdirs(path);
-
Map<String, Object> metadata = new HashMap<>(extraMetadata);
metadata.put("className", stage.getClass().getName());
metadata.put("timestamp", System.currentTimeMillis());
@@ -104,15 +102,27 @@ public class ReadWriteUtils {
// TODO: add version in the metadata.
String metadataStr = OBJECT_MAPPER.writeValueAsString(metadata);
- Path metadataPath = new Path(path, "metadata");
- if (fs.exists(metadataPath)) {
- throw new IOException("File " + metadataPath + " already exists.");
+ saveToFile(new Path(path, "metadata").toUri().toString(), metadataStr,
false);
+ }
+
+ /** Saves a given string to the specified file. */
+ public static void saveToFile(String pathStr, String content, boolean
isOverwrite)
+ throws IOException {
+ Path path = new Path(pathStr);
+
+ // Creates parent directories if not already created.
+ FileSystem fs = mkdirs(path.getParent());
+
+ FileSystem.WriteMode writeMode = FileSystem.WriteMode.OVERWRITE;
+ if (!isOverwrite) {
+ writeMode = FileSystem.WriteMode.NO_OVERWRITE;
+ if (fs.exists(path)) {
+ throw new IOException("File " + path + " already exists.");
+ }
}
try (BufferedWriter writer =
- new BufferedWriter(
- new OutputStreamWriter(
- fs.create(metadataPath,
FileSystem.WriteMode.NO_OVERWRITE)))) {
- writer.write(metadataStr);
+ new BufferedWriter(new OutputStreamWriter(fs.create(path,
writeMode)))) {
+ writer.write(content);
}
}
@@ -195,7 +205,7 @@ public class ReadWriteUtils {
public static void savePipeline(Stage<?> pipeline, List<Stage<?>> stages,
String path)
throws IOException {
// Creates parent directories if not already created.
- mkdirs(path);
+ mkdirs(new Path(path));
Map<String, Object> extraMetadata = new HashMap<>();
extraMetadata.put("numStages", stages.size());
@@ -232,10 +242,9 @@ public class ReadWriteUtils {
return stages;
}
- private static FileSystem mkdirs(String path) throws IOException {
- Path temp = new Path(path);
- FileSystem fs = temp.getFileSystem();
- fs.mkdirs(temp);
+ private static FileSystem mkdirs(Path path) throws IOException {
+ FileSystem fs = path.getFileSystem();
+ fs.mkdirs(path);
return fs;
}
@@ -249,7 +258,7 @@ public class ReadWriteUtils {
public static void saveGraph(Stage<?> graph, GraphData graphData, String
path)
throws IOException {
// Creates parent directories if not already created.
- mkdirs(path);
+ mkdirs(new Path(path));
Map<String, Object> extraMetadata = new HashMap<>();
extraMetadata.put("graphData", graphData.toMap());
@@ -277,6 +286,7 @@ public class ReadWriteUtils {
* @param expectedClassName The expected class name of the pipeline.
* @return A Graph or GraphModel instance.
*/
+ @SuppressWarnings("unchecked")
public static Stage<?> loadGraph(
StreamTableEnvironment tEnv, String path, String
expectedClassName) throws IOException {
Map<String, ?> metadata = loadMetadata(path, expectedClassName);
@@ -311,20 +321,23 @@ public class ReadWriteUtils {
graphData.outputModelDataIds);
}
- // A helper method that sets stage's parameter value. We can not call
stage.set(param, value)
- // directly because stage::set(...) needs the actual type of the value.
- public static <T> void setParam(Stage<?> stage, Param<T> param, Object
value) {
- stage.set(param, (T) value);
+ // A helper method that sets WithParams object's parameter value. We can
not call
+ // WithParams.set(param, value)
+ // directly because WithParams::set(...) needs the actual type of the
value.
+ @SuppressWarnings("unchecked")
+ public static <T> void setParam(WithParams<?> instance, Param<T> param,
Object value) {
+ instance.set(param, (T) value);
}
- // A helper method that updates stage's param map using values from the
paramOverrides. This
- // method only
- // updates values for parameters already defined in the stage's param map.
- public static void updateExistingParams(Stage<?> stage, Map<Param<?>,
Object> paramOverrides) {
- Set<Param<?>> existingParams = stage.getParamMap().keySet();
+ // A helper method that updates WithParams instance's param map using
values from the
+ // paramOverrides. This method only updates values for parameters already
defined in the
+ // instance's param map.
+ public static void updateExistingParams(
+ WithParams<?> instance, Map<Param<?>, Object> paramOverrides) {
+ Set<Param<?>> existingParams = instance.getParamMap().keySet();
for (Map.Entry<Param<?>, Object> entry : paramOverrides.entrySet()) {
if (existingParams.contains(entry.getKey())) {
- setParam(stage, entry.getKey(), entry.getValue());
+ setParam(instance, entry.getKey(), entry.getValue());
}
}
}
@@ -343,31 +356,41 @@ public class ReadWriteUtils {
* @param <T> The class type of the Stage subclass.
* @return An instance of class type T.
*/
- @SuppressWarnings("unchecked")
public static <T extends Stage<T>> T loadStageParam(String path) throws
IOException {
- Map<String, ?> metadata = loadMetadata(path, "");
- String className = (String) metadata.get("className");
- Map<String, String> paramMap = (Map<String, String>)
metadata.get("paramMap");
-
try {
- Class<T> clazz = (Class<T>) Class.forName(className);
- T instance = InstantiationUtil.instantiate(clazz);
-
- Map<String, Param<?>> nameToParam = new HashMap<>();
- for (Param<?> param :
ParamUtils.getPublicFinalParamFields(instance)) {
- nameToParam.put(param.name, param);
- }
-
- for (Map.Entry<String, String> entry : paramMap.entrySet()) {
- Param<?> param = nameToParam.get(entry.getKey());
- setParam(instance, param, param.jsonDecode(entry.getValue()));
- }
- return instance;
+ return instantiateWithParams(loadMetadata(path, ""));
} catch (ClassNotFoundException e) {
throw new RuntimeException("Failed to load stage.", e);
}
}
+ /**
+ * Instantiates a WithParams subclass from the provided json map.
+ *
+ * @param jsonMap a map containing className and paramMap.
+ * @return the instantiated WithParams subclass instance.
+ */
+ @SuppressWarnings("unchecked")
+ public static <T extends WithParams<T>> T
instantiateWithParams(Map<String, ?> jsonMap)
+ throws ClassNotFoundException, IOException {
+ String className = (String) jsonMap.get("className");
+ Class<T> clazz = (Class<T>) Class.forName(className);
+ T instance = InstantiationUtil.instantiate(clazz);
+
+ Map<String, Param<?>> nameToParam = new HashMap<>();
+ for (Param<?> param : ParamUtils.getPublicFinalParamFields(instance)) {
+ nameToParam.put(param.name, param);
+ }
+
+ Map<String, Object> paramMap = (Map<String, Object>)
jsonMap.get("paramMap");
+ for (Map.Entry<String, Object> entry : paramMap.entrySet()) {
+ Param<?> param = nameToParam.get(entry.getKey());
+ setParam(instance, param, param.jsonDecode(entry.getValue()));
+ }
+
+ return instance;
+ }
+
/**
* Loads the stage from the given path by invoking the static load()
method of the stage. The
* stage class name is read from the metadata file under the given path.
The load() method is
diff --git a/flink-ml-core/src/test/java/org/apache/flink/ml/api/StageTest.java
b/flink-ml-core/src/test/java/org/apache/flink/ml/api/StageTest.java
index 18b9be8..091502b 100644
--- a/flink-ml-core/src/test/java/org/apache/flink/ml/api/StageTest.java
+++ b/flink-ml-core/src/test/java/org/apache/flink/ml/api/StageTest.java
@@ -57,9 +57,18 @@ public class StageTest {
Param<Integer> INT_PARAM =
new IntParam("intParam", "Description", 1,
ParamValidators.lt(100));
+ // This param might be regarded as integer type given a small value.
Param<Long> LONG_PARAM =
new LongParam("longParam", "Description", 2L,
ParamValidators.lt(100));
+ // This param must be regarded as long type given a large enough value.
+ Param<Long> LONG_PARAM2 =
+ new LongParam(
+ "longParam2",
+ "Description",
+ Integer.MAX_VALUE + 1L,
+ ParamValidators.lt(Integer.MAX_VALUE + 100L));
+
Param<Float> FLOAT_PARAM =
new FloatParam("floatParam", "Description", 3.0f,
ParamValidators.lt(100));
@@ -259,6 +268,7 @@ public class StageTest {
MyStage stage = new MyStage();
assertInvalidValue(stage, MyParams.INT_PARAM, 100);
assertInvalidValue(stage, MyParams.LONG_PARAM, 100L);
+ assertInvalidValue(stage, MyParams.LONG_PARAM2, Integer.MAX_VALUE +
100L);
assertInvalidValue(stage, MyParams.FLOAT_PARAM, 100.0f);
assertInvalidValue(stage, MyParams.DOUBLE_PARAM, 100.0);
assertInvalidClass(stage, MyParams.INT_PARAM, "100");
@@ -281,11 +291,14 @@ public class StageTest {
stage.set(MyParams.LONG_PARAM, 50L);
Assert.assertEquals(50L, (long) stage.get(MyParams.LONG_PARAM));
+ stage.set(MyParams.LONG_PARAM2, Integer.MAX_VALUE + 50L);
+ Assert.assertEquals(Integer.MAX_VALUE + 50L, (long)
stage.get(MyParams.LONG_PARAM2));
+
stage.set(MyParams.FLOAT_PARAM, 50f);
- Assert.assertEquals(50f, (float) stage.get(MyParams.FLOAT_PARAM),
0.0001);
+ Assert.assertEquals(50f, stage.get(MyParams.FLOAT_PARAM), 0.0001);
stage.set(MyParams.DOUBLE_PARAM, 50.0);
- Assert.assertEquals(50, (double) stage.get(MyParams.DOUBLE_PARAM),
0.0001);
+ Assert.assertEquals(50, stage.get(MyParams.DOUBLE_PARAM), 0.0001);
stage.set(MyParams.STRING_PARAM, "50");
Assert.assertEquals("50", stage.get(MyParams.STRING_PARAM));
@@ -311,9 +324,39 @@ public class StageTest {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
MyStage stage = new MyStage();
+
stage.set(stage.paramWithNullDefault, 1);
+ stage.set(MyParams.BOOLEAN_PARAM, true);
+ stage.set(MyParams.INT_PARAM, 50);
+ stage.set(MyParams.LONG_PARAM, 50L);
+ stage.set(MyParams.LONG_PARAM2, Integer.MAX_VALUE + 50L);
+ stage.set(MyParams.FLOAT_PARAM, 50f);
+ stage.set(MyParams.DOUBLE_PARAM, 50.0);
+ stage.set(MyParams.STRING_PARAM, "50");
+ stage.set(MyParams.INT_ARRAY_PARAM, new Integer[] {50, 51});
+ stage.set(MyParams.LONG_ARRAY_PARAM, new Long[] {50L, 51L});
+ stage.set(MyParams.FLOAT_ARRAY_PARAM, new Float[] {50.0f, 51.0f});
+ stage.set(MyParams.DOUBLE_ARRAY_PARAM, new Double[] {50.0, 51.0});
+ stage.set(MyParams.STRING_ARRAY_PARAM, new String[] {"50", "51"});
+
Stage<?> loadedStage = validateStageSaveLoad(tEnv, stage,
Collections.emptyMap());
- Assert.assertEquals(1, (int) loadedStage.get(MyParams.INT_PARAM));
+
+ Assert.assertEquals(1, (int)
loadedStage.get(stage.paramWithNullDefault));
+ Assert.assertEquals(true, loadedStage.get(MyParams.BOOLEAN_PARAM));
+ Assert.assertEquals(50, (int) loadedStage.get(MyParams.INT_PARAM));
+ Assert.assertEquals(50L, (long) loadedStage.get(MyParams.LONG_PARAM));
+ Assert.assertEquals(Integer.MAX_VALUE + 50L, (long)
loadedStage.get(MyParams.LONG_PARAM2));
+ Assert.assertEquals(50f, loadedStage.get(MyParams.FLOAT_PARAM),
0.0001);
+ Assert.assertEquals(50, loadedStage.get(MyParams.DOUBLE_PARAM),
0.0001);
+ Assert.assertEquals("50", loadedStage.get(MyParams.STRING_PARAM));
+ Assert.assertArrayEquals(new Integer[] {50, 51},
loadedStage.get(MyParams.INT_ARRAY_PARAM));
+ Assert.assertArrayEquals(new Long[] {50L, 51L},
loadedStage.get(MyParams.LONG_ARRAY_PARAM));
+ Assert.assertArrayEquals(
+ new Float[] {50.0f, 51.0f},
loadedStage.get(MyParams.FLOAT_ARRAY_PARAM));
+ Assert.assertArrayEquals(
+ new Double[] {50.0, 51.0},
loadedStage.get(MyParams.DOUBLE_ARRAY_PARAM));
+ Assert.assertArrayEquals(
+ new String[] {"50", "51"},
loadedStage.get(MyParams.STRING_ARRAY_PARAM));
}
@Test
@@ -323,8 +366,9 @@ public class StageTest {
MyStage stage = new MyStage();
stage.set(stage.paramWithNullDefault, 1);
Stage<?> loadedStage =
- validateStageSaveLoad(tEnv, stage,
Collections.singletonMap("intParam", 10));
- Assert.assertEquals(10, (int) loadedStage.get(MyParams.INT_PARAM));
+ validateStageSaveLoad(
+ tEnv, stage,
Collections.singletonMap("paramWithNullDefault", 10));
+ Assert.assertEquals(10, (int)
loadedStage.get(stage.paramWithNullDefault));
}
@Test
diff --git a/flink-ml-dist/pom.xml b/flink-ml-dist/pom.xml
new file mode 100644
index 0000000..8cabfa1
--- /dev/null
+++ b/flink-ml-dist/pom.xml
@@ -0,0 +1,94 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+ http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ 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>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-ml-parent</artifactId>
+ <version>2.1-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>flink-ml-dist_${scala.binary.version}</artifactId>
+ <name>Flink ML : Dist</name>
+ <packaging>jar</packaging>
+
+ <dependencies>
+
+ <!-- Flink ML project binaries -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-ml-uber_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <!-- Stateful Functions Dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>statefun-flink-core</artifactId>
+ <version>3.1.0</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java_2.12</artifactId>
+ </exclusion>
+ </exclusions>
+ <scope>compile</scope>
+ </dependency>
+
+ </dependencies>
+
+ <build>
+ <pluginManagement>
+ <plugins>
+ <plugin>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <configuration>
+ <finalName>flink-ml-${project.version}-bin</finalName>
+ <appendAssemblyId>false</appendAssemblyId>
+ </configuration>
+ </plugin>
+ </plugins>
+ </pluginManagement>
+
+ <plugins>
+ <plugin>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>bin</id>
+ <phase>package</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ <configuration>
+ <descriptors>
+
<descriptor>src/main/assemblies/bin.xml</descriptor>
+ </descriptors>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
diff --git a/flink-ml-dist/src/main/assemblies/bin.xml
b/flink-ml-dist/src/main/assemblies/bin.xml
new file mode 100644
index 0000000..de9bc46
--- /dev/null
+++ b/flink-ml-dist/src/main/assemblies/bin.xml
@@ -0,0 +1,80 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+<assembly
+
xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+
xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0
http://maven.apache.org/xsd/assembly-1.1.0.xsd";>
+ <id>bin</id>
+ <formats>
+ <format>dir</format>
+ </formats>
+
+ <includeBaseDirectory>true</includeBaseDirectory>
+ <baseDirectory>flink-ml-${project.version}</baseDirectory>
+
+ <dependencySets>
+ <dependencySet>
+ <outputDirectory>lib</outputDirectory>
+ <unpack>false</unpack>
+
+ <includes>
+ <include>org.apache.flink:statefun-flink-core</include>
+
<include>org.apache.flink:flink-ml-uber_${scala.binary.version}</include>
+ </includes>
+ </dependencySet>
+ </dependencySets>
+
+ <files>
+ <file>
+ <source>../LICENSE</source>
+ <outputDirectory/>
+ <fileMode>0644</fileMode>
+ </file>
+ </files>
+
+ <fileSets>
+ <!-- copy scripts -->
+ <fileSet>
+ <directory>src/main/flink-ml-bin/bin</directory>
+ <outputDirectory>bin</outputDirectory>
+ <fileMode>0755</fileMode>
+ </fileSet>
+
+ <!-- copy benchmark json configuration -->
+ <fileSet>
+ <directory>../flink-ml-benchmark/src/main/resources/</directory>
+ <outputDirectory>conf/</outputDirectory>
+ <fileMode>0644</fileMode>
+ <includes>
+ <include>*.json</include>
+ </includes>
+ </fileSet>
+
+ <!-- copy *.txt files -->
+ <fileSet>
+ <directory>src/main/flink-ml-bin/</directory>
+ <outputDirectory/>
+ <fileMode>0644</fileMode>
+ <includes>
+ <include>*.txt</include>
+ </includes>
+ </fileSet>
+
+ </fileSets>
+</assembly>
diff --git a/flink-ml-dist/src/main/flink-ml-bin/README.txt
b/flink-ml-dist/src/main/flink-ml-bin/README.txt
new file mode 100644
index 0000000..ef1de7a
--- /dev/null
+++ b/flink-ml-dist/src/main/flink-ml-bin/README.txt
@@ -0,0 +1,32 @@
+For the latest information about Flink, please visit our website at:
+
+ https://flink.apache.org
+
+and our GitHub Account
+
+ https://github.com/apache/flink
+
+If you have any questions, ask on our Mailing lists:
+
+ [email protected]
+ [email protected]
+
+This distribution includes cryptographic software. The country in
+which you currently reside may have restrictions on the import,
+possession, use, and/or re-export to another country, of
+encryption software. BEFORE using any encryption software, please
+check your country's laws, regulations and policies concerning the
+import, possession, or use, and re-export of encryption software, to
+see if this is permitted. See <http://www.wassenaar.org/> for more
+information.
+
+The U.S. Government Department of Commerce, Bureau of Industry and
+Security (BIS), has classified this software as Export Commodity
+Control Number (ECCN) 5D002.C.1, which includes information security
+software using or performing cryptographic functions with asymmetric
+algorithms. The form and manner of this Apache Software Foundation
+distribution makes it eligible for export under the License Exception
+ENC Technology Software Unrestricted (TSU) exception (see the BIS
+Export Administration Regulations, Section 740.13) for both object
+code and source code.
+
diff --git a/flink-ml-dist/src/main/flink-ml-bin/bin/flink-ml-benchmark.sh
b/flink-ml-dist/src/main/flink-ml-bin/bin/flink-ml-benchmark.sh
new file mode 100644
index 0000000..fb6b357
--- /dev/null
+++ b/flink-ml-dist/src/main/flink-ml-bin/bin/flink-ml-benchmark.sh
@@ -0,0 +1,42 @@
+#!/usr/bin/env bash
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+
+current_path=$(pwd)
+flink_ml_bin_path="$( cd -- "$(dirname "$0")" >/dev/null 2>&1 ; pwd -P )"
+flink_ml_root_path="$(dirname "$flink_ml_bin_path")"
+
+# Checks FLINK_HOME.
+if [ "$FLINK_HOME" == "" ]; then
+ echo "FLINK_HOME not found. Please make sure you have installed Flink and
configured FLINK_HOME in your environment path."
+ exit 1
+fi
+
+# Checks flink version.
+expected_version="1.14"
+actual_version=`$FLINK_HOME/bin/flink --version | cut -d" " -f2 | tr -d ","`
+unsorted_versions="${expected_version}\n${actual_version}\n"
+sorted_versions=`printf ${unsorted_versions} | sort -V`
+unsorted_versions=`printf ${unsorted_versions}`
+if [ "${unsorted_versions}" != "${sorted_versions}" ]; then
+ echo "$flink_cmd $expected_version or a higher version is required, but
found $actual_version"
+ exit 1
+fi
+
+# Submits benchmark flink job.
+$FLINK_HOME/bin/flink run -c org.apache.flink.ml.benchmark.Benchmark
$FLINK_HOME/lib/flink-ml-uber*.jar ${@:1}
diff --git a/flink-ml-uber/pom.xml b/flink-ml-uber/pom.xml
index 1fb970b..95277b2 100644
--- a/flink-ml-uber/pom.xml
+++ b/flink-ml-uber/pom.xml
@@ -52,6 +52,12 @@ under the License.
<artifactId>flink-ml-lib_${scala.binary.version}</artifactId>
<version>${project.version}</version>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-ml-benchmark_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
</dependencies>
<build>
@@ -72,6 +78,7 @@ under the License.
<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>dev.ludovic.netlib:blas</include>
</includes>
</artifactSet>
diff --git a/pom.xml b/pom.xml
index f29d1e6..4e5fd02 100644
--- a/pom.xml
+++ b/pom.xml
@@ -57,6 +57,8 @@ under the License.
<module>flink-ml-lib</module>
<module>flink-ml-tests</module>
<module>flink-ml-uber</module>
+ <module>flink-ml-benchmark</module>
+ <module>flink-ml-dist</module>
</modules>
<properties>
@@ -343,6 +345,8 @@ under the License.
<exclude>**/LICENSE*</exclude>
<!-- artifacts created during release process -->
<exclude>release/**</exclude>
+ <!-- Test Data. -->
+ <exclude>flink-ml-benchmark/src/main/resources/*.json</exclude>
</excludes>
</configuration>
</plugin>
