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 0c93277 [FLINK-29115] Improve quickstart documents
0c93277 is described below
commit 0c93277a31a95b806c72b898416a5bc5b12ab357
Author: yunfengzhou-hub <[email protected]>
AuthorDate: Tue Nov 1 12:58:46 2022 +0800
[FLINK-29115] Improve quickstart documents
This closes #164.
---
.../{building.md => build-and-install.md} | 18 +-
.../try-flink-ml/java/build-your-own-project.md | 178 ++++++++++--
docs/content/docs/try-flink-ml/java/quick-start.md | 76 ++++--
.../try-flink-ml/python/build-your-own-project.md | 232 ----------------
.../docs/try-flink-ml/python/quick-start.md | 298 +++++++++++++++++++--
5 files changed, 492 insertions(+), 310 deletions(-)
diff --git a/docs/content/docs/development/building.md
b/docs/content/docs/development/build-and-install.md
similarity index 81%
rename from docs/content/docs/development/building.md
rename to docs/content/docs/development/build-and-install.md
index a91f7a1..cfddf24 100644
--- a/docs/content/docs/development/building.md
+++ b/docs/content/docs/development/build-and-install.md
@@ -1,9 +1,9 @@
---
-title: "Building Flink ML from Source"
+title: "Building And Installing Flink ML From Source"
weight: 999
type: docs
aliases:
-- /development/building.html
+- /development/build-and-install.html
---
@@ -26,11 +26,11 @@ specific language governing permissions and limitations
under the License.
-->
-# Building Flink ML from Source
+# Building And Installing Flink ML From Source
-This page covers how to build Flink ML from sources.
+This page covers how to build and install Flink ML from sources.
-## Build Flink ML Java SDK
+## Build and Install Java SDK
In order to build Flink ML you need the source code. Either [download the
source
of a release](https://flink.apache.org/downloads.html) or [clone the git
@@ -52,7 +52,7 @@ mvn clean install -DskipTests
```
This instructs [Maven](http://maven.apache.org/) (`mvn`) to first remove all
-existing builds (`clean`) and then create a new Flink binary (`install`).
+existing builds (`clean`) and then create a new Flink ML binary (`install`).
After the build finishes, you can acquire the build result in the following
path
from the root directory of Flink ML:
@@ -61,7 +61,11 @@ from the root directory of Flink ML:
./flink-ml-dist/target/flink-ml-*-bin/flink-ml*/
```
-## Build Flink ML Python SDK
+The `mvn clean install` command would have installed the binary into your local
+Maven repository so other projects can refer to it and grab it from the
+repository. There is no additional step required for installation.
+
+## Build and Install Python SDK
### Prerequisites
diff --git a/docs/content/docs/try-flink-ml/java/build-your-own-project.md
b/docs/content/docs/try-flink-ml/java/build-your-own-project.md
index 5525548..f3e4fb1 100644
--- a/docs/content/docs/try-flink-ml/java/build-your-own-project.md
+++ b/docs/content/docs/try-flink-ml/java/build-your-own-project.md
@@ -28,43 +28,152 @@ under the License.
This document provides a quick introduction to using Flink ML. Readers of this
document will be guided to create a simple Flink job that trains a Machine
-Learning Model and use it to provide prediction service.
+Learning Model and uses it to provide prediction service.
-## Maven Setup
+## What Will You Be Building?
-In order to use Flink ML in a Maven project, add the following dependencies to
-`pom.xml`.
+Kmeans is a widely-used clustering algorithm and has been supported by Flink
ML.
+This walkthrough guides you to create a Flink job with Flink ML that
initializes
+and trains a Kmeans model, and finally uses it to predict the cluster id of
+certain data points.
+
+## Prerequisites
+
+This walkthrough assumes that you have some familiarity with Java, but you
+should be able to follow along even if you are coming from a different
+programming language.
+
+## Help, I’m Stuck!
+
+If you get stuck, check out the [community support
+resources](https://flink.apache.org/gettinghelp.html). In particular, Apache
+Flink's [user mailing
+list](https://flink.apache.org/community.html#mailing-lists) is consistently
+ranked as one of the most active of any Apache project and a great way to get
+help quickly.
+
+## How To Follow Along
+
+If you want to follow along, you will require a computer with:
+
+- Java 8
+- Maven 3
+
+{{< unstable >}}
+
+Before walking through the following sections of this document, make sure you
+have downloaded Flink ML's latest code and installed Flink ML's Java SDK in
your
+local machine. You can refer to this [guideline]({{< ref
+"docs/development/build-and-install#build-and-install-java-sdk" >}}) for how to
+build and install Flink ML.
+
+{{< /unstable >}}
+
+While commands to be executed in a CLI are provided to walk through this
example
+in the following steps, it is recommended to use an IDE, like IntelliJ IDEA, to
+manage, build and execute the example codes below.
-{{< artifact flink-ml-uber >}}
+Please use the following command to create a Flink Maven Archetype that
provides
+the basic skeleton of a project, with some necessary Flink dependencies.
-The example code provided in this document requires additional dependencies on
-the Flink Table API. In order to execute the example code successfully, please
-make sure the following dependencies also exist in `pom.xml`.
+```shell
+$ mvn archetype:generate \
+ -DarchetypeGroupId=org.apache.flink \
+ -DarchetypeArtifactId=flink-quickstart-java \
+ -DarchetypeVersion=1.15.1 \
+ -DgroupId=kmeans-example \
+ -DartifactId=kmeans-example \
+ -Dversion=0.1 \
+ -Dpackage=myflinkml \
+ -DinteractiveMode=false
+```
+
+The command above would create a maven project named `kmeans-example` in your
+current directory with the following structure:
+
+```
+$ tree kmeans-example
+kmeans-example
+├── pom.xml
+└── src
+ └── main
+ ├── java
+ │ └── myflinkml
+ │ └── DataStreamJob.java
+ └── resources
+ └── log4j2.properties
+```
+
+Change the dependencies provided in `pom.xml` as follows:
```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-ml-uber</artifactId>
+ <version>2.2-SNAPSHOT</version>
+ <scope>provided</scope>
+</dependency>
+
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-files</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
+</dependency>
+
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
- <version>1.15.1</version>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
</dependency>
-```
-```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-java-bridge</artifactId>
+ <version>${flink.version}</version>
+ <scope>provided</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>1.15.1</version>
+ <version>${flink.version}</version>
+ <scope>provided</scope>
</dependency>
-```
-## Flink ML Example
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>statefun-flink-core</artifactId>
+ <version>3.2.0</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>
+```
-Kmeans is a widely-used clustering algorithm and has been supported by Flink
ML.
-The example code below creates a Flink job with Flink ML that initializes and
-trains a Kmeans model, and finally uses it to predict the cluster id of certain
-data points.
+Create file `src/main/java/myflinkml/KMeansExample.java`, and save the
following
+content into the file. You may feel free to ignore and delete
+`src/main/java/myflinkml/DataStreamJob.java` as it is not used in this
+walkthrough.
```java
+package myflinkml;
+
import org.apache.flink.ml.clustering.kmeans.KMeans;
import org.apache.flink.ml.clustering.kmeans.KMeansModel;
import org.apache.flink.ml.linalg.DenseVector;
@@ -76,7 +185,7 @@ import
org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
-public class QuickStart {
+public class KMeansExample {
public static void main(String[] args) {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
@@ -121,8 +230,25 @@ public class QuickStart {
}
```
-After placing the code above into your Maven project and executing it,
-information like below will be printed out to your terminal window.
+After placing the code above into your Maven project, you may use the following
+command or your IDE to build and execute the example job.
+
+```shell
+cd kmeans-example/
+mvn clean package
+mvn exec:java -Dexec.mainClass="myflinkml.KMeansExample"
-Dexec.classpathScope="compile"
+```
+
+If you are running the project in an IDE, you may get a
+`java.lang.NoClassDefFoundError` exception. This is probably because you do not
+have all required Flink dependencies implicitly loaded into the classpath.
+
+- IntelliJ IDEA: Go to Run > Edit Configurations > Modify options > Select
+ `include dependencies with "Provided" scope`. This run configuration will now
+ include all required classes to run the application from within the IDE.
+
+After executing the job, information like below will be printed out to your
+terminal window.
```
Vector: [0.3, 0.0] Cluster ID: 1
@@ -133,6 +259,12 @@ Vector: [0.0, 0.3] Cluster ID: 1
Vector: [9.0, 0.0] Cluster ID: 0
```
+<!-- TODO: figure out why the process above does not terminate. -->
+The program might get stuck after printing out the information above, and you
+may need to enter ^C to terminate the process. This only happens when the
+program is executed locally and would not happen when the job is submitted to a
+Flink cluster.
+
## Breaking Down The Code
### The Execution Environment
@@ -153,8 +285,8 @@ StreamTableEnvironment tEnv =
StreamTableEnvironment.create(env);
### Creating Training & Inference Data Table
Then the program creates the Table containing data for the training and
-prediction process of the following Kmeans algorithm. Flink ML operators
-search the names of the columns of the input table for input data, and produce
+prediction process of the following Kmeans algorithm. Flink ML operators search
+the names of the columns of the input table for input data, and produce
prediction results to designated column of the output Table.
```java
diff --git a/docs/content/docs/try-flink-ml/java/quick-start.md
b/docs/content/docs/try-flink-ml/java/quick-start.md
index 645a8ec..725df28 100644
--- a/docs/content/docs/try-flink-ml/java/quick-start.md
+++ b/docs/content/docs/try-flink-ml/java/quick-start.md
@@ -28,20 +28,38 @@ under the License.
This document provides a quick introduction to using Flink ML. Readers of this
document will be guided to submit a simple Flink job that trains a Machine
-Learning Model and use it to provide prediction service.
+Learning Model and uses it to provide prediction service.
+
+## Help, I’m Stuck!
+
+If you get stuck, check out the [community support
+resources](https://flink.apache.org/gettinghelp.html). In particular, Apache
+Flink's [user mailing
+list](https://flink.apache.org/community.html#mailing-lists) is consistently
+ranked as one of the most active of any Apache project and a great way to get
+help quickly.
## Prerequisites
-### Install Flink
+Make sure Java 8 or a higher version has been installed in your local machine.
+To check the Java version installed, type in your terminal:
-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.
+```shell
+$ java -version
+```
-### Set Up Flink Environment Variables
+## Download Flink
-After having installed Flink, please register `$FLINK_HOME` as an environment
+[Download 1.15 or a higher version of
+Flink](https://flink.apache.org/downloads.html), then extract the archive:
+
+```shell
+$ tar -xzf flink-*.tgz
+```
+
+## Set Up Flink Environment Variables
+
+After having downloaded Flink, please register `$FLINK_HOME` as an environment
variable into your local environment.
```bash
@@ -49,31 +67,43 @@ cd ${path_to_flink}
export FLINK_HOME=`pwd`
```
+## Add Flink ML library to Flink's library folder
-[//]: # (TODO: Add instructions to download binary distribution when release is
- available)
-### Build Flink ML library
+You need to copy Flink ML's library files to Flink's folder for proper
+initialization.
-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.
+{{< stable >}}
-```bash
+Please [download the correponding binary
+release](https://flink.apache.org/downloads.html) of Flink ML, then extract the
+archive:
+
+```shell
+tar -xzf flink-ml-*.tgz
+```
+
+Then you may copy the extracted library files to Flink's folder with the
+following commands.
+
+```shell
cd ${path_to_flink_ml}
-mvn clean package -DskipTests
-cd ./flink-ml-dist/target/flink-ml-*-bin/flink-ml*/
+cp ./lib/*.jar $FLINK_HOME/lib/
```
-### Add Flink ML binaries to Flink
+{{< /stable >}} {{< unstable >}}
-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.
+Please walk through this [guideline]({{< ref
+"docs/development/build-and-install#build-and-install-java-sdk" >}}) to build
+Flink ML's Java SDK. After that, you may copy the generated library files to
+Flink's folder with the following commands.
-```bash
-cp ./lib/*.jar $FLINK_HOME/lib/
+```shell
+cd ${path_to_flink_ml}
+cp ./flink-ml-dist/target/flink-ml-*-bin/flink-ml*/lib/*.jar $FLINK_HOME/lib/
```
+{{< /unstable >}}
+
## Run Flink ML example job
Please start a Flink standalone cluster in your local environment with the
diff --git a/docs/content/docs/try-flink-ml/python/build-your-own-project.md
b/docs/content/docs/try-flink-ml/python/build-your-own-project.md
deleted file mode 100644
index e13a687..0000000
--- a/docs/content/docs/try-flink-ml/python/build-your-own-project.md
+++ /dev/null
@@ -1,232 +0,0 @@
----
-title: "Building your own Flink ML project"
-weight: 2
-type: docs
-aliases:
-- /try-flink-ml/python/building-your-own-project.html
-
----
-
-<!--
-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.
--->
-
-# Building your own Flink ML project
-
-This document provides a quick introduction to using Flink ML. Readers of this
-document will be guided to create a simple Flink job that trains a Machine
-Learning Model and use it to provide prediction service.
-
-## Prerequisites
-
-Python version (3.6, 3.7, or 3.8) is required for Flink ML. Please run the
-following command to make sure that it meets the requirements:
-
-```shell
-$ python --version
-# the version printed here must be 3.6, 3.7 or 3.8
-```
-
-## Installation of Flink ML Python SDK
-
-Flink ML Python SDK is available in
-[PyPi](https://pypi.org/project/apache-flink-ml/) and can be installed as
-follows:
-
-{{< stable >}}
-
-```bash
-$ python -m pip install apache-flink-ml=={{< version >}}
-```
-
-{{< /stable >}} {{< unstable >}}
-
-```bash
-$ python -m pip install apache-flink-ml
-```
-
-{{< /unstable >}}
-
-You can also build Flink ML Python SDK from sources by following the
-[development guide]({{< ref "docs/development/building" >}}).
-
-## Flink ML Example
-
-Kmeans is a widely-used clustering algorithm and has been supported by Flink
ML.
-The example code below creates a Flink job with Flink ML that initializes and
-trains a Kmeans model, and finally uses it to predict the cluster id of certain
-data points.
-
-```python
-from pyflink.common import Types
-from pyflink.datastream import StreamExecutionEnvironment
-from pyflink.ml.core.linalg import Vectors, DenseVectorTypeInfo
-from pyflink.ml.lib.clustering.kmeans import KMeans
-from pyflink.table import StreamTableEnvironment
-
-# create a new StreamExecutionEnvironment
-env = StreamExecutionEnvironment.get_execution_environment()
-
-# create a StreamTableEnvironment
-t_env = StreamTableEnvironment.create(env)
-
-# generate input data
-input_data = t_env.from_data_stream(
- env.from_collection([
- (Vectors.dense([0.0, 0.0]),),
- (Vectors.dense([0.0, 0.3]),),
- (Vectors.dense([0.3, 3.0]),),
- (Vectors.dense([9.0, 0.0]),),
- (Vectors.dense([9.0, 0.6]),),
- (Vectors.dense([9.6, 0.0]),),
- ],
- type_info=Types.ROW_NAMED(
- ['features'],
- [DenseVectorTypeInfo()])))
-
-# create a kmeans object and initialize its parameters
-kmeans = KMeans().set_k(2).set_seed(1)
-
-# train the kmeans model
-model = kmeans.fit(input_data)
-
-# use the kmeans model for predictions
-output = model.transform(input_data)[0]
-
-# extract and display the results
-field_names = output.get_schema().get_field_names()
-for result in t_env.to_data_stream(output).execute_and_collect():
- features = result[field_names.index(kmeans.get_features_col())]
- cluster_id = result[field_names.index(kmeans.get_prediction_col())]
- print('Features: ' + str(features) + ' \tCluster Id: ' + str(cluster_id))
-```
-
-After placing the code above into your Python file and executing it,
information
-like the below will be printed out to your terminal window.
-
-```
-Vector: [0.3, 0.0] Cluster ID: 1
-Vector: [9.6, 0.0] Cluster ID: 0
-Vector: [9.0, 0.6] Cluster ID: 0
-Vector: [0.0, 0.0] Cluster ID: 1
-Vector: [0.0, 0.3] Cluster ID: 1
-Vector: [9.0, 0.0] Cluster ID: 0
-```
-
-## Breaking Down The Code
-
-### The Execution Environment
-
-The first lines set up the `StreamExecutionEnvironment` to execute the Flink ML
-job. You would have been familiar with this concept if you have experience
using
-Flink. For the example program in this document, a simple
-`StreamExecutionEnvironment` without specific configurations would be enough.
-
-Given that Flink ML uses Flink's Table API, a `StreamTableEnvironment` would
-also be necessary for the following program.
-
-```python
-# create a new StreamExecutionEnvironment
-env = StreamExecutionEnvironment.get_execution_environment()
-
-# create a StreamTableEnvironment
-t_env = StreamTableEnvironment.create(env)
-```
-
-### Creating Training & Inference Data Table
-
-Then the program creates the Table containing data for the training and
-prediction process of the following Kmeans algorithm. Flink ML operators search
-the names of the columns of the input table for input data, and produce
-prediction results to designated column of the output Table.
-
-```python
-# generate input data
-input_data = t_env.from_data_stream(
- env.from_collection([
- (Vectors.dense([0.0, 0.0]),),
- (Vectors.dense([0.0, 0.3]),),
- (Vectors.dense([0.3, 3.0]),),
- (Vectors.dense([9.0, 0.0]),),
- (Vectors.dense([9.0, 0.6]),),
- (Vectors.dense([9.6, 0.0]),),
- ],
- type_info=Types.ROW_NAMED(
- ['features'],
- [DenseVectorTypeInfo()])))
-```
-
-### Creating, Configuring, Training & Using Kmeans
-
-Flink ML classes for Kmeans algorithm include `KMeans` and `KMeansModel`.
-`KMeans` implements the training process of Kmeans algorithm based on the
-provided training data, and finally generates a `KMeansModel`.
-`KmeansModel.transform()` method encodes the Transformation logic of this
-algorithm and is used for predictions.
-
-Both `KMeans` and `KMeansModel` provides getter/setter methods for Kmeans
-algorithm's configuration parameters. The example program explicitly sets the
-following parameters, and other configuration parameters will have their
default
-values used.
-
-- `k`, the number of clusters to create
-- `seed`, the random seed to initialize cluster centers
-
-When the program invokes `KMeans.fit()` to generate a `KMeansModel`, the
-`KMeansModel` will inherit the `KMeans` object's configuration parameters. Thus
-it is supported to set `KMeansModel`'s parameters directly in `KMeans` object.
-
-```python
-# create a kmeans object and initialize its parameters
-kmeans = KMeans().set_k(2).set_seed(1)
-
-# train the kmeans model
-model = kmeans.fit(input_data)
-
-# use the kmeans model for predictions
-output = model.transform(input_data)[0]
-
-```
-
-### Collecting Prediction Result
-
-Like all other Flink programs, the codes described in the sections above only
-configures the computation graph of a Flink job, and the program only evaluates
-the computation logic and collects outputs after the `execute()` method is
-invoked. Collected outputs from the output table would be `Row`s in which
-`featuresCol` contains input feature vectors, and `predictionCol` contains
-output prediction results, i.e., cluster IDs.
-
-```python
-# extract and display the results
-field_names = output.get_schema().get_field_names()
-for result in t_env.to_data_stream(output).execute_and_collect():
- features = result[field_names.index(kmeans.get_features_col())]
- cluster_id = result[field_names.index(kmeans.get_prediction_col())]
- print('Features: ' + str(features) + ' \tCluster Id: ' + str(cluster_id))
-```
-
-```
-Features: [9.6,0.0] Cluster Id: 0
-Features: [9.0,0.6] Cluster Id: 0
-Features: [0.0,0.3] Cluster Id: 1
-Features: [0.0,0.0] Cluster Id: 1
-Features: [0.3,3.0] Cluster Id: 1
-Features: [9.0,0.0] Cluster Id: 0
-```
-
diff --git a/docs/content/docs/try-flink-ml/python/quick-start.md
b/docs/content/docs/try-flink-ml/python/quick-start.md
index bb632fa..bc9ef71 100644
--- a/docs/content/docs/try-flink-ml/python/quick-start.md
+++ b/docs/content/docs/try-flink-ml/python/quick-start.md
@@ -29,54 +29,299 @@ under the License.
# Quick Start
This document provides a quick introduction to using Flink ML. Readers of this
-document will be guided to submit a simple Flink job that trains a Machine
-Learning Model and use it to provide prediction service.
+document will be guided to create a simple Flink job that trains a Machine
+Learning Model and uses it to provide prediction service.
+
+## What Will You Be Building?
+
+Kmeans is a widely-used clustering algorithm and has been supported by Flink
ML.
+This walkthrough guides you to create a Flink job with Flink ML that
initializes
+and trains a Kmeans model, and finally uses it to predict the cluster id of
+certain data points.
## Prerequisites
-Python version (3.6, 3.7, or 3.8) is required for Flink ML. Please run the
-following command to make sure that it meets the requirements:
+This walkthrough assumes that you have some familiarity with Python, but you
+should be able to follow along even if you come from a different programming
+language.
-```shell
-$ python --version
-# the version printed here must be 3.6, 3.7 or 3.8
-```
+## Help, I’m Stuck!
+
+If you get stuck, check out the [community support
+resources](https://flink.apache.org/gettinghelp.html). In particular, Apache
+Flink's [user mailing
+list](https://flink.apache.org/community.html#mailing-lists) is consistently
+ranked as one of the most active of any Apache project and a great way to get
+help quickly.
+
+## How To Follow Along
-## Installation of Flink ML Python SDK
+If you want to follow along, you will require a computer with:
-Flink ML Python SDK is available in
-[PyPi](https://pypi.org/project/apache-flink-ml/) and can be installed as
-follows:
+{{< stable >}}
+- Java 8
+- Python 3.6, 3.7 or 3.8 {{< /stable >}} {{< unstable >}}
+- Java 8
+- Maven 3
+- Python 3.6, 3.7 or 3.8 {{< /unstable >}}
{{< stable >}}
+This walkthrough requires installing Flink ML Python SDK, which is available on
+[PyPi](https://pypi.org/project/apache-flink-ml/) and can be easily installed
+using pip.
+
```bash
$ python -m pip install apache-flink-ml=={{< version >}}
```
{{< /stable >}} {{< unstable >}}
+Please walk through this [guideline]({{< ref
+"docs/development/build-and-install#build-and-install-python-sdk" >}}) to build
+and install Flink ML's Python SDK in your local environment.
+
+{{< /unstable >}}
+
+## Writing a Flink ML Python Program
+
+Flink ML programs begin by setting up the `StreamExecutionEnvironment` to
+execute the Flink ML job. You would have been familiar with this concept if you
+have experience using Flink. For the example program in this document, a simple
+`StreamExecutionEnvironment` without specific configurations would be enough.
+
+Given that Flink ML uses Flink's Table API, a `StreamTableEnvironment` would
+also be necessary for the following program.
+
+```python
+# create a new StreamExecutionEnvironment
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# create a StreamTableEnvironment
+t_env = StreamTableEnvironment.create(env)
+```
+
+Then you can create the Table containing data for the training and prediction
+process of the following Kmeans algorithm. Flink ML operators search the names
+of the columns of the input table for input data, and produce prediction
results
+to designated column of the output Table.
+
+```python
+# generate input data
+input_data = t_env.from_data_stream(
+ env.from_collection([
+ (Vectors.dense([0.0, 0.0]),),
+ (Vectors.dense([0.0, 0.3]),),
+ (Vectors.dense([0.3, 3.0]),),
+ (Vectors.dense([9.0, 0.0]),),
+ (Vectors.dense([9.0, 0.6]),),
+ (Vectors.dense([9.6, 0.0]),),
+ ],
+ type_info=Types.ROW_NAMED(
+ ['features'],
+ [DenseVectorTypeInfo()])))
+```
+
+Flink ML classes for Kmeans algorithm include `KMeans` and `KMeansModel`.
+`KMeans` implements the training process of Kmeans algorithm based on the
+provided training data, and finally generates a `KMeansModel`.
+`KmeansModel.transform()` method encodes the Transformation logic of this
+algorithm and is used for predictions.
+
+Both `KMeans` and `KMeansModel` provides getter/setter methods for Kmeans
+algorithm's configuration parameters. This example program explicitly sets the
+following parameters, and other configuration parameters will have their
default
+values used.
+
+- `k`, the number of clusters to create
+- `seed`, the random seed to initialize cluster centers
+
+When the program invokes `KMeans.fit()` to generate a `KMeansModel`, the
+`KMeansModel` will inherit the `KMeans` object's configuration parameters. Thus
+it is supported to set `KMeansModel`'s parameters directly in `KMeans` object.
+
+```python
+# create a kmeans object and initialize its parameters
+kmeans = KMeans().set_k(2).set_seed(1)
+
+# train the kmeans model
+model = kmeans.fit(input_data)
+
+# use the kmeans model for predictions
+output = model.transform(input_data)[0]
+
+```
+
+Like all other Flink programs, the codes described in the sections above only
+configures the computation graph of a Flink job, and the program only evaluates
+the computation logic and collects outputs after the `execute()` method is
+invoked. Collected outputs from the output table would be `Row`s in which
+`featuresCol` contains input feature vectors, and `predictionCol` contains
+output prediction results, i.e., cluster IDs.
+
+```python
+# extract and display the results
+field_names = output.get_schema().get_field_names()
+for result in t_env.to_data_stream(output).execute_and_collect():
+ features = result[field_names.index(kmeans.get_features_col())]
+ cluster_id = result[field_names.index(kmeans.get_prediction_col())]
+ print('Features: ' + str(features) + ' \tCluster Id: ' + str(cluster_id))
+```
+
+The complete code so far:
+
+```python
+from pyflink.common import Types
+from pyflink.datastream import StreamExecutionEnvironment
+from pyflink.ml.core.linalg import Vectors, DenseVectorTypeInfo
+from pyflink.ml.lib.clustering.kmeans import KMeans
+from pyflink.table import StreamTableEnvironment
+
+# create a new StreamExecutionEnvironment
+env = StreamExecutionEnvironment.get_execution_environment()
+
+# create a StreamTableEnvironment
+t_env = StreamTableEnvironment.create(env)
+
+# generate input data
+input_data = t_env.from_data_stream(
+ env.from_collection([
+ (Vectors.dense([0.0, 0.0]),),
+ (Vectors.dense([0.0, 0.3]),),
+ (Vectors.dense([0.3, 3.0]),),
+ (Vectors.dense([9.0, 0.0]),),
+ (Vectors.dense([9.0, 0.6]),),
+ (Vectors.dense([9.6, 0.0]),),
+ ],
+ type_info=Types.ROW_NAMED(
+ ['features'],
+ [DenseVectorTypeInfo()])))
+
+# create a kmeans object and initialize its parameters
+kmeans = KMeans().set_k(2).set_seed(1)
+
+# train the kmeans model
+model = kmeans.fit(input_data)
+
+# use the kmeans model for predictions
+output = model.transform(input_data)[0]
+
+# extract and display the results
+field_names = output.get_schema().get_field_names()
+for result in t_env.to_data_stream(output).execute_and_collect():
+ features = result[field_names.index(kmeans.get_features_col())]
+ cluster_id = result[field_names.index(kmeans.get_prediction_col())]
+ print('Features: ' + str(features) + ' \tCluster Id: ' + str(cluster_id))
+```
+
+## Executing a Flink ML Python Program locally
+
+After creating a python file (e.g. kmeans_example.py) and saving the code above
+into the file, you can run the example on the command line:
+
+```shell
+python kmeans_example.py
+```
+
+The command above would build the example job and run it in a local mini
+cluster. A sample output in your terminal is as follows.
+
+```
+Features: [9.6,0.0] Cluster Id: 0
+Features: [9.0,0.6] Cluster Id: 0
+Features: [0.0,0.3] Cluster Id: 1
+Features: [0.0,0.0] Cluster Id: 1
+Features: [0.3,3.0] Cluster Id: 1
+Features: [9.0,0.0] Cluster Id: 0
+```
+
+## Executing a Flink ML Python Program on a Flink Cluster
+
+### Prerequisites
+
+Make sure Java 8 or a higher version has been installed in your local machine.
+To check the Java version installed, type in your terminal:
+
+```shell
+$ java -version
+```
+
+### Download Flink
+
+[Download 1.15 or a higher version of
+Flink](https://flink.apache.org/downloads.html), then extract the archive:
+
+```shell
+$ tar -xzf flink-*.tgz
+```
+
+### Set Up Flink Environment Variables
+
+After having downloaded Flink, please register `$FLINK_HOME` as an environment
+variable into your local environment.
+
```bash
-$ python -m pip install apache-flink-ml
+cd ${path_to_flink}
+export FLINK_HOME=`pwd`
```
-{{< /unstable >}}
+### Add Flink ML library to Flink's library folder
+
+You need to copy Flink ML's library files to Flink's folder for proper
+initialization.
-You can also build Flink ML Python SDK from sources by following the
-[development guide]({{< ref "docs/development/building" >}}).
+{{< stable >}}
+
+Please [download the correponding binary
+release](https://flink.apache.org/downloads.html) of Flink ML, then extract the
+archive:
+
+```shell
+tar -xzf flink-ml-*.tgz
+```
+
+Then you may copy the extracted library files to Flink's folder with the
+following commands.
-## Run Flink ML example job
+```shell
+cd ${path_to_flink_ml}
+cp ./lib/*.jar $FLINK_HOME/lib/
+```
+
+{{< /stable >}} {{< unstable >}}
-After setting up Flink ML Python SDK, you can run a Flink ML example job as
-follows.
+Given that you have followed this [guideline]({{< ref
+"docs/development/build-and-install#build-and-install-python-sdk" >}}), you
+would have already built Flink ML's Java SDK. Now, you need to copy the
+generated library files to Flink's folder with the following commands.
```shell
-$ python -m pyflink.examples.ml.clustering.kmeans_example
+cd ${path_to_flink_ml}
+cp ./flink-ml-dist/target/flink-ml-*-bin/flink-ml*/lib/*.jar $FLINK_HOME/lib/
+```
+
+{{< /unstable >}}
+
+### Run Flink ML job
+
+Please start a Flink standalone cluster in your local environment with the
+following command.
+
+```bash
+$FLINK_HOME/bin/start-cluster.sh
```
-The command above would create a Flink mini-cluster and execute Flink ML’s
-`kmeans_example` job. There are also example jobs for other Flink ML
algorithms,
-and you can find them in `pyflink.ml.examples` module.
+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.
+
+
+After creating a python file (e.g. kmeans_example.py) and saving the code above
+into the file, you may submit the example job to the cluster as follows.
+
+```bash
+$FLINK_HOME/bin/flink run -py kmeans_example.py
+```
A sample output in your terminal is as follows.
@@ -87,8 +332,11 @@ Features: [0.0,0.3] Cluster Id: 1
Features: [0.0,0.0] Cluster Id: 1
Features: [0.3,3.0] Cluster Id: 1
Features: [9.0,0.0] Cluster Id: 0
-
```
-Now you have successfully run a Flink ML job.
+Now you have successfully run the Flink ML job on a Flink Cluster. Other
+detailed instructions to submit it to a Flink cluster can be found in [Job
+Submission
+Examples](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/cli/#submitting-pyflink-jobs).
+
