PyFlink 开发环境利器：Zeppelin Notebook

也许你早就听说过 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 以上版本，然后：

Step 3.

安装以下软件 (这些软件是用于创建 Conda env 的)：

二、搭建 PyFlink 环境

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

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

因为 Zeppelin 天生支持 Shell，所以可以在 Zeppelin 里用 Shell 来制作 PyFlink 环境。注意这里的 Python 第三方包是在 PyFlink 客户端 (JobManager) 需要的包，比如 Matplotlib 这些，并且确保至少安装了下面这些包：

剩下的包可以根据需要来指定：

%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/mamba

echo "name: pyflink_env
channels:
  - conda-forge
  - defaults
dependencies:
  - 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_env
mamba env create -f pyflink_env.yml

运行下面的代码打包 PyFlink 的 Conda 环境并且上传到 HDFS (注意这里打包出来的文件格式是 tar.gz)：

%sh

rm -rf pyflink_env.tar.gz
conda pack --ignore-missing-files -n pyflink_env -o pyflink_env.tar.gz

hadoop fs -rmr /tmp/pyflink_env.tar.gz
hadoop 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 UDF 需要依赖的包，比如这里指定了 pandas：

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

运行下面的代码打包 PyFlink 的 conda 环境并且上传到 HDFS (注意这里使用的是 zip 格式）

%sh

rm -rf pyflink_tm_env.zip
conda pack --ignore-missing-files --zip-symlinks -n pyflink_tm_env -o pyflink_tm_env.zip

hadoop fs -rmr /tmp/pyflink_tm_env.zip
hadoop 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.conf


flink.execution.mode yarn-application

yarn.ship-archives /mnt/disk1/jzhang/zeppelin/pyflink_env.tar.gz
zeppelin.pyflink.Python pyflink_env.tar.gz/bin/Python
zeppelin.interpreter.conda.env.name pyflink_env.tar.gz

Python.archives hdfs:///tmp/pyflink_tm_env.zip
Python.executable  pyflink_tm_env.zip/bin/Python3.7

flink.jm.memory 2048
flink.tm.memory 2048

接下来就可以如一开始所说的那样在 Zeppelin 里使用 PyFlink 以及指定的 Conda 环境了。有 2 种场景:

三、总结与未来

本文内容就是在 Zeppelin notebook 里利用 Conda 来创建 Python env 自动部署到 Yarn 集群中，无需手动在集群上去安装任何 Pyflink 的包，并且可以在一个 Yarn 集群里同时使用多个版本的 PyFlink。

每个 PyFlink 的环境都是隔离的，而且可以随时定制更改 Conda 环境。可以下载下面这个 note 并导入到 Zeppelin，就可以复现今天讲的内容：http://23.254.161.240/#/notebook/2G8N1WTTS

此外还有很多可以改进的地方：