Alink

Introduction: Alink is the Machine Learning algorithm platform based on Flink, developed by the PAI team of Alibaba computing platform.

More: Author ReportBugs

Tags:

English| 简体中文

Alink 是基于 Flink 的通用算法平台,由阿里巴巴计算平台 PAI 团队研发,欢迎大家加入 Alink 开源用户钉钉群进行交流。

Alink 组件列表：http://alinklab.cn/manual/index.html
Alink 教程：http://alinklab.cn/tutorial/index.html
Alink 插件下载器：https://www.yuque.com/pinshu/alink_guide/plugin_downloader

Alink 教程

Alink 教程（Java 版）：http://alinklab.cn/tutorial/book_java.html
Alink 教程（Python 版）：http://alinklab.cn/tutorial/book_python.html
源代码地址：https://github.com/alibaba/Alink/tree/master/tutorial
Java 版的数据和资料链接：http://alinklab.cn/tutorial/book_java_00_reference.html
Python 版的数据和资料链接：http://alinklab.cn/tutorial/book_python_00_reference.html
Alink 教程(Java 版)代码的运行攻略 http://alinklab.cn/tutorial/book_java_00_code_help.html
Alink 教程(Python 版)代码的运行攻略 http://alinklab.cn/tutorial/book_python_00_code_help.html

开源算法列表

PyAlink 使用截图

快速开始

PyAlink 使用介绍

使用前准备：

包名和版本说明：

PyAlink 根据 Alink 所支持的 Flink 版本提供不同的 Python 包：其中，pyalink 包对应为 Alink 所支持的最新 Flink 版本，当前为 1.13，而 pyalink-flink-*** 为旧版本的 Flink 版本，当前提供 pyalink-flink-1.12, pyalink-flink-1.11, pyalink-flink-1.10 和 pyalink-flink-1.9。
Python 包的版本号与 Alink 的版本号一致，例如1.6.2。

安装步骤：

确保使用环境中有 Python3，版本限于 3.6，3.7 和 3.8。
确保使用环境中安装有 Java 8。
使用 pip 命令进行安装： pip install pyalink、pip install pyalink-flink-1.12、pip install pyalink-flink-1.11、pip install pyalink-flink-1.10 或者 pip install pyalink-flink-1.9。

安装注意事项：

pyalink 和 pyalink-flink-*** 不能同时安装，也不能与旧版本同时安装。如果之前安装过 pyalink 或者 pyalink-flink-***，请使用pip uninstall pyalink 或者 pip uninstall pyalink-flink-*** 卸载之前的版本。
出现pip安装缓慢或不成功的情况，可以参考这篇文章修改 pip 源，或者直接使用下面的链接下载 whl 包，然后使用 pip 安装：
- Flink 1.13：链接 (MD5: d4b7b1fe6474b11ca7f45d0fb0daf5bc)
- Flink 1.12：链接 (MD5: 527b9ac24383ccc8593cd61b06cc610d)
- Flink 1.11：链接 (MD5: 7e59ba00b3739386996cf55d8f522ed2)
- Flink 1.10：链接 (MD5: 6d5d9048c9a44f27285467c5117e8deb)
- Flink 1.9: 链接 (MD5: e89ac35a6a1c63c0426f3d9ca1025880)
如果有多个版本的 Python，可能需要使用特定版本的 pip，比如 pip3；如果使用 Anaconda，则需要在 Anaconda 命令行中进行安装。

开始使用：

可以通过 Jupyter Notebook 来开始使用 PyAlink，能获得更好的使用体验。

使用步骤：

在命令行中启动 Jupyter：jupyter notebook，并新建 Python 3 的 Notebook 。
导入 pyalink 包：from pyalink.alink import *。
使用方法创建本地运行环境： useLocalEnv(parallism, flinkHome=None, config=None)。其中，参数 parallism 表示执行所使用的并行度；flinkHome 为 flink 的完整路径，一般情况不需要设置；config为 Flink 所接受的配置参数。运行后出现如下所示的输出，表示初始化运行环境成功：
```
JVM listening on ***
```

开始编写 PyAlink 代码，例如：

source = CsvSourceBatchOp()\
 .setSchemaStr("sepal_length double, sepal_width double, petal_length double, petal_width double, category string")\
 .setFilePath("https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/iris.csv")
res = source.select(["sepal_length", "sepal_width"])
df = res.collectToDataframe()
print(df)

编写代码：

在 PyAlink 中，算法组件提供的接口基本与 Java API 一致，即通过默认构造方法创建一个算法组件，然后通过 setXXX 设置参数，通过 link/linkTo/linkFrom 与其他组件相连。这里利用 Jupyter Notebook 的自动补全机制可以提供书写便利。

对于批式作业，可以通过批式组件的 print/collectToDataframe/collectToDataframes 等方法或者 BatchOperator.execute() 来触发执行；对于流式作业，则通过 StreamOperator.execute() 来启动作业。

Java 接口使用介绍

示例代码

String URL = "https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/iris.csv";
String SCHEMA_STR = "sepal_length double, sepal_width double, petal_length double, petal_width double, category string";

BatchOperator data = new CsvSourceBatchOp()
        .setFilePath(URL)
        .setSchemaStr(SCHEMA_STR);

VectorAssembler va = new VectorAssembler()
        .setSelectedCols(new String[]{"sepal_length", "sepal_width", "petal_length", "petal_width"})
        .setOutputCol("features");

KMeans kMeans = new KMeans().setVectorCol("features").setK(3)
        .setPredictionCol("prediction_result")
        .setPredictionDetailCol("prediction_detail")
        .setReservedCols("category")
        .setMaxIter(100);

Pipeline pipeline = new Pipeline().add(va).add(kMeans);
pipeline.fit(data).transform(data).print();

Flink-1.13 的 Maven 依赖

<dependency>
    <groupId>com.alibaba.alink</groupId>
    <artifactId>alink_core_flink-1.13_2.11</artifactId>
    <version>1.6.2</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_2.11</artifactId>
    <version>1.13.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner_2.11</artifactId>
    <version>1.13.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients_2.11</artifactId>
    <version>1.13.0</version>
</dependency>

Flink-1.12 的 Maven 依赖

<dependency>
    <groupId>com.alibaba.alink</groupId>
    <artifactId>alink_core_flink-1.12_2.11</artifactId>
    <version>1.6.2</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_2.11</artifactId>
    <version>1.12.1</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner_2.11</artifactId>
    <version>1.12.1</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients_2.11</artifactId>
    <version>1.12.1</version>
</dependency>

Flink-1.11 的 Maven 依赖

<dependency>
    <groupId>com.alibaba.alink</groupId>
    <artifactId>alink_core_flink-1.11_2.11</artifactId>
    <version>1.6.2</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_2.11</artifactId>
    <version>1.11.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner_2.11</artifactId>
    <version>1.11.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients_2.11</artifactId>
    <version>1.11.0</version>
</dependency>

Flink-1.10 的 Maven 依赖

<dependency>
    <groupId>com.alibaba.alink</groupId>
    <artifactId>alink_core_flink-1.10_2.11</artifactId>
    <version>1.6.2</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_2.11</artifactId>
    <version>1.10.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner_2.11</artifactId>
    <version>1.10.0</version>
</dependency>

Flink-1.9 的 Maven 依赖

<dependency>
    <groupId>com.alibaba.alink</groupId>
    <artifactId>alink_core_flink-1.9_2.11</artifactId>
    <version>1.6.2</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-scala_2.11</artifactId>
    <version>1.9.0</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner_2.11</artifactId>
    <version>1.9.0</version>
</dependency>

快速开始在集群上运行 Alink 算法

准备 Flink 集群

wget https://archive.apache.org/dist/flink/flink-1.13.0/flink-1.13.0-bin-scala_2.11.tgz
tar -xf flink-1.13.0-bin-scala_2.11.tgz && cd flink-1.13.0
./bin/start-cluster.sh

准备 Alink 算法包

git clone https://github.com/alibaba/Alink.git
# add <scope>provided</scope> in pom.xml of alink_examples.
cd Alink && mvn -Dmaven.test.skip=true clean package shade:shade

运行 Java 示例

./bin/flink run -p 1 -c com.alibaba.alink.ALSExample [path_to_Alink]/examples/target/alink_examples-1.5-SNAPSHOT.jar
# ./bin/flink run -p 1 -c com.alibaba.alink.GBDTExample [path_to_Alink]/examples/target/alink_examples-1.5-SNAPSHOT.jar
# ./bin/flink run -p 1 -c com.alibaba.alink.KMeansExample [path_to_Alink]/examples/target/alink_examples-1.5-SNAPSHOT.jar