PyFlink 作為 Flink 的 Python 語言入口，其 Python 語言的確很簡單易學，但是 PyFlink 的開發環境卻不容易搭建，稍有不慎，PyFlink 環境就會亂掉，而且很難排查原因。今天給大家介紹一款能夠幫你解決這些問題的 PyFlink 開發環境利器：Zeppelin Notebook。主要內容為：

1.準備工作

2.搭建 PyFlink 環境

3.總結與未來

也許你早就聽說過 Zeppelin，但是之前的文章都偏重講述如何在 Zeppelin 里開發 Flink SQL，今天則來介紹下如何在 Zeppelin 里高效的開發 PyFlink Job，特別是解決 PyFlink 的環境問題。

一句來總結這篇文章的主題，就是在 Zeppelin notebook 里利用 Conda 來創建 Python env 自動部署到 Yarn 集群中，你無需手動在集群上去安裝任何 PyFlink 的包，并且你可以在一個 Yarn 集群里同時使用互相隔離的多個版本的 PyFlink。最后你能看到的效果就是這樣：

1. 能夠在 PyFlink 客戶端使用第三方 Python 庫，比如 matplotlib：

2. 可以在 PyFlink UDF 里使用第三方 Python 庫，如：

接下來看看如何來實現。

一、準備工作

Step 1.

準備好最新版本的 Zeppelin 的搭建，這個就不在這邊展開了，如果有問題可以加入 Flink on Zeppelin 釘釘群 (34517043) 咨詢。另外需要注意的是，Zeppelin 部署集群需要是 Linux，如果是 Mac 的話，會導致在 Mac 機器上打的 Conda 環境無法在 Yarn 集群里使用 (因為 Conda 包在不同系統間是不兼容的)。

Step 2.

下載 Flink 1.13， 需要注意的是，本文的功能只能用在 Flink 1.13 以上版本，然后：

把 flink-Python-*.jar 這個 jar 包 copy 到 Flink 的 lib 文件夾下；
把 opt/Python 這個文件夾 copy 到 Flink 的 lib 文件夾下。

Step 3.

安裝以下軟件 (這些軟件是用于創建 Conda env 的)：

miniconda：https://docs.conda.io/en/latest/miniconda.html
conda pack：https://conda.github.io/conda-pack/
mamba：https://github.com/mamba-org/mamba

二、搭建 PyFlink 環境

接下來就可以在 Zeppelin 里搭建并且使用 PyFlink 了。

Step 1. 制作 JobManager 上的 PyFlink Conda 環境

因為 Zeppelin 天生支持 Shell，所以可以在 Zeppelin 里用 Shell 來制作 PyFlink 環境。注意這里的 Python 第三方包是在 PyFlink 客戶端 (JobManager) 需要的包，比如 Matplotlib 這些，并且確保至少安裝了下面這些包：

某個版本的 Python (這里用的是 3.7）
apache-flink (這里用的是 1.13.1)
jupyter，grpcio，protobuf (這三個包是 Zeppelin 需要的)
剩下的包可以根據需要來指定：

%sh# make sure you have conda and momba installed.# install miniconda: https://docs.conda.io/en/latest/miniconda.html# install mamba: https://github.com/mamba-org/mambaecho "name: pyflink_envchannels:  - conda-forge  - defaultsdependencies:  - Python=3.7  - pip  - pip:    - apache-flink==1.13.1  - jupyter  - grpcio  - protobuf  - matplotlib  - pandasql  - pandas  - scipy  - seaborn  - plotnine " > pyflink_env.yml    mamba env remove -n pyflink_envmamba env create -f pyflink_env.yml 

運行下面的代碼打包 PyFlink 的 Conda 環境并且上傳到 HDFS (注意這里打包出來的文件格式是 tar.gz)：

%shrm -rf pyflink_env.tar.gzconda pack --ignore-missing-files -n pyflink_env -o pyflink_env.tar.gzhadoop fs -rmr /tmp/pyflink_env.tar.gzhadoop fs -put pyflink_env.tar.gz /tmp# The Python conda tar should be public accessible, so need to change permission here.hadoop fs -chmod 644 /tmp/pyflink_env.tar.gz 

Step 2. 制作 TaskManager 上的 PyFlink Conda 環境

運行下面的代碼來創建 TaskManager 上的 PyFlink Conda 環境，TaskManager 上的 PyFlink 環境至少包含以下 2 個包：

某個版本的 Python (這里用的是 3.7）
apache-flink (這里用的是 1.13.1)
剩下的包是 Python UDF 需要依賴的包，比如這里指定了 pandas：

echo "name: pyflink_tm_envchannels:  - conda-forge  - defaultsdependencies:  - Python=3.7  - pip  - pip:    - apache-flink==1.13.1  - pandas " > pyflink_tm_env.yml    mamba env remove -n pyflink_tm_envmamba env create -f pyflink_tm_env.yml 

運行下面的代碼打包 PyFlink 的 conda 環境并且上傳到 HDFS (注意這里使用的是 zip 格式）

%shrm -rf pyflink_tm_env.zipconda pack --ignore-missing-files --zip-symlinks -n pyflink_tm_env -o pyflink_tm_env.ziphadoop fs -rmr /tmp/pyflink_tm_env.ziphadoop fs -put pyflink_tm_env.zip /tmp# The Python conda tar should be public accessible, so need to change permission here.hadoop fs -chmod 644 /tmp/pyflink_tm_env.zip 

Step 3. 在 PyFlink 中使用 Conda 環境

接下來就可以在 Zeppelin 中使用上面創建的 Conda 環境了，首先需要在 Zeppelin 里配置 Flink，主要配置的選項有：

flink.execution.mode 為 yarn-application, 本文所講的方法只適用于 yarn-application 模式；
指定 yarn.ship-archives，zeppelin.pyflink.Python 以及 zeppelin.interpreter.conda.env.name 來配置 JobManager 側的 PyFlink Conda 環境；
指定 Python.archives 以及 Python.executable 來指定 TaskManager 側的 PyFlink Conda 環境；
指定其他可選的 Flink 配置，比如這里的 flink.jm.memory 和 flink.tm.memory。

%flink.confflink.execution.mode yarn-applicationyarn.ship-archives /mnt/disk1/jzhang/zeppelin/pyflink_env.tar.gzzeppelin.pyflink.Python pyflink_env.tar.gz/bin/Pythonzeppelin.interpreter.conda.env.name pyflink_env.tar.gzPython.archives hdfs:///tmp/pyflink_tm_env.zipPython.executable  pyflink_tm_env.zip/bin/Python3.7flink.jm.memory 2048flink.tm.memory 2048 

接下來就可以如一開始所說的那樣在 Zeppelin 里使用 PyFlink 以及指定的 Conda 環境了。有 2 種場景:

下面的例子里，可以在 PyFlink 客戶端 (JobManager 側) 使用上面創建的 JobManager 側的 Conda 環境，比如下邊使用了 Matplotlib。
下面的例子是在 PyFlink UDF 里使用上面創建的 TaskManager 側 Conda 環境里的庫，比如下面在 UDF 里使用 Pandas。

三、總結與未來

本文內容就是在 Zeppelin notebook 里利用 Conda 來創建 Python env 自動部署到 Yarn 集群中，無需手動在集群上去安裝任何 Pyflink 的包，并且可以在一個 Yarn 集群里同時使用多個版本的 PyFlink。

每個 PyFlink 的環境都是隔離的，而且可以隨時定制更改 Conda 環境。可以下載下面這個 note 并導入到 Zeppelin，就可以復現今天講的內容：http://23.254.161.240/#/notebook/2G8N1WTTS

此外還有很多可以改進的地方：

目前我們需要創建 2 個 conda env ，原因是 Zeppelin 支持 tar.gz 格式，而 Flink 只支持 zip 格式。等后期兩邊統一之后，只要創建一個 conda env 就可以；
apache-flink 現在包含了 Flink 的 jar 包，這就導致打出來的 conda env 特別大，yarn container 在初始化的時候耗時會比較長，這個需要 Flink 社區提供一個輕量級的 Python 包 (不包含 Flink jar 包)，就可以大大減小 conda env 的大小。