来！PyFlink 作业的多种部署模式

关于 PyFlink 的博客我们曾介绍过 PyFlink 的功能开发，比如，如何使用各种算子(Join/Window/AGG etc.)，如何使用各种 Connector(Kafka, CSV, Socket etc.)，还有一些实际的案例。这些都停留在开发阶段，一旦开发完成，我们就面临激动人心的时刻，那就是将我们精心设计开发的作业进行部署，那么问题来了，你知道怎样部署 PyFlink 的作业吗？

本文将为大家全面介绍部署 PyFlink 作业的各种模式。

组件栈回顾

上面的组件栈除了 PyFlink 是第一次添加上去，其他部分大家应该非常熟悉了。目前 PyFlink 基于 Java 的 Table API 之上，同时在 Runtime 层面有 Python 的算子和执行容器。那么我们聚焦重点，看最底层的 Deploy 部分，上图我们分成了三种部署模式，Local/Cluster/Cloud，其中 Local 模式还有 2 种不同方式，一是 SingleJVM，也即是 MiniCluster, 前面博客里面运行示例所使用的就是 MiniCluster。二是 SingleNode，也就是虽然是集群模式，但是所有角色都在一台机器上。下面我们简单介绍一下上面这几种部署模式的区别：

• Local-SingleJVM 模式：该模式大多是开发测试阶段使用的方式，所有角色TM，JM 等都在同一个 JVM 里面。
• Local-SingleNode 模式：意在所有角色都运行在同一台机器，直白一点就是从运行的架构上看，这种模式虽然是分布式的，但集群节点只有 1 个，该模式大多是测试和 IoT 设备上进行部署使用。
• Cluster 模式：也就是我们经常用于投产的分布式部署方式，上图根据对资源管理的方式不同又分为了多种，如：Standalone 是 Flink 自身进行资源管理，YARN，顾名思义就是利用资源管理框架 Yarn 来负责 Flink运行资源的分配，还有结合 Kubernetes 等等。
• Cloud 模式：该部署模式是结合其他云平台进行部署。

接下来我们看看 PyFlink 的作业可以进行怎样的模式部署？

环境依赖

• JDK 1.8+ (1.8.0_211)
• Maven 3.x (3.2.5)
• Scala 2.11+ (2.12.0)
• Python 3.5+ (3.7.6)
• Git 2.20+ (2.20.1)

源码构建及安装

在 Apache Flink 1.10 发布之后，我们除了源码构建之外，还支持直接利用 pip install 安装 PyFlink。那么现在我们还是以源码构建的方式进行今天的介绍。

下载源码

git clone https://github.com/apache/flink.git

签出 release-1.10 分支(1.10 版本是 PyFlink 的第二个版本）

git fetch origin release-1.10
git checkout -b release-1.10 origin/release-1.10

构建编译

mvn clean package -DskipTests

如果一起顺利，你会最终看到如下信息：

...
...
[INFO] flink-walkthrough-table-scala ...................... SUCCESS [  0.070 s]
[INFO] flink-walkthrough-datastream-java .................. SUCCESS [  0.081 s]
[INFO] flink-walkthrough-datastream-scala ................. SUCCESS [  0.067 s]
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time:  16:22 min
[INFO] Finished at: 2019-12-31T10:37:21+08:00
[INFO] ------------------------------------------------------------------------

构建 PyFlink 发布包

上面我们构建了 Java 的发布包，接下来我们构建 PyFlink 的发布包,如下：

cd flink-Python; Python setup.py sdist

最终输出如下信息，证明是成功的：

copying pyflink/util/exceptions.py -> apache-flink-1.10.dev0/pyflink/util
copying pyflink/util/utils.py -> apache-flink-1.10.dev0/pyflink/util
Writing apache-flink-1.10.dev0/setup.cfg
creating dist
Creating tar archive
removing 'apache-flink-1.10.dev0' (and everything under it)

在 dist 目录的 apache-flink-1.10.dev0.tar.gz 就是我们可以用于 pip install 的 PyFlink 包。

安装 PyFlink

上面我们构建了 PyFlink 的发布包，接下来我们利用 pip 进行安装，检测是否之前已经安装过 PyFlink，如下命令：

pip3 list|grep flink
...
flink                         1.0      
pyflink-demo-connector        0.1

上面信息说明我本机已经安装过 PyFlink，我们要先删除，如下：

pip3 uninstall flink

删除以前的安装之后，我们再安装新的如下：

pip3 install dist/*.tar.gz

...
Successfully built apache-flink
Installing collected packages: apache-flink
Successfully installed apache-flink-1.10.dev0

我们再用 list 命令检查一遍：

pip3 list|grep flink

...
apache-flink                  1.10.dev0
pyflink-demo-connector        0.1

其中 pyflink-demo-connector 是我以前做实验时候的安装，对本篇没有影响。

安装 Apache Beam 依赖

我们需要使用 Python3.5+ 版本，检验一下 Python 版本，如下：

jincheng.sunjc$ Python --version
Python 3.7.6

我本机是 Python 3.7.6，现在我们需要安装 Apache Beam，如下：

python -m pip install apache-beam==2.15.0

...
Installing collected packages: apache-beam
Successfully installed apache-beam-2.15.0

如果顺利的出现上面信息，说明 Apache-beam 已经安装成功。

PyFlink 示例作业

接下来我们开发一个简单的 PyFlink 作业，源码如下：

import logging
import os
import shutil
import sys
import tempfile

from pyflink.table import BatchTableEnvironment, EnvironmentSettings
from pyflink.table.descriptors import FileSystem, OldCsv, Schema
from pyflink.table.types import DataTypes
from pyflink.table.udf import udf


def word_count():
   environment_settings = EnvironmentSettings.new_instance().in_batch_mode().use_blink_planner().build()
   t_env = BatchTableEnvironment.create(environment_settings=environment_settings)

   # register Results table in table environment
   tmp_dir = tempfile.gettempdir()
   result_path = tmp_dir + '/result'
   if os.path.exists(result_path):
       try:
           if os.path.isfile(result_path):
               os.remove(result_path)
           else:
               shutil.rmtree(result_path)
       except OSError as e:
           logging.error("Error removing directory: %s - %s.", e.filename, e.strerror)

   logging.info("Results directory: %s", result_path)

   # we should set the Python verison here if `Python` not point
   t_env.get_config().set_python_executable("python3")

   t_env.connect(FileSystem().path(result_path)) \
       .with_format(OldCsv()
                    .field_delimiter(',')
                    .field("city", DataTypes.STRING())
                    .field("sales_volume", DataTypes.BIGINT())
                    .field("sales", DataTypes.BIGINT())) \
       .with_schema(Schema()
                    .field("city", DataTypes.STRING())
                    .field("sales_volume", DataTypes.BIGINT())
                    .field("sales", DataTypes.BIGINT())) \
       .register_table_sink("Results")

   @udf(input_types=DataTypes.STRING(), result_type=DataTypes.ARRAY(DataTypes.STRING()))
   def split(input_str: str):
       return input_str.split(",")

   @udf(input_types=[DataTypes.ARRAY(DataTypes.STRING()), DataTypes.INT()], result_type=DataTypes.STRING())
   def get(arr, index):
       return arr[index]

   t_env.register_function("split", split)
   t_env.register_function("get", get)

   t_env.get_config().get_configuration().set_string("parallelism.default", "1")

   data = [("iPhone 11,30,5499,Beijing", ),
           ("iPhone 11 Pro,20,8699,Guangzhou", ),
           ("MacBook Pro,10,9999,Beijing", ),
           ("AirPods Pro,50,1999,Beijing", ),
           ("MacBook Pro,10,11499,Shanghai", ),
           ("iPhone 11,30,5999,Shanghai", ),
           ("iPhone 11 Pro,20,9999,Shenzhen", ),
           ("MacBook Pro,10,13899,Hangzhou", ),
           ("iPhone 11,10,6799,Beijing", ),
           ("MacBook Pro,10,18999,Beijing", ),
           ("iPhone 11 Pro,10,11799,Shenzhen", ),
           ("MacBook Pro,10,22199,Shanghai", ),
           ("AirPods Pro,40,1999,Shanghai", )]
   t_env.from_elements(data, ["line"]) \
       .select("split(line) as str_array") \
       .select("get(str_array, 3) as city, "
               "get(str_array, 1).cast(LONG) as count, "
               "get(str_array, 2).cast(LONG) as unit_price") \
       .select("city, count, count * unit_price as total_price") \
       .group_by("city") \
       .select("city, "
               "sum(count) as sales_volume, "
               "sum(total_price) as sales") \
       .insert_into("Results")

   t_env.execute("word_count")


if __name__ == '__main__':
   logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")
   word_count()

接下来我们就介绍如何用不同部署模式运行 PyFlink 作业！

Local-SingleJVM 模式部署

该模式多用于开发测试阶段，简单的利用 Python pyflink_job.py 命令，PyFlink 就会默认启动一个 Local-SingleJVM 的 Flink 环境来执行作业，如下：

首先确认你 Python 是 3.5+，然后执行上面的 PyFlink 作业 Python deploy_demo.py,结果写入到本地文件，然后 cat 计算结果，如果出现如图所示的结果，则说明准备工作已经就绪。

这里运行时 SingleJVM，在运行这个 job 时候大家可以查看 java 进程：

我们发现只有一个 JVM 进程，里面包含了所有 Flink 所需角色。

Local-SingleNode 模式部署

这种模式一般用在单机环境中进行部署，如 IoT 设备中，我们从 0 开始进行该模式的部署操作。我们进入到 flink/build-target 目录，执行如下命令(个人爱好，我把端口改成了 8888)：

jincheng:build-target jincheng.sunjc$ bin/start-cluster.sh 
...
Starting cluster.
Starting standalonesession daemon on host jincheng.local.

查看一下 Flink 的进程：

我们发现有 TM 和 JM 两个进程，虽然在一台机器(Local）但是也是一个集群的架构。

上面信息证明已经启动完成，我们可以查看 web 界面：http://localhost:8888/（我个人爱好端口是 8888，默认是 8080）, 如下：

目前集群环境已经准备完成，我们看如果将作业部署到集群中，一条简单的命令，如下：

bin/flink run -m localhost:8888 -py ~/deploy_demo.py

这里如果你不更改端口可以不添加 -m 选项。如果一切顺利，你会得到如下输出：

jincheng:build-target jincheng.sunjc$ bin/flink run -m localhost:8888 -py ~/deploy_demo.py 
Results directory: /var/folders/fp/s5wvp3md31j6v5gjkvqbkhrm0000gp/T/result
Job has been submitted with JobID 3ae7fb8fa0d1867daa8d65fd87ed3bc6
Program execution finished
Job with JobID 3ae7fb8fa0d1867daa8d65fd87ed3bc6 has finished.
Job Runtime: 5389 ms

其中 /var/folders/fp/s5wvp3md31j6v5gjkvqbkhrm0000gp/T/result 目录是计算结果目录，我们可以产看一下，如下：

jincheng:build-target jincheng.sunjc$  cat /var/folders/fp/s5wvp3md31j6v5gjkvqbkhrm0000gp/T/result
Beijing,110,622890
Guangzhou,20,173980
Shanghai,90,596910
Shenzhen,30,317970
Hangzhou,10,138990

同时我们也可以在 WebUI 上面进行查看，在完成的 job 列表中，显示如下：

到此，我们完成了在 Local 模式，其实也是只有一个节点的 Standalone 模式下完成 PyFlink 的部署。

最后我们为了继续下面的操作，请停止集群：

jincheng:build-target jincheng.sunjc$ bin/stop-cluster.sh
Stopping taskexecutor daemon (pid: 45714) on host jincheng.local.
Stopping standalonesession daemon (pid: 45459) on host jincheng.local.

Cluster YARN 模式部署

这个模式部署，我们需要一个 YARN 环境，我们一切从简，以单机部署的方式准备 YARN 环境，然后再与 Flink 进行集成。

准备 YARN 环境

• 安装 Hadoop

我本机是 mac 系统，所以我偷懒一下直接用 brew 进行安装：

jincheng:bin jincheng.sunjc$ brew install Hadoop
Updating Homebrew...
==> Auto-updated Homebrew!
Updated 2 taps (homebrew/core and homebrew/cask).
==> Updated Formulae
Python ✔        doxygen         minio           ntopng          typescript
certbot         libngspice      mitmproxy       ooniprobe
doitlive        minimal-racket  ngspice         openimageio

==> Downloading https://www.apache.org/dyn/closer.cgi?path=hadoop/common/hadoop-
==> Downloading from http://mirror.bit.edu.cn/apache/hadoop/common/hadoop-3.2.1/
######################################################################## 100.0%

🍺  /usr/local/Cellar/Hadoop/3.2.1: 22,397 files, 815.6MB, built in 5 minutes 12 seconds

完成之后，检验一下 Hadoop 版本：

jincheng:bin jincheng.sunjc$ hadoop version
Hadoop 3.2.1

超级顺利，Hadoop 被安装到了 /usr/local/Cellar/hadoop/3.2.1/ 目录下，brew 还是很能提高生产力啊～

• 配置免登(SSH)

Mac 系统自带了 ssh，我们可以简单配置一下即可，我们先打开远程登录。 系统偏好设置 -> 共享 中，左边勾选远程登录，右边选择仅这些用户（选择所有用户更宽松），并添加当前用户。

jincheng:bin jincheng.sunjc$ whoami
jincheng.sunjc

我当前用户是 jincheng.sunjc。配置图如下：

然后生产证书，如下操作：

ssh-keygen -t rsa -P '' -f ~/.ssh/id_rsa
Generating public/private rsa key pair.
/Users/jincheng.sunjc/.ssh/id_rsa already exists.
Overwrite (y/n)? y
Your identification has been saved in /Users/jincheng.sunjc/.ssh/id_rsa.
Your public key has been saved in /Users/jincheng.sunjc/.ssh/id_rsa.pub.
The key fingerprint is:
SHA256:IkjKkOjfMx1fxWlwtQYg8hThph7Xlm9kPutAYFmQR0A jincheng.sunjc@jincheng.local
The key's randomart image is:
+---[RSA 2048]----+
|       ..EB=.o.. |
|..      =.+.+ o .|
|+ .      B.  = o |
|+o .    + o + .  |
|.o. . .+S. * o   |
|  . ..o.= + =    |
|   . + o . . =   |
|      o     o o  |
|            .o   |
+----[SHA256]-----+

接下来将公钥追加到如下文件，并修改文件权限：

jincheng.sunjc$ cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys
jincheng.sunjc$ chmod 0600 ~/.ssh/authorized_keys

利用 ssh localhost 验证，看到 Last login: 字样为 ssh 成功：

jincheng:~ jincheng.sunjc$ ssh localhost
Password:
Last login: Tue Dec 31 18:26:48 2019 from ::1

• 设置环境变量

设置 JAVA_HOME,HADOOP_HOME 和 HADOOP_CONF_DIR，vi ~/.bashrc:

export JAVA_HOME=/Library/Java/JavaVirtualMachines/jdk1.8.0_211.jdk/Contents/Home

export HADOOP_HOME=/usr/local/Cellar/hadoop/3.2.1/libexec

export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoop

NOTE: 后续操作要确保的 terminal 环境变量是生效哦， 如果不生效可以执行 source ~/.bashrc。:)

• 修改配置

1) 修改 core-site.xml

<configuration>
   <property>
      <name>hadoop.tmp.dir</name>
      <value>/tmp</value>
   </property>
   <property>
      <name>fs.defaultFS</name>
      <value>hdfs://localhost:9000</value>
    </property>
</configuration>

2) 修改 hdfs-site.xml

<configuration>

    <property>
        <name>dfs.namenode.name.dir</name>
        <value>/tmp/hadoop/name</value>
    </property>
    
    <property>
        <name>dfs.datanode.data.dir</name>
        <value>/tmp/hadoop/data</value>
    </property>
    
</configuration>

**3) 修改 yarn-site.xml
**

配置 YARN 作为资源管理框架：

<configuration>
    <property>
        <name>yarn.nodemanager.aux-services</name>
        <value>mapreduce_shuffle</value>    
    </property>
    <property>
        <name>yarn.nodemanager.env-whitelist</name>  <value>JAVA_HOME,HADOOP_COMMON_HOME,HADOOP_HDFS_HOME,HADOOP_CONF_DIR,CLASSPATH_PREPEND_DISTCACHE,HADOOP_YARN_HOME,HADOOP_MAPRED_HOME</value>
    </property>
</configuration>

简单的配置已经完成，我们执行一下简单命令启动环境：

• 格式化文档系统

jincheng:libexec jincheng.sunjc$ hadoop namenode -format
...
...
2019-12-31 18:58:53,260 INFO namenode.NameNode: SHUTDOWN_MSG: 
/************************************************************
SHUTDOWN_MSG: Shutting down NameNode at jincheng.local/127.0.0.1
************************************************************/

• 启动服务

我们先启动 hdf 再启动 yarn，如下图：

Okay,一切顺利的话，我们会启动 namenodes，datanodes，resourcemanager 和 nodemanagers。我们有几个 web 界面可以查看，如下：

1)Overview 界面， http://localhost:9870 如下：

2)NodeManager 界面， http://localhost:8042，如下：

3)ResourceManager 管理界面 http://localhost:8088/,如下：

目前 YARN 的环境已经准备完成，我们接下来看如何与 Flink 进行集成。

Flink 集成 Hadoop 包

切换到编译结果目录下 flink/build-target,并将 Haddop 的 JAR 包放到 lib 目录。

在官网下载 Hadoop 包：

cd lib;
curl https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.8.3-7.0/flink-shaded-hadoop-2-uber-2.8.3-7.0.jar > flink-shaded-hadoop-2-uber-2.8.3-7.0.jar

下载后，lib 目录下文件如下：

到现在为止我们可以提交 PyFlink 的作业到由 YARN 进行资源分配的集群了。但为了确保集群上有正确的 Python 环境我们最好打包一个 Python 环境到集群上面。因为大部分情况下我们无法得知 YARN 集群上的 Python 版本是否符合我们的要求(Python 3.5+，装有 apache-beam 2.15.0)，因此我们需要打包一个符合条件的 Python 环境，并随 job 文件提交到 YARN 集群上。

打包 Python 环境

再次检查一下当前 Python 的版本是否 3.5+，如下：

jincheng:lib jincheng.sunjc$ Python
Python 3.7.6 (default, Dec 31 2019, 09:48:30)

由于这个 Python 环境是用于集群的，所以打包时的系统需要和集群一致。如果不一致，比如集群是 linux 而本机是 mac，我们需要在虚拟机或者 docker 中打包。以下列出两种情况的示范方法，读者根据需求选择一种即可。

1）本地打包（集群和本机操作系统一致时）

如果集群所在机器的操作系统和本地一致（都是 mac 或者都是 linux），直接通过 virtualenv 打包一个符合条件的 Python 环境：

• 安装 virtualenv

使用 python -m pip install virtualenv 进行安装如下：

jincheng:tmp jincheng.sunjc$ python -m pip install virtualenv
Collecting virtualenv
  Downloading https://files.Pythonhosted.org/packages/05/f1/2e07e8ca50e047b9cc9ad56cf4291f4e041fa73207d000a095fe478abf84/virtualenv-16.7.9-py2.py3-none-any.whl (3.4MB)
     |████████████████████████████████| 3.4MB 2.0MB/s 
Installing collected packages: virtualenv
Successfully installed virtualenv-16.7.9

我本地环境已经成功安装。

• 创建 Python 环境

用 virtualenv 以 always-copy 方式建立一个全新的 Python 环境，名字随意，以 venv 为例，virtualenv --always-copy venv:

jincheng:tmp jincheng.sunjc$ virtualenv --always-copy venv
Using base prefix '/usr/local/Cellar/Python/3.7.6/Frameworks/Python.framework/Versions/3.7'
New Python executable in /Users/jincheng.sunjc/temp/hadoop/tmp/venv/bin/Python3.7
Also creating executable in /Users/jincheng.sunjc/temp/hadoop/tmp/venv/bin/Python
Installing setuptools, pip, wheel...
done.

• 在新环境中安装 apache-beam 2.15.0

使用 venv/bin/pip install apache-beam==2.15.0 进行安装：

jincheng:tmp jincheng.sunjc$ venv/bin/pip install apache-beam==2.15.0
Collecting apache-beam==2.15.0
...
...
Successfully installed apache-beam-2.15.0 avro-Python3-1.9.1 certifi-2019.11.28 chardet-3.0.4 crcmod-1.7 dill-0.2.9 docopt-0.6.2 fastavro-0.21.24 future-0.18.2 grpcio-1.26.0 hdfs-2.5.8 httplib2-0.12.0 idna-2.8 mock-2.0.0 numpy-1.18.0 oauth2client-3.0.0 pbr-5.4.4 protobuf-3.11.2 pyarrow-0.14.1 pyasn1-0.4.8 pyasn1-modules-0.2.7 pydot-1.4.1 pymongo-3.10.0 pyparsing-2.4.6 pytz-2019.3 pyyaml-3.13 requests-2.22.0 rsa-4.0 six-1.13.0 urllib3-1.25.7

上面信息已经说明我们成功的在 Python 环境中安装了 apache-beam==2.15.0。接下来我们打包 Python 环境。

• 打包 Python 环境

我们将 Python 打包成 zip 文件，zip -r venv.zip venv 如下：

zip -r venv.zip venv
...
...
  adding: venv/lib/Python3.7/re.py (deflated 68%)
  adding: venv/lib/Python3.7/struct.py (deflated 46%)
  adding: venv/lib/Python3.7/sre_parse.py (deflated 80%)
  adding: venv/lib/Python3.7/abc.py (deflated 72%)
  adding: venv/lib/Python3.7/_bootlocale.py (deflated 63%)

查看一下 zip 大小：

jincheng:tmp jincheng.sunjc$ du -sh venv.zip 
 81M  venv.zip

这个大小实在太大了，核心问题是 Beam 的包非常大，后面我会持续在 Beam 社区提出优化建议。我们先忍一下:(。

2）Docker 中打包（比如集群为 linux，本机为 mac 时）

我们选择在 docker 中打包，可以从以下链接下载最新版 docker 并安装：
https://download.docker.com/mac/stable/Docker.dmg 安装完毕后重启终端，执行 docker version 确认 docker 安装成功：

jincheng:tmp jincheng.sunjc$ docker version
Client: Docker Engine - Community
 Version:           19.03.4
 API version:       1.40
 Go version:        go1.12.10
 Git commit:        9013bf5
 Built:             Thu Oct 17 23:44:48 2019
 OS/Arch:           darwin/amd64
 Experimental:      false

Server: Docker Engine - Community
 Engine:
  Version:          19.03.4
  API version:      1.40 (minimum version 1.12)
  Go version:       go1.12.10
  Git commit:       9013bf5
  Built:            Thu Oct 17 23:50:38 2019
  OS/Arch:          linux/amd64
  Experimental:     false
 containerd:
  Version:          v1.2.10
  GitCommit:        b34a5c8af56e510852c35414db4c1f4fa6172339
 runc:
  Version:          1.0.0-rc8+dev
  GitCommit:        3e425f80a8c931f88e6d94a8c831b9d5aa481657
 docker-init:
  Version:          0.18.0
  GitCommit:        fec3683

• 启动容器

我们启动一个 Python 3.7 版本的容器如果是第一次启动可能需要较长时间来拉取镜像：docker run -it Python:3.7 /bin/bash, 如下：

jincheng:libexec jincheng.sunjc$  docker run -it Python:3.7 /bin/bash
Unable to find image 'Python:3.7' locally
3.7: Pulling from library/Python
8f0fdd3eaac0: Pull complete 
d918eaefd9de: Pull complete 
43bf3e3107f5: Pull complete 
27622921edb2: Pull complete 
dcfa0aa1ae2c: Pull complete 
bf6840af9e70: Pull complete 
167665d59281: Pull complete 
ffc544588c7f: Pull complete 
4ebe99df65fe: Pull complete 
Digest: sha256:40d615d7617f0f3b54614fd228d41a891949b988ae2b452c0aaac5bee924888d
Status: Downloaded newer image for Python:3.7

• 容器中安装 virtualenv

我们在刚才启动的容器中安装 virtualenv， pip install virtualenv,如下：

root@1b48d2b526ae:/# pip install virtualenv
Collecting virtualenv
  Downloading https://files.Pythonhosted.org/packages/05/f1/2e07e8ca50e047b9cc9ad56cf4291f4e041fa73207d000a095fe478abf84/virtualenv-16.7.9-py2.py3-none-any.whl (3.4MB)
     |████████████████████████████████| 3.4MB 2.0MB/s 
Installing collected packages: virtualenv
Successfully installed virtualenv-16.7.9
root@1b48d2b526ae:/#

• 创建 Python 环境

以 always copy 方式建立一个全新的 Python 环境，名字随意，以 venv 为例，virtualenv --always-copy venv, 如下：

root@1b48d2b526ae:/# virtualenv --always-copy venv
Using base prefix '/usr/local'
New Python executable in /venv/bin/Python
Installing setuptools, pip, wheel...
done.
root@1b48d2b526ae:/#

• 安装 Apache Beam

在新的 Python 环境中安装 apache-beam 2.15.0，venv/bin/pip install apache-beam==2.15.0,如下：

root@1b48d2b526ae:/# venv/bin/pip install apache-beam==2.15.0
Collecting apache-beam==2.15.0
...
...
Successfully installed apache-beam-2.15.0 avro-Python3-1.9.1 certifi-2019.11.28 chardet-3.0.4 crcmod-1.7 dill-0.2.9 docopt-0.6.2 fastavro-0.21.24 future-0.18.2 grpcio-1.26.0 hdfs-2.5.8 httplib2-0.12.0 idna-2.8 mock-2.0.0 numpy-1.18.0 oauth2client-3.0.0 pbr-5.4.4 protobuf-3.11.2 pyarrow-0.14.1 pyasn1-0.4.8 pyasn1-modules-0.2.7 pydot-1.4.1 pymongo-3.10.0 pyparsing-2.4.6 pytz-2019.3 pyyaml-3.13 requests-2.22.0 rsa-4.0 six-1.13.0 urllib3-1.25.7

• 查看 docker 中的 Python 环境

用 exit 命令退出容器，用 docker ps - a 找到 docker 容器的 id，用于拷贝文件,如下：

root@1b48d2b526ae:/# exit
exit
jincheng:libexec jincheng.sunjc$ docker ps -a
CONTAINER ID        IMAGE               COMMAND             CREATED             STATUS                     PORTS               NAMES
1b48d2b526ae        Python:3.7          "/bin/bash"         7 minutes ago       Exited (0) 8 seconds ago                       elated_visvesvaraya

由于刚刚结束，一般来说是列表中的第一条，可以根据容器的镜像名 Python:3.7 来分辨。我们记下最左边的容器 ID。如上是 1b48d2b526ae。

• 打包 Python 环境

从将容器中的 Python 环境拷贝出来，我们切换到 flink/build-target 目录下，拷贝 docker cp 1b48d2b526ae:/venv ./ 并打包 zip -r venv.zip venv。

最终 flink/build-target 录下生成 venv.zip。

部署作业

终于到部署作业的环节了:), Flink on YARN 支持两种模式，per-job 和 session。per-job 模式在提交 job 时会为每个 job 单独起一个 Flink 集群，session 模式先在 Yarn 上起一个 Flink 集群，之后提交 job 都提交到这个 Flink 集群。

• Pre-Job 模式部署作业

执行以下命令，以 Pre-Job 模式部署 PyFlink 作业：
bin/flink run -m yarn-cluster -pyarch venv.zip -pyexec venv.zip/venv/bin/Python -py deploy_demo.py,如下：

jincheng:build-target jincheng.sunjc$ bin/flink run -m yarn-cluster -pyarch venv.zip -pyexec venv.zip/venv/bin/Python -py deploy_demo.py
2020-01-02 13:04:52,889 WARN  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - The configuration directory ('/Users/jincheng.sunjc/blog/demo_dev/flink/flink-dist/target/flink-1.10-SNAPSHOT-bin/flink-1.10-SNAPSHOT/conf') already contains a LOG4J config file.If you want to use logback, then please delete or rename the log configuration file.
2020-01-02 13:04:52,889 WARN  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - The configuration directory ('/Users/jincheng.sunjc/blog/demo_dev/flink/flink-dist/target/flink-1.10-SNAPSHOT-bin/flink-1.10-SNAPSHOT/conf') already contains a LOG4J config file.If you want to use logback, then please delete or rename the log configuration file.
Results directory: /var/folders/fp/s5wvp3md31j6v5gjkvqbkhrm0000gp/T/result
2020-01-02 13:04:55,945 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at /0.0.0.0:8032
2020-01-02 13:04:56,049 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2020-01-02 13:05:01,153 WARN  org.apache.flink.yarn.YarnClusterDescriptor                   - Neither the HADOOP_CONF_DIR nor the YARN_CONF_DIR environment variable is set. The Flink YARN Client needs one of these to be set to properly load the Hadoop configuration for accessing YARN.
2020-01-02 13:05:01,177 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=1, slotsPerTaskManager=1}
2020-01-02 13:05:01,294 WARN  org.apache.flink.yarn.YarnClusterDescriptor                   - The file system scheme is 'file'. This indicates that the specified Hadoop configuration path is wrong and the system is using the default Hadoop configuration values.The Flink YARN client needs to store its files in a distributed file system
2020-01-02 13:05:02,600 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - Submitting application master application_1577936885434_0004
2020-01-02 13:05:02,971 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl         - Submitted application application_1577936885434_0004
2020-01-02 13:05:02,972 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - Waiting for the cluster to be allocated
2020-01-02 13:05:02,975 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - Deploying cluster, current state ACCEPTED
2020-01-02 13:05:23,138 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - YARN application has been deployed successfully.
2020-01-02 13:05:23,140 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - Found Web Interface localhost:61616 of application 'application_1577936885434_0004'.
Job has been submitted with JobID a41d82194a500809fd715da8f29894a0
Program execution finished
Job with JobID a41d82194a500809fd715da8f29894a0 has finished.
Job Runtime: 35576 ms

上面信息已经显示运行完成，在 Web 界面可以看到作业状态：

我们再检验一下计算结果:

cat/var/folders/fp/s5wvp3md31j6v5gjkvqbkhrm0000gp/T/result：

到这里，我们以 Pre-Job 的方式成功部署了 PyFlink 的作业！相比提交到本地 Standalone 集群，多了三个参数，我们简单说明如下：

参数	说明
-m yarn-cluster	以 Per-Job 模式部署到 yarn 集群
-pyarch venv.zip	将当前目录下的 venv.zip 上传到 yarn 集群
-pyexec venv.zip/venv/bin/Python	指定 venv.zip 中的 Python 解释器来执行 Python UDF，路径需要和 zip 包内部结构一致。

• Session 模式部署作业

以 Session 模式部署作业也非常简单，我们实际操作一下：

jincheng:build-target jincheng.sunjc$ bin/yarn-session.sh 
2020-01-02 13:58:53,049 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.address, localhost
2020-01-02 13:58:53,050 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.port, 6123
2020-01-02 13:58:53,050 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.heap.size, 1024m
2020-01-02 13:58:53,050 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.memory.process.size, 1024m
2020-01-02 13:58:53,050 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.numberOfTaskSlots, 1
2020-01-02 13:58:53,050 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: parallelism.default, 1
2020-01-02 13:58:53,051 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.execution.failover-strategy, region
2020-01-02 13:58:53,413 WARN  org.apache.hadoop.util.NativeCodeLoader                       - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2020-01-02 13:58:53,476 INFO  org.apache.flink.runtime.security.modules.HadoopModule        - Hadoop user set to jincheng.sunjc (auth:SIMPLE)
2020-01-02 13:58:53,509 INFO  org.apache.flink.runtime.security.modules.JaasModule          - Jaas file will be created as /var/folders/fp/s5wvp3md31j6v5gjkvqbkhrm0000gp/T/jaas-3848984206030141476.conf.
2020-01-02 13:58:53,521 WARN  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - The configuration directory ('/Users/jincheng.sunjc/blog/demo_dev/flink/flink-dist/target/flink-1.10-SNAPSHOT-bin/flink-1.10-SNAPSHOT/conf') already contains a LOG4J config file.If you want to use logback, then please delete or rename the log configuration file.
2020-01-02 13:58:53,562 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at /0.0.0.0:8032
2020-01-02 13:58:58,803 WARN  org.apache.flink.yarn.YarnClusterDescriptor                   - Neither the HADOOP_CONF_DIR nor the YARN_CONF_DIR environment variable is set. The Flink YARN Client needs one of these to be set to properly load the Hadoop configuration for accessing YARN.
2020-01-02 13:58:58,824 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=1, slotsPerTaskManager=1}
2020-01-02 13:59:03,975 WARN  org.apache.flink.yarn.YarnClusterDescriptor                   - The file system scheme is 'file'. This indicates that the specified Hadoop configuration path is wrong and the system is using the default Hadoop configuration values.The Flink YARN client needs to store its files in a distributed file system
2020-01-02 13:59:04,779 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - Submitting application master application_1577936885434_0005
2020-01-02 13:59:04,799 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl         - Submitted application application_1577936885434_0005
2020-01-02 13:59:04,799 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - Waiting for the cluster to be allocated
2020-01-02 13:59:04,801 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - Deploying cluster, current state ACCEPTED
2020-01-02 13:59:24,711 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - YARN application has been deployed successfully.
2020-01-02 13:59:24,713 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - Found Web Interface localhost:62247 of application 'application_1577936885434_0005'.
JobManager Web Interface: http://localhost:62247

执行成功后不会返回，但会启动一个 JoBManager Web，地址如上http://localhost:62247，可复制到浏览器查看:

我们可以修改 conf/flink-conf.yaml 中的配置参数。如果要更改某些内容，请参考官方文档。接下来我们提交作业，首先按组合键 Ctrl+Z 将 yarn-session.sh 进程切换到后台，并执行 bg 指令让其在后台继续执行, 然后执行以下命令，即可向 Session 模式的 Flink 集群提交 job bin/flink run -m yarn-cluster -pyarch venv.zip -pyexec venv.zip/venv/bin/Python -py deploy_demo.py：

jincheng:build-target jincheng.sunjc$ bin/flink run -pyarch venv.zip -pyexec venv.zip/venv/bin/Python -py deploy_demo.py

2020-01-02 14:10:48,285 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - Found Web Interface localhost:62247 of application 'application_1577936885434_0005'.
Job has been submitted with JobID bea33b7aa07c0f62153ab5f6e134b6bf
Program execution finished
Job with JobID bea33b7aa07c0f62153ab5f6e134b6bf has finished.
Job Runtime: 34405 ms

如果在打印 finished 之前查看之前的 web 页面，我们会发现 Session 集群会有一个正确运行的作业，如下：

如果已经运行完成，那么我们应该会看到状态也变成结束：

相比 per job 模式提交，少了”-m”参数。因为之前已经启动了 yarn-session.sh，所以 Flink 默认会向 yarn-session.sh 启动的集群上提交 job。执行完毕后，别忘了关闭 yarn-session.sh（session 模式）：先将 yarn-session.sh 调到前台，执行 fg,然后在再按 Ctrl+C 结束进程或者执行 stop，结束时 YARN 上的集群也会被关闭。

Docker 模式部署

我们还可以将 Flink Python job 打包成 docker 镜像，然后使用 docker-compose 或者 Kubernetes 部署执行，由于现在的 docker 镜像打包工具并没有完美支持运行 Python UDF，因此我们需要往里面添加一些额外的文件。首先是一个仅包含PythonDriver 类的 jar 包. 我们在 build-target 目录下执行如下命令：

jincheng:build-target jincheng.sunjc$ mkdir temp
jincheng:build-target jincheng.sunjc$ cd temp
jincheng:temp jincheng.sunjc$ unzip ../opt/flink-Python_2.11-1.10-SNAPSHOT.jar org/apache/flink/client/Python/PythonDriver.class
Archive:  ../opt/flink-Python_2.11-1.10-SNAPSHOT.jar
  inflating: org/apache/flink/client/Python/PythonDriver.class

解压之后，我们再进行压缩打包：

jincheng:temp jincheng.sunjc$ zip Python-driver.jar org/apache/flink/client/Python/PythonDriver.class
  adding: org/apache/flink/client/Python/PythonDriver.class (deflated 56%)

我们得到 Python-driver.jar。然后下载一个 pyArrow 的安装文件(我准备了一个大家下载直接使用即可 pyarrow-0.12.0a0-cp36-cp36m-linux_x86_64.whl。执行以下命令构建 Docker 镜像，需要作为 artifacts 引入的文件有作业文件，Python-driver 的 jar 包和 pyarrow 安装文件,./build.sh --job-artifacts ~/deploy_demo.py,Python-driver.jar,pyarrow-0.12.0a0-cp36-cp36m-linux_x86_64.whl --with-Python3 --from-local-dist(进入 flink/flink-container/docker 目录)。

jincheng:docker jincheng.sunjc$ ./build.sh --job-artifacts ~/deploy_demo.py,Python-driver.jar,pyarrow-0.12.0a0-cp36-cp36m-linux_x86_64.whl --with-Python3 --from-local-dist
Using flink dist: ../../flink-dist/target/flink-*-bin
a .
a ./flink-1.10-SNAPSHOT
a ./flink-1.10-SNAPSHOT/temp
...
...
Removing intermediate container a0558bbcbdd1
 ---> 00ecda6117b7
Successfully built 00ecda6117b7
Successfully tagged flink-job:latest

构建 Docker 镜像需要较长时间，请耐心等待。构建完毕之后，可以输入 docker images 命令在镜像列表中找到构建结果 docker images：

然后我们在构建好的镜像基础上安装好 Python udf 所需依赖，并删除过程中产生的临时文件：

• 启动 docker 容器

1. run -it --user root --entrypoint /bin/bash --name flink-job-container flink-job

• 安装一些依赖

1. add --no-cache g++ Python3-dev musl-dev

• 安装 PyArrow

1. -m pip3 install /opt/artifacts/pyarrow-0.12.0a0-cp36-cp36m-linux_x86_64.whl

• 安装 Apache Beam

1. -m pip3 install apache-beam==2.15.0

• 删除临时文件

1. -rf /root/.cache/pip

执行完如上命令我可以执行 exit 退出容器了，然后把这个容器提交为新的 flink-job 镜像 docker commit -c 'CMD ["--help"]' -c "USER flink" -c 'ENTRYPOINT ["/docker-entrypoint.sh"]' flink-job-container flink-job:latest：

jincheng:docker jincheng.sunjc$ docker commit -c 'CMD ["--help"]' -c "USER flink" -c 'ENTRYPOINT ["/docker-entrypoint.sh"]' flink-job-container flink-job:latest 
sha256:0740a635e2b0342ddf776f33692df263ebf0437d6373f156821f4dd044ad648b

到这里包含 Python UDF 作业的 Docker 镜像就制作好了，这个 Docker 镜像既可以以 docker-compose 使用，也可以结合 Kubernetes 中使用。

我们以使用 docker-compose 执行为例，mac 版 docker 自带 docker-compose，用户可以直接使用，在 flink/flink-container/docker 目录下，使用以下命令启动作业,FLINK_JOB=org.apache.flink.client.Python.PythonDriver FLINK_JOB_ARGUMENTS="-py /opt/artifacts/deploy_demo.py" docker-compose up:

jincheng:docker jincheng.sunjc$ FLINK_JOB=org.apache.flink.client.Python.PythonDriver FLINK_JOB_ARGUMENTS="-py /opt/artifacts/deploy_demo.py" docker-compose up
WARNING: The SAVEPOINT_OPTIONS variable is not set. Defaulting to a blank string.
Recreating docker_job-cluster_1 ... done
Starting docker_taskmanager_1   ... done
Attaching to docker_taskmanager_1, docker_job-cluster_1
taskmanager_1  | Starting the task-manager
job-cluster_1  | Starting the job-cluster
...
...
job-cluster_1  | 2020-01-02 08:35:03,796 INFO  org.apache.flink.runtime.entrypoint.ClusterEntrypoint         - Terminating cluster entrypoint process StandaloneJobClusterEntryPoint with exit code 0.
docker_job-cluster_1 exited with code 0

在 log 中出现“docker_job-cluster_1 exited with code 0”表示 job 已执行成功，JobManager 已经退出。TaskManager 还需要较长的时间等待超时后才会退出，我们可以直接按快捷键 Ctrl+C 提前退出。

查看执行结果，可以从 TaskManager 的容器中将结果文件拷贝出来查看,执行 docker cp docker_taskmanager_1:/tmp/result ./; cat result。

Okay, 到这里本篇要与大家分享的内容已经接近尾声了，如果你期间也很顺利的成功了，可以 Cheers 了:)

小结

本篇核心向大家分享了如何以多种方式部署 PyFlink 作业。期望在 PyFlink 1.10 发布之后，大家能有一个顺利快速体验的快感！作者博客原文在开篇说道部分，为大家分享了老子倡导大家的 “致虚极，守静笃。万物并作，吾以观其复”的大道，同时也给大家带来了 2020 的祝福，祝福大家 “2020 安！”点击「阅读原文」即可查看～

作者介绍：

孙金城（金竹），Committer & PMC Member at @Apache Flink，And Staff Engineer at @Alibaba。自 2015 年以来一直专注于大数据计算领域，并持续贡献于Apache Flink 社区。2011 年加入阿里巴巴集团，目前就职于阿里巴巴计算平台事业部，负责 Apache Flink Python API(PyFlink) 的整体架构开发工作。