本文由 Apache Flink PMC、阿里巴巴技术专家程鹤群分享，主要介绍 PyFlink 的核心功能及应用，分为以下三部分：

PyFlink 的发展史。

PyFlink 的核心功能及原理。

PyFlink Demo 演示。

更多内容，查看 Apache Flink运维和实战系列文章。

1.PyFlink 的发展史

1.1、v1.8.x

Flink 在 1.8 版本的时候就已经提供 Python API，只在 Datase/Stream 上提供支持。
存在一些问题，比如：

Table API 不支持 Python。
两套各自独立实现的一个 Python API。
底层实现是 JPython，JPython 无法支持 Python3.x。

1.2、v1.9.x

2019 年 8 月发布。
支持 Python Table API。

1.3、v1.10.x

2020 年 2 月发布。
提供了 Python UDF 的支持。
提供 UDF 的依赖管理。

1.4、未来发展

提供 Pandas UDF 的支持。
提供用户自定义的一些 UDF Metrics。
ML API。
在易用性方面，提供 SQL DDL 支持 Python UDF。
在后面的一些版本中，我们也希望越来越多的人能够参与到 PyFlink 的贡献和开发中去。

2.PyFlink 核心功能及原理介绍

PyFlink 核心功能将主要从每个版本的划分来跟大家进行介绍，第1个 PyFlink 1.9 版本里面提供 Python Table API 的支持，然后是 PyFlink 1.10 里面提供了 Python UDF 还有相关依赖管理，最后 1.11 版本里面提供了 Pandas UDF 和用户自定义的 Metrics。

2.1、Python Table API (PyFlink 1.9)

■ 1.Python Table API

什么是 Python Table API 呢？我们可以从编程的角度来介绍一下。Python Table API 大概提供了一些 Python 的 API ，比如这里主要可以看一下 Table 的接口， Table 接口上有很多 Table 相关的算子，这些算子可以分为两类：

跟 SQL 相关的算子。比如 select、filter、join、window 等；
在 SQL 的基础上扩展的一些算子。比如 drop_columns(…)，可以用来提升 SQL 的便利性，比如：

当有一个很大的表并且想删除某一列的时候，可以用 drop_columns 来删除某一列。

对于我们来说，可以随意组合 Table 上的方法，然后编写不同的业务逻辑。我们接下来看一下，如何用 Table API 来写一个 WordCount 的例子，可以让大家有一个比较完整的认识。

■ 2.WordCount

如下图所示，是一个完整的 Python Table API 的 WordCount 的例子。主要可以包含4个部分。

首先，我们需要去初始化环境，比如第 6 行，我们先拿到了一个 ExecutionEnvironment，然后第 7 行，去创建一个 TableEnvironment。
创建 TableEnvironment 之后，需要去定义 source 跟 sink ，这里 source 跟 sink 都是指定了输入和输出的文件路径，还指定了文件中 Table 对应的一些字段，以及字段对应的数据类型。而且可以定义输出分隔符。
定义好 source 跟 sink 之后，再来看一下如何编写计算逻辑。可以用 from_path 算子来读取 source 表，读取完之后，就可以进行 group by 的一些聚合，做 group by 跟 wordcount。
做完之后，可以把结果表用 insert_into 进行输出。最后调用 Environment 的 execute 来提交作业。

经过上面4步，我们就完整的写出了一个 Python Table API 的 WordCount。那么对于 WordCount 例子，它的底层实现逻辑是怎么样的呢？接下来看一下，Python Table API 的一个架构。

■ 3.Table API 架构

通过这个架构图，可以看到，Python Table API 是建立在 Java Table API 的基础上的，我们并没有单独的从上到下实现一套 Python Table API。
Python Table API 是一个特别的存在，它是在 Java Table API 的基础上加了一层薄薄的 API，这两层 API 是可以相互调用的。
在 client 端的时候，会起一个 Python VM 然后也会起一个 Java VM ，两个 VM 进行通信。通信的细节可以看下面这张图。

我们可以看到 Python 跟 Java VM 里面都会用 Py4J 各自起一个 Gateway。然后 Gateway 会维护一些对象。
比如我们在 Python 这边创建一个 table 对象的时候，它也会在相应的 Java 这边创建一个相同 table 对象。如果创建一个 TableEnvironment 对象，在 Java 部分也会创建一个 TableEnvironment 对象。
如果你调用 table 对象上的方法，那么也会映射到 Java 这边，所以是一个一一映射的关系。
基于这一套架构，我们可以得出一个结论：如果你用 Python Table API 写出了一个作业，这个作业没有 Python UDF 的时候，那么这个作业的性能跟你用 Java 写出来的作业性能是一样的。因为底层的架构都是同一套 Java 的架构。

刚刚我们介绍了 PyFlink 1.9 版本里面的 Python Table API ，也提到了 table 的接口上面提供了很多不同的算子，而且可以用这些算子去组合，实现不同的业务逻辑。但是对于这些算子来说，它的功能还无法满足一些特定的情况，比如某些业务需要编写一些自定义的逻辑，此时就需要强依赖 Python UDF，所以在 PyFlink 1.10 版本里面，提供了 Python UDF 并且提供了相应的依赖管理。

2.2、Python UDF & 依赖管理 (PyFlink 1.10)

■ 1.Python UDF 架构

如果你的作业是包含一个 Python UDF 的作业，那么从提交的时候，就是左边的架构图，然后 deploy 到 Remote 端的时候，可以看到 Remote 端的架构图分为两个部分。左边部分是 Java 的 Operator，右边部分是 Python 的 Operator。

大体的流程我们可以大概看一下：

在 open 方法里进行 Java Operator 和 Python Operator 环境的初始化。
环境初始化好之后，会进行数据处理。当 Java Operator 收到数据之后，先把数据放到一个 input buffer 缓冲区中，达到一定的阈值后，才会 flash 到 Python 这边。Python 处理完之后，也会先将数据放到一个结果的缓冲区中，当达到一定阈值，比如达到一定的记录的行数，或者是达到一定的时间，才会把结果 flush 到这边。
state 访问的链路。
logging 访问的链路。
metrics 汇报的链路。

■ 2.Python UDF 的使用

PyFlink-1.9 版本中，Python API 中支持注册使用 Java UDF，使用方法如下：可以调 TableEnvironment 上的 register_java_function 这个方法，有两个参数，一个参数是给 UDF 的命名，第 2 个是 Java 类的路径。

table_env.register_java_function("func1", "java.user.defined.function.class.name")

下面是一个例子：

Python UDF 的使用：

可以调 TableEnvironment 上的 register_function 这个方法，有两个参数，一个参数是给 UDF 起的名字，第 2 个是 python_udf 的一个对象。

table_env.register_function("func1", python_udf)

下面是一个例子：

■ 3.Python UDF 的定义方式

PyFlink 里面也支持一些其他的方式去定义 UDF，我们可以看一下，总共有4种方式：

可以继承 ScalaFunction 基类，并重写 eval 方法。
直接定义一个 Named Function，然后再用 UDF 的签名去声明 UDF 的输入类型和输出类型。
也可以用刚刚例子里面的 Lambda Function 的这种方式，来定义 Python UDF。
最后一种是 Callable Function 的方式。也是声明其输入和输出的类型。

■ 4.依赖管理

写完 UDF 的时候，经常遇到一个问题，UDF 里面可能会有一些依赖，如何去解决这些依赖问题呢？PyFlink 提供了 4 种依赖的 API，如下所示。

依赖文件

如果 UDF 里面依赖一个文件的话，可以用 add_python_file 加载依赖的文件的路径，指定完之后，作业提交的时候，就会把这个文件分发到集群，那么在远程执行的时候，你的 UDF 就可以去访问这个文件。

table_env.add_python_file(file_path)

依赖存档（打包）文件

可能会去依赖一个存档的文件，这个时候你可以用 add_python_archive 方法，传入两个参数。第 2 个参数是一个可选的参数。第 1 个参数表示对你存档文件的重命名。如果调用了 API，那么在 UDF 里面就可以去访问存档文件里面的所有文件。

table_env.add_python_archive("py_env.zip", "myenv") 
# the files contained in the archive file can be accessed in UDF 
def my_udf(): 
    with open("myenv/py_env/data/data.txt") as f:

依赖第三方项目

可以用 set_python_requirements 方法去指定你的第三方依赖。也是有两个参数，第 1 个参数是传一个文件，文件中写了所依赖的第三方项目，以及它对应的版本。第 2 个参数是一个可选的参数，如果集群是一个有网络的环境，那么第 2 个参数可以不填，当第 2 个参数不填的时候，作业提交开始初始化的时候， Python 就会去根据你的 requirements 文件里面配置的依赖，自动的去网络下载你的依赖，然后安装。如果集群是没有网络的，可以预先把这些依赖下载好，下载到 cached 的目录里面去。然后把目录也一起提交到集群，集群拿到这个目录会去安装这些依赖。

# commands executed in shell 
echo numpy==1.16.5 > requirements.txt 
pip download -d cached_dir -r requirements.txt --no-binary :all: 
# python code 
table_env.set_python_requirements("requirements.txt", "cached_dir")

指定 Python Interpreter 路径

假设你的 Python UDF 运行的时候，会依赖某一个版本的 Python 解释器。那么这个时候可以去指定你所希望 Python UDF 运行的一个解释器的路径。

table_env.get_config().set_python_executable("py_env.zip/py_env/bin/python")

2.3、Pandas UDF & User-defined Metrics (PyFlink 1.11)

我们在 Pyflink 1.11 的版本里面提供了 Pandas UDF，还有用户自定义的 Metrics。当然 Pyflink 1.11 版本里面，不光是这两个功能，我这里主要是介绍一下这两个功能。Pyflink 1.11 版本也会即将在 2020 年的 6 月份进行发布。

接下来会从功能和性能两个角度来介绍一下 Pandas UDF。

■ 1.Pandas UDF – 功能

我们先来看一下功能方面，如果你要编写一个 Pandas UDF，那么跟刚才定义普通 UDF 的形式基本上是一致的。这里只需要去声明一个 udf_type，指定为 Pandas 就行了。

指定之后， UDF 运行起来的时候系统传的 i 跟 j 就变成一个 pandas.Series 的数据结构。这个时候可以直接用 series 来进行操作。与此同时会有一个好处，就是我们拿到的是一个 pandas 的数据结构，可以调用 pandas 相关的一些库函数，并且可以调用一些数值计算相关的库函数，这样可以极大的扩展功能。不需要再去实现一套逻辑。

■ 2.Pandas UDF - 性能

那么性能上 Pandas UDF 的好处，主要有两点。

减少了调用的开销，因为刚刚说到了系统传给 UDF 的是一个 pandas.series，它相当于是将多行的数据一次性的传给了 UDF。而普通 UDF 处理多行时，每行都需要调用一次 UDF。所以对比可以发现，Pandas UDF 可以显著减少 UDF 的调用开销。
可以减少 UDF 的序列化和反序列化开销。这里具体介绍一下，Pandas UDF 是如何减少了序列化和反序列化的。

我们可以看一下右边这个图，左边是 Java Operator，右边是 Python Operator。假设 Operator 收到了一个 X，然后 X 在这里会进行一个序列化，变成 arrow 的内存数据格式，这个时候用 X’ 来表示。那么这个时候 Java 这边会把 X’ 传给 Python，Python 就可以直接来访问 arrow 数据结构，因为 pandas 底层的数据结构就是用 arrow 来表示的，所以这个时候不需要在 Python 进行反序列化，可以直接来操作 X’。然后在 X’ 加一之后，得到 Y’， Y’ 也是直接生成的 arrow 内存数据格式，这里也不需要反序列化。那么把 Y’ 传到 Java 时，就需要进行一个反序列化。

我们可以发现，只需要在 Java 进行一个序列化和反序列化。Python 这边可以省去了序列化和反序列化开销。

而且这里需要提出的一点是，如果你的输入 X 也是一个 arrow 的内存数据格式，那么 Java 这边的序列化跟反序列化也是可以避免的。比如你的 source 是一个 Parquet Source，那么它输出的数据格式也是 arrow 数据格式，这个时候就可以避免掉 Java 的序列化和反序列化。所以，Pandas UDF 也是可以减少序列化反序列化的开销。

■ 3.User-defined Metrics

我们再来看一下用户自定义 Metrics。

Metric 注册

先来看一下 Metric 的注册，Metric 注册可以是在 metric_group 上调用对应的 Metric 方法来注册。

Metric Scope

metric_group 还可以调用他的 add_group 方法去定义你的 Metric 的一个域，可以对 Metric 进行分类。

Metric 类型

目前 PyFlink 里面提供的 Metric 类型有以下4种：

Counter

类似累加器。一开始需要在 open 方法里面进行 Counter 的注册，然后调用 match_group 上 Counter 方法，这里我们给了一个 Metric 的名字叫 my_counter。定义完之后，就可以在 Eval 方法里面进行使用。然后 Counter 可以提供 Inc 方法，你可以调用 Inc 进行相应的增加。

Gauge

它是用来反映一个瞬时值。假设我们需要在 Metric 上显示 length 值的变化情况。那么我们需要用 Gauge 方法来注册，名字是 my_gauge。第2个参数这里需要注意它是一个 UDF ，我们需要返回要监控数值的值是什么，返回这个值。然后在 Eval 方法里或者其他 UDF 的调用里可以改变这个值。框架底层就会不断去汇报这个值当前值是多少。

Meter

Meter 这种 Metric 是表示当前这一秒往前一个时间区间内所有数值相加的一个均值。我们看可以调用 Meter 方法来注册。第2个参数是一个默认的参数，默认是60秒，表示60秒内所有值的一个均值。这里需要注意的是，Meter 每一秒都会去汇报当前这一秒往前60秒时间区间内，所有值的均值。可以用 Meter 的 mark_event 方法来汇报。

Distribution (sum/count/min/max/mean)

最后一种是 Distribution 的一个 Metric 类型，它对你的值能提供一些 sum/count/min/max/mean 等统计信息。可以调用 metric_group 上的 Distribution 这个方法。更新上可以调用 distribution.update。

3. PyFlink 的 Demo 演示

接下来对这些核心功能做一些 Demo 的演示跟讲解。此处我们提供了一个 playgrounds 的 git。主要是希望帮助大家更快速地熟悉 PyFlink 所有的功能及使用，并附上了相关代码示例。具体参考信息请见下方链接：

https://github.com/pyflink/playgrounds

推荐阅读：

Apache Flink运维和实战系列文章
Apache Flink 零基础入门到进阶系列文章

创作场景

PyFlink 核心功能与应用全解