开发者学堂课程【开源 Flink 极客训练营：PyFlink 快速上手】学习笔记，与课程紧密联系，让用户快速学习知识。
课程地址：https://developer.aliyun.com/learning/course/760/detail/13342
PyFlink 快速上手
三.PyFlink 功能演示
打开该网页页面，提供了 PyFlink 相关的一些 demo,这些 demo运行在 Docker 中，首先需要在本地安装 Docker 环境。环境安装完成后，首先运行命令 docker-compose up -d,该命令会启动一个Flink 的 standalone 集群，之后运行的 Flink 例子都会提交到该集群。输入命令 docker ps 可以查看。
例一. WordCount
首先定义了 blink_planner,同时应用在批模式下，作业并发度是1。在作业中还定义了 source,只有一个 string 类型字段，输入数据来自一个文件。同时也定义了一个 sink,包括两个字段，一个是 word,一个是 count,类型分别是 string 类型和 BIGINT 类型，sink 位置也是在本地一个文件中。作业基本执行逻辑是读取 source 中的数据，按照 group_by 计算 word 的频率，将结果写入 sink。最后Execute 语句会将作业提交执行。运行该作业，可以看到作业运行之后提交到刚刚的 Flink 集群中执行。查看作业执行结果：
flink,2
pyflink,1
例二.Python UDF
运行在 blink_planner,运行在批处理模式下，作业并发度为1。在作业中定义了一个 UDF,该 UDF 输入包括两个列，类型为 BIGINT，输出类型也为 BIGINT。UDF逻辑是将这两个列相加作为一个结果输出。 source 表与 sink 表与上例类似。在执行逻辑中，读取 source表然后运行 UDF，将两列相加再将计算结果写入 sink 表。执行该作业，查看执行结果：
3
例三.带有依赖的 Python UDF
在该 UDF中引用了一个第三方依赖，在作业执行中执行该依赖，使用set_python_requirements。查看 requirements文件，可以看到文件只包含了一行，指定了需要的依赖及版本。运行该作业，在初始化阶段需要下载 Python 相关的依赖，查看执行结果：
3
例四.向量化 Python UDF
在 UDF 定义时加了一个 udf_type 字段，说明是向量化的 Python UDF。其他逻辑与普通 Python UDF 类似。执行作业，查看执行结果：
3
例五.在 Java Table API 中使用 Python UDF
用 Java 语言写，通过一个 DDL 语句的方式注册 Python UDF，然后在 execute Sql 语句中使用注册的 UDF。先编译该 UDF,执行作业，通过 flink run -j 命令指定 jar 包，然后通过 pyfs 参数指定运行 Python UDF 所用到的路径。
例六.在 SQL 中使用 Python UDF
如上图，声明了一个 UDF,名字是 add_one,类型是 Python，位置是 udfs.add_one。运行作业，查看执行结果：
2
3
4
四．PyFlink 下一步规划
1. Python DataStream API
2. Python UDAF 支持/Pandas UDAF 支持
3. Java/Python 通信优化
常用资源链接
1. Python Table API 文档
https://ci.apache.org/projects/flink/flink-docs-master/api/python/
2. PyFlink 文档
https://ci.apache.org/projects/flink/flink-docs-master/dev/table/python/
3. PyFlink playground
https://github.com/pyflink/playgrounds/tree/1.11
总结：首先是PyFlink的目标以及发展历程，PyFlink的一些核心功能，比如Python Table API，Python UDF，向量化Python UDF
，如何在Python UDF中使用Metrics以及Python的依赖管理，Python UDF目前所做的一些优化工作，同时也了解了一些Demo，对于PyFlink的功能有了更直观的了解，最后是PyFlink的下一步规划。
