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

1.準(zhǔn)備工作

2.搭建 PyFlink 環(huán)境

3.總結(jié)與未來

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

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

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

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

接下來看看如何來實(shí)現(xiàn)。

一、準(zhǔn)備工作

Step 1.

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

Step 2.

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

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

Step 3.

安裝以下軟件 (這些軟件是用于創(chuàng)建 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 環(huán)境

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

Step 1. 制作 JobManager 上的 PyFlink Conda 環(huán)境

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

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

%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 

運(yùn)行下面的代碼打包 PyFlink 的 Conda 環(huán)境并且上傳到 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 環(huán)境

運(yùn)行下面的代碼來創(chuàng)建 TaskManager 上的 PyFlink Conda 環(huán)境，TaskManager 上的 PyFlink 環(huán)境至少包含以下 2 個(gè)包：

某個(gè)版本的 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 

運(yùn)行下面的代碼打包 PyFlink 的 conda 環(huán)境并且上傳到 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 環(huán)境

接下來就可以在 Zeppelin 中使用上面創(chuàng)建的 Conda 環(huán)境了，首先需要在 Zeppelin 里配置 Flink，主要配置的選項(xiàng)有：

flink.execution.mode 為 yarn-application, 本文所講的方法只適用于 yarn-application 模式；
指定 yarn.ship-archives，zeppelin.pyflink.Python 以及 zeppelin.interpreter.conda.env.name 來配置 JobManager 側(cè)的 PyFlink Conda 環(huán)境；
指定 Python.archives 以及 Python.executable 來指定 TaskManager 側(cè)的 PyFlink Conda 環(huán)境；
指定其他可選的 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 環(huán)境了。有 2 種場景:

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

三、總結(jié)與未來

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

每個(gè) PyFlink 的環(huán)境都是隔離的，而且可以隨時(shí)定制更改 Conda 環(huán)境。可以下載下面這個(gè) note 并導(dǎo)入到 Zeppelin，就可以復(fù)現(xiàn)今天講的內(nèi)容：http://23.254.161.240/#/notebook/2G8N1WTTS

此外還有很多可以改進(jìn)的地方：

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