导读： Flink Machine Learning Library 提供了机器学习相关的 API 与基础设施，方便用户构建机器学习的工作流。用户可以在 Flink ML Library 提供的 API 的基础之上，将机器学习算法封装为 Flink 的算子，并且构建相关的工作流来运行训练或推理服务。

本文主要介绍：

• Flink ML Java API 的设计与使用
• Flink ML Python 介绍
• Flink ML 的生态建设与未来发展方向

分享嘉宾｜黄兴勃/周云峰 阿里巴巴
编辑整理｜晏世千
出品平台｜DataFunTalk

01 Flink ML Java API 的设计与使用

1. 为什么需要 Flink ML API？

为了方便复用算法模块 ，将每个机器学习算法封装为一个模块，通过这种方式来减少重复开发从而提升社区合作的效率。同时用户只需要调用 API 即可使用相应的算法，而算法的实现细节对用户来说是隐藏的，同时不同的算法使用同一套 API 进行描述，统一算法的使用体验，有助于提升业务的开发效率。

另外，可以将不同的算法模块进行组合封装，从而可以实现包含多个机器学习模块的复杂任务，同时除了通过编程方式之外，API 还支持以 JSON 格式来表达不同算法模块的配置和组合， 有利于提升算法的开发效率 。

2. Flink ML 的 API 设计

WithParams 接口是所有的 API 中， 最上层的接口，它是 Flink ML 用来存取参数的 API 。通过继承这一接口及其子类，不同的算子能够通过遵循同一规则的 API 来设置参数，并且可以将参数通过 JSON 的格式进行存取。Flink ML 将一些常见的算法配置参数，比如训练时候的迭代数，或者说是训练时候的收敛的阈值等，以通用的 WithParams 子类的方式来提供，算法开发者在实现算法的时候，就可以通过继承这些子类的方式来使实现类具备这些参数所需的功能，从而减少重复开发的工作量。

Flink ML 提供的 Stage 接口，代表的是一个算法模块的基本节点。Stage 提供了用来从文件系统中保存和恢复及 Save Load 相关的方法，这些方法能够辅助将一个算法模块的配置参数以及训练后得到的有界数据流保存文件之中，用于备份或者在不同的物理机之间进行传输。

Stage 接口之后，根据算法模块的目的和功能的不同，Flink ML 提供了两套子接口，其中 Estimator 所代表的算法具备训练相关的语义，而 AlgoOperator 及其子类代表的是推理相关的语义 。Estimator 提供一个 FIT 方法，这个方法可以接受由多个 Table 组成的训练数据，然后进行训练得到相关的模型数据，并且创建出使用这些模型数据进行推理服务的 Model 实例。

AlgoOperator 代表的是推理语义，它提供一个无模型语义的抽象类，它的推理语义是通过 API 定义的 Transform 方法，该方法可以接受由多个 Table 所表示的待推理数据，并且返回相应的推理结果。

Transformer 接口与 AlgoOperator 的接口是相同的，也是通过 Transform 方法来提供推理服务。它和 AlgoOperator 接口的不同点主要在于如果一个算法对于每条输入的推理数据都会输出一条相应的推理结果的话，适合使用 Transformer 来进行表达，而如果这个算法对于每条输入的数据可以产生多条结果，或者说是对于多条推理数据做一个聚合的话，那这样的算子更适合使用 AlgoOperator 来进行表达。

Model 类是 Transformer 语义，在提供推理功能的同时，它的推理功能适合模型数据相关的，它额外提供了 API 用来设置或者获取模型数据流。用户可以使用 SetModelData 来设置一个 Model 实例的模型数据。如果说某个 Model 实例是通过 Estimator.fit() 方法得到的话，那么这个实例的内部则是已经被设置了 Estimator 算子训练时所得到的模型数据流。

以上类是算法的核心类，每个具体算法都是 Stage 的某一个子类的具体实现 。这里所提到的所有的 API，他们的方法都是使用 Flink Table API 来表示的，这样的话，一方面是为了能够支持流批一体的语义，在实现流批一体的算子的时候，能够保持 Flink ML API 的相对稳定，同时能够让 Flink ML 使用 Flink 对于 Table 相关的一些优化方案，并且这里的每一个 API 都使用多个table来表示参数或者返回值，这也使 Flink ML API 能够表示多输入多输出算相关的意义。

3. Flink ML 训练与部署模型进行推理流处理的流程

使用上述核心的 Flink ML API 来训练或者部署模型进行推理数据流的处理流程如下。

假设用户的数据可以分为静态数据和动态数据，其中静态数据是从文件系统，例如 HDFS 中所读取的数据，动态数据是从消息队列，例如 Kafka 中所读取的数据，这些数据通常会首先通过由 AlgoOperator 或 Transformer 所代表的特征处理算子进行预处理之后得到一些训练数据，包含训练语义的Estimator 模块在收到这些训练数据之后创建对应的 Model 算子，并且把从训练数据得到的模型数据流输入到这个 Model 算子中。

同样的静态数据和动态数据经过特征处理后得到的推理数据输入到这个 Model 模块，使用 Model 的 Transform 方法进行转换之后，就可以得到预测结果并输出到系统之外。在这幅图里面，每一个方块可以代表一个或者多个算法模块，主要提供计算功能，而每个箭头代表一个或者是多个的 Table 所组成的数据流，而由这些数据流所在的各个模块是可以部署在不同的在生产环境节点之上的。比如对于 Estimator 模块来说，可以先在一些离线的节点中完成它的训练过程，而训练得到的 Model 实例的参数、以及静态的模型数据可以通过 Save 方法来保存在文件系统之中，而实时模型数据流 Table，或者 Data Stream 可以通过 GetModelData 从 Model 中获取，通过 Flink 的 Sink 来写入到比如消息队列之中，然后在前端服务器节点上可以重新创建这个 Model 实例，从文件系统中和消息队列中读取数据，重新部署并提供推理服务。这是分开训练和部署的大致流程。

4. 使用 Pipeline/Graph API 构建复杂机器学习作业

为了更好地将 Flink ML 的子模块组装为更加具备更加高级功能的复杂结构，Flink ML 提供了 Pipeline 和 Graph 这两套 API。这两套 API 它们的共同点在于都是可以用来构建复杂逻辑的机器学习作业，将 Flink ML 的模块进行组合和封装，并且将封装的结果对外表现为一个单独的模块，从而隐藏内部的实现细节。Pipeline 与 Graph 区别在于如果用户想要表示的 Flink ML 作业是多个算子一对一的串联形式的话，那么 Pipeline 能够比较简单地表示这种结构，而 Graph 主要支持通过 DAG 图来表示的方式来表达更加复杂的结构。

02 Flink ML python 模块介绍

Flink ML Python 其实是对 Java API 进行了一层封装，而 Python 用户在使用Python API 时是无感知的，算法在执行时还是使用 Java 去执行的。如下的 Flink ML Python 的架构 ：

03 Flink ML 的生态建设与未来发展方向

1. Flink ML 的开源生态建设

（1）独立的代码库与文档网站

Flink ML 有单独的代码库，不在 Flink 的核心代码库中，这样做是为了避免 Flink 核心代码库过于复杂，同时方便 Flink ML 的快速迭代，另外，Flink ML 也提供了相关的文档网站，主要包括上述提到的基础设施的原理解释、API 的使用，以及 Flink ML 提供的各个算子和配置参数以及使用示例。

（2）基础设施建设

对于希望向 Flink ML 贡献算法的开发者，Flink ML 提供了性能开发工具，例如算子的性能测试工具，这个工具可以在 Flink ML 的代码库中 Flink-ML-Benchmark 模块下找到。目前性能测试工具支持对算子吞吐量测试，未来还会增加对于延迟和算法的准确度等方面的测试，另外 Flink-ML 增加了一个数据算子数据管理的工具，这个工具一方面使用托管内存，方便进行细度的单个算子的内存管理，同时它使用内存和文件系统的共同存储，当数据量需要管理的数据量比较小的时候，数据会被完全存储在内存当中，当数据量超过一定的阈值的时候，多余的数据被溢出到文件系统进行存储，这样一方面是在减少文件 IO 的同时，尽量保证当数据量超过内存托管上限时不会造成 OutOfMemory 的问题。

Flink-ML 提供了一个数据广播工具 With Broadcast，它和普通的 Broadcast 区别在于它除了支读取广播数据到一个算子的所有 Task 之外，它还支持优先读取的方式，也就是说当广播数据被完全读取之前，它不会去尝试消费非广播数据流。这个工具的使用场景包括比如在推理算子之中使用这个工具可以保证推理算子在读取到完整的初始模型之前不会去尝试读取待推理的数据，因为此时它是没有一个模型数据来对它进行推理的。

（3）Flink-Extended 生态项目组织

在 Flink-PMC 的授权下，Flink-ML 建立了一个中立的生态项目组织Flink-Extend，建立这个项目的目的在于，如果开发者有一些项目希望贡献给开源到社区，那么就可以贡献到这个组织下面，目前已经贡献了 Deep Learning On Flink 和 Clink 两个项目。Deep Learning On Flink 可以把 Predict Flow，甚至包括以后 Pytorch 程序打包为 Flink 的 Java 算子，放到 Flink 当中运行。

这样做的优点是可以把 Flink 的预处理程序和 Tensorflow 深度学习算法相结合，形成端到端的训练。Clink 项目提供 C++实现的机器学习算子，并且通过 JNI 的方式为 C++ 的算子提供适配 Flink-ML 的 API。这样做是因为一些用户希望能够部署低延迟、高可用的作业来提供推理服务，而在这些领域方面，C++ 比 JAVA 有明显优势的，并且通过 JNI 的方式也可以使 C++ 算子提供 JAVA API，这样也可以减少开发工作量，使一套实现提供两套接口。

2. Flink ML 的未来发展方向

Flink-ML 未来的发展方向，主要有以下几个方面。

① Flink-ML 会尝试将 Alink 算子适配 Flink-ML 的 API，将 Alink 算子迁移到 Flink-ML 的代码库当中，方便用户获取更多开箱即用的机器学习算子，而在所有能提供的算子之中，会优先丰富特征工程的算法库。

② 在上述提到的性能测试工具的基础之上，对 Flink-ML所有的算法的性能，包括吞吐量、延迟和算法精度等方面进行优化，使其达到生产可用。

③ Flink-ML 还会尝试支持在线算法，在算子中支持输入流批一体的训练或者推理数据流。

分享嘉宾｜

黄兴勃

阿里巴巴 高级开发工程师

Apache Flink Committer

2019 年北京邮电大学硕士毕业，加入阿里巴巴实时计算团队，主要从事 PyFlink 及 Flink Python 生态的开发。

周云峰

阿里巴巴 开发工程师

毕业于美国卡耐基梅隆大学，现在阿里巴巴任开发工程师，目前主要负责Flink ML的开发工作。