算法|PyFlink 开发环境利器：Zeppelin Notebook 算法

文章图片
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 ，主要配置的选项有：