Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up

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

License

NotificationsYou must be signed in to change notification settings

alibaba/Alink

Repository files navigation

English| 简体中文

Alink

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

Alink教程

开源算法列表

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。

####安装步骤：

  1. 确保使用环境中有Python3，版本限于 3.6，3.7 和 3.8。
  2. 确保使用环境中安装有 Java 8。
  3. 使用 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。

安装注意事项：

  1. pyalink 和pyalink-flink-*** 不能同时安装，也不能与旧版本同时安装。如果之前安装过pyalink 或者pyalink-flink-***，请使用pip uninstall pyalink 或者pip uninstall pyalink-flink-*** 卸载之前的版本。
  2. 出现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)
  3. 如果有多个版本的 Python，可能需要使用特定版本的pip，比如pip3；如果使用 Anaconda，则需要在 Anaconda 命令行中进行安装。

开始使用：

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

使用步骤：

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

示例代码

StringURL ="https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/iris.csv";StringSCHEMA_STR ="sepal_length double, sepal_width double, petal_length double, petal_width double, category string";BatchOperatordata =newCsvSourceBatchOp()        .setFilePath(URL)        .setSchemaStr(SCHEMA_STR);VectorAssemblerva =newVectorAssembler()        .setSelectedCols(newString[]{"sepal_length","sepal_width","petal_length","petal_width"})        .setOutputCol("features");KMeanskMeans =newKMeans().setVectorCol("features").setK(3)        .setPredictionCol("prediction_result")        .setPredictionDetailCol("prediction_detail")        .setReservedCols("category")        .setMaxIter(100);Pipelinepipeline =newPipeline().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算法

  1. 准备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
  1. 准备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
  1. 运行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

部署

集群部署

[8]ページ先頭
©2009-2025 Movatter.jp