flink系列-11、PyFlink 核心功能介绍（整理自 Flink 中文社区）

视频地址

PyFlink 核心功能介绍

• 文章概述：PyFlink 的核心功能原理介绍及相关 demo 演示。
• 作者：程鹤群（军长）（Apache Flink Committer，阿里巴巴技术专家），是 Flink 社区的一名 PMC ,现在在阿里巴巴的实时计算团队。2015年加入阿里巴巴搜索事业部，从事主搜离线相关开发。2017年开始参与 Flink SQL 相关的开发，2019年开始深入参与 PyFlink 相关的开发。
• 整理：谢县东
• 校对：***

课程概要

今天的分享主要包含以下几个部分：

1. PyFlink 的发展史。
2. 介绍 PyFlink 的核心功能以及其背后的一些原理。
3. PyFlink 的 demo 演示。
4. PyFlink 社区扶持计划。

1.PyFlink 的发展史

1.1、v1.8.x

1. Flink 在1.8版本的时候就已经提供 Python API，只在 Datase/Stream 上提供支持。
2. 存在一些问题，比如：
   2.1 Table API 不支持 Python。
   2.2 两套各自独立实现的一个 Python API。
   2.3 底层实现是 JPython，JPython 无法支持 Python3.x。

1.2、v1.9.x

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

1.3、v1.10.x

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

1.4、未来发展

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

2.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 相关的算子，这些算子可以分为两类：

• 1.跟 sql 相关的算子。比如 select、filter、join、window 等；
• 2.在 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。
• 他是很特别的，而是在 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。
• 大体的流程我们可以大概看一下：
  1. 在 open 方法里进行 Java Operator 和 Python Operator 环境的初始化。
  2. 环境初始化好之后，会进行数据处理。当 Java Operator 收到数据之后，先把数据放到一个input buffer 缓冲区中，达到一定的阈值后，才会 flash 到 Python 这边。Python 这边在处理完之后，也会先将数据放到一个结果的缓冲区中，当达到一定阈值，比如达到一定的记录的行数，或者是达到一定的时间的位置，才会把结果 flash 到这边。
  3. state 访问的链路。
  4. logging 访问的链路。
  5. metrics 汇报的链路。

2.Python UDF 的使用

1.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")

下面是一个例子：

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

table_env.register_function("func1", python_udf)

下面是一个例子：

3.Python UDF 的定义方式

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

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

4.依赖管理

我们写完 UDF 的时候，经常遇到一个问题，UDF 里面可能会有一些依赖，那么我们怎么去解决这些依赖问题呢？PyFlink 提供了4种依赖的 API，我们可以一起看一下。

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

table_env.add_python_file(file_path)

2. 依赖存档（打包）文件
   可能会去依赖一个存档的文件，这个时候你可以用 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: 

3. 依赖第三方项目
   这个时候我们可以用 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")

4. 指定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年的5月份进行发布。
接下来会从功能跟性能的两个角度来介绍一下 Pandas UDF。

1.Pandas UDF – 功能

我们先来看一下功能方面，如果你要编写一个 Pandas UDF，那么跟刚才定义普通 UDF 的形式基本上是一致的。你这里只需要去声明一个 udf_type，指定他是 Pandas 就行了。当你指定之后， UDF 运行起来的时候系统传给他的 i 跟 j 就变成一个 pandas.Series 的数据结构。这个时候你就可以直接用 series 来进行操作。与此同时这个会有一个好处，就是我们拿到的是一个 pandas 的数据结构，我们就可以去调用 pandas 相关的一些库函数，并且还可以去调用一些数字计算相关的库函数，这样可以极大的扩展我们的功能。不需要去自己去实现一套逻辑。

2.Pandas UDF - 性能

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

1. 减少了调用的开销，因为刚刚说到了 Pandas UDF 传给你的 UDF 是一个 pandas.series，它相当于是将多行的数据统一一次性的传给了你的 UDF。相比较普通的 UDF 比如多行，他每一行都会去调用你 UDF。Pandas UDF 就是多行只需要调用一次，所以这可以减少 Pandas UDF 调用的开销。
2. 可以减少 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 是一个 pocket，那么他底层是一个 arrow 数据格式，这个时候就可以避免掉 Java 这边的序列化和反序列化。这个就是 Pandas UDF 的一个性能提升。

3.User-defined Metrics

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

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

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

3. 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。 git 的目的是让我们的用户能够尽快的去熟悉 PyFlink 所有的功能。具体参考：https://github.com/pyflink/playgrounds

4.PyFlink 社区扶持计划

• 为什么要发起 PyFlink 社区扶持计划？
  用户逐渐变多、有经验用户少
• 社区目标：并肩作战，营造双赢
• 如何参与 PyFlink 计划？
  https://survey.aliyun.com/apps/zhiliao/B5JOoruzY
  初步审核符合条件后我们会在收到问卷的 10 个工作日内与您联系。
• 扶持目标
  面向所有 PyFlink 社区公司客户
• PyFlink 问题支持&共享
  如果你有一些相关的问题或者是其他的一些意见，可以发到社区的邮件列表里面去。