如何在zeppelin里高效的开发PyFlinkJob?

本文重点介绍下如何在 Zeppelin 里高效的开发 PyFlink Job&＃xff0c;特别是解决 PyFlink 的环境问题。

大家都知道PyFlink 的开发环境不容易搭建&＃xff0c;稍有不慎&＃xff0c;PyFlink 环境就会乱掉&＃xff0c;而且很难排查原因
本文使用miniconda、conda-pack、mamba分别制作JobManager 上的 PyFlink Conda 环境、TaskManager 上的 PyFlink Conda 环境&＃xff0c;然后在 Zeppelin 里使用 PyFlink 以及指定 Conda 环境&＃xff0c;这样就可以在一个 Yarn 集群里同时使用多个版本的 PyFlink。

本文使用环境&＃xff1a;flink 1.13.2&＃xff0c; zeppelin&＃xff1a;0.10.0-bin-all

需要改进的地方&＃xff1a;

• 需要创建 2 个 conda env &＃xff0c;原因是 Zeppelin 支持 tar.gz 格式&＃xff0c;而 Flink 只支持 zip 格式
• apache-flink 目前包含了 Flink 的 jar 包&＃xff0c;导致打出来的 conda env 特别大&＃xff08;500MB以上&＃xff09;&＃xff0c;yarn container 在初始化的时候耗时会比较长&＃xff0c;这个需要 Flink 社区提供一个轻量级的 Python 包 (不包含 Flink jar 包)&＃xff0c;就可以大大减小 conda env 的大小。

1. 搭建 PyFlink 环境

1.1.制作 JobManager 上的 PyFlink Conda 环境

• 注意dependencies所列的第三方包是在 PyFlink 客户端 (JobManager) 需要的包&＃xff0c;比如 Matplotlib &＃xff0c;并且确保至少安装了所列的这些包&＃xff1a;
• jupyter&＃xff0c;grpcio&＃xff0c;protobuf 是Zeppelin 需要的
• apache-flink 指定flink的版本

echo "name: pyflink_env
channels:- https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main/- https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free/- https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/conda-forge/- https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/msys2/
dependencies:- Python&＃61;3.7- pip- pip:- apache-flink&＃61;&＃61;1.13.2- jupyter- grpcio- protobuf- matplotlib- pandasql- pandas- scipy- seaborn- plotnine" > pyflink_env.ymlmamba env remove -n pyflink_env
mamba env create -f pyflink_env.ymlrm -rf pyflink_env.tar.gz
conda pack --ignore-missing-files -n pyflink_env -o pyflink_env.tar.gzhadoop 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


1.2.制作 TaskManager 上的 PyFlink Conda 环境

echo "name: pyflink_tm_env
channels:- https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main/- https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free/- https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/conda-forge/- https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/msys2/
dependencies:- Python&＃61;3.7- pip- pip:- apache-flink&＃61;&＃61;1.13.2- pandas" > pyflink_tm_env.ymlmamba env remove -n pyflink_tm_env
mamba env create -f pyflink_tm_env.ymlrm -rf pyflink_tm_env.zip
conda pack --ignore-missing-files --zip-symlinks -n pyflink_tm_env -o pyflink_tm_env.ziphadoop 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


1.3.安装本地的flink

下载 flink 1.13并解压&＃xff0c;然后&＃xff1a;

• 把 opt目录下的flink-python-*.jar 这个 jar 包 copy 到flink 的 lib 文件夹下&＃xff1b;
• 把 opt/python 这个文件夹 copy 到flink 的 lib 文件夹下。

1.4. 在 PyFlink 中使用 Conda 环境

在 Zeppelin 里配置 Flink&＃xff0c;主要配置的选项有&＃xff1a;

• flink.execution.mode 为 yarn-application, 本文所讲的方法只适用于 yarn-application 模式&＃xff1b;
• 指定 yarn.ship-archives&＃xff0c;zeppelin.pyflink.Python 以及 zeppelin.interpreter.conda.env.name 来配置 JobManager 侧的 PyFlink Conda 环境&＃xff1b;
• 指定 Python.archives 以及 Python.executable 来指定 TaskManager 侧的 PyFlink Conda 环境&＃xff1b;
• 指定其他可选的 Flink 配置&＃xff0c;比如这里的 flink.jm.memory 和 flink.tm.memory。

FLINK_HOME /data/flink/flink-1.13.2
HADOOP_CONF_DIR /etc/hadoop/conf
HIVE_CONF_DIR /etc/hive/conf
flink.execution.mode yarn-applicationzeppelin.pyflink.python pythonyarn.ship-archives /data/flink/pyflink_env.tar.gz
zeppelin.interpreter.conda.env.name pyflink_env.tar.gz
zeppelin.pyflink.Python pyflink_env.tar.gz/bin/python Python.archives hdfs://172.25.21.170:8020/tmp/pyflink_tm_env.zip
Python.executable pyflink_tm_env.zip/bin/python3.7 flink.jm.memory 2048
flink.tm.memory 2048zeppelin.interpreter.connect.timeout 600000


1.5. 使用示例

%flink.ipyflink
%matplotlib inline
import matplotlib.pyplot as plt
plt.plot([1,2,3,4])
plt.ylabel(&＃39;numbers&＃39;)
plt.show()%flink.ipyflink
import time
class PandasVersion(ScalarFunction):def eval(self,s):import pandas as pdreturn pd.__version__ &＃43; " " &＃43; s
bt_env.register_funtion("pandas_version",udf(PandasVersion(),DataTypes.STRING,DataTypes.STRING)%bsql
select pandas_version(&＃39;hello world&＃39;)


附录

准备conda环境

在centos 7.x上安装miniconda

安装miniconda

miniconda是一个免费的conda最小安装程序。
它是Anaconda的一个小型的引导版本&＃xff0c;只包含conda、Python、它们所依赖的包&＃xff0c;以及少量其他有用的包&＃xff0c;包括pip、zlib和其他一些包

下载并安装

根据实际需要&＃xff0c;下载对应的版本&＃xff0c;latest表示最新的python版本&＃xff0c;是python3.9.5

• Miniconda3-py37_4.10.3-Linux64.sh
• Miniconda3-py38_4.10.3-Linux-x86_64.sh
• Miniconda3-py39_4.10.3-Linux-x86_64.sh

# 下载软件包
curl -O https://mirrors.tuna.tsinghua.edu.cn/anaconda/miniconda/Miniconda3-latest-Linux-x86_64.sh# 安装&＃xff1a;按照指示输入一次回车&＃xff0c;接受协议&＃xff0c;遇到选择yes&＃xff0c;然后就安装好了
bash Miniconda3-latest-Linux-x86_64.sh


检查环境配置

• 检查是否在文件尾添加如下内容 vi /root/.bashrc
• 加载环境变量 source /root/.bashrc

# >>> conda initialize >>>
# !! Contents within this block are managed by &＃39;conda init&＃39; !!
__conda_setup&＃61;"$(&＃39;/data/flink/miniconda3/bin/conda&＃39; &＃39;shell.bash&＃39; &＃39;hook&＃39; 2> /dev/null)"
if [ $? -eq 0 ]; theneval "$__conda_setup"
elseif [ -f "/data/flink/miniconda3/etc/profile.d/conda.sh" ]; then. "/data/flink/miniconda3/etc/profile.d/conda.sh"elseexport PATH&＃61;"/data/flink/miniconda3/bin:$PATH"fi
fi
unset __conda_setup
# <<

修改镜像地址

将镜像地址修改为国内源
通过如下命令检查配置
conda config --show 或 conda info

• 命令行方式

conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free/
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main/
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/conda-forge
conda config --add channels https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/msys2/
conda config --set show_channel_urls yes


• 修改文件方式
  vim ~/.condarc

default_channels:- https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main/- https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free/- https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/conda-forge/- https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/msys2/
show_channel_urls: true
channels:- https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/main/- https://mirrors.tuna.tsinghua.edu.cn/anaconda/pkgs/free/- https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/conda-forge/- https://mirrors.tuna.tsinghua.edu.cn/anaconda/cloud/msys2/


安装conda-pack

conda-pack是一个命令行工具&＃xff0c;用于创建conda环境的存档文件【archives 】&＃xff0c;这些存档文件可以安装在其他系统上&＃xff0c;这对于在一致的环境中部署代码非常有用。

安装

conda install -c conda-forge conda-pack

使用conda pack进行环境迁移

# 创建pyhon环境
conda create -n py3.7 python&＃61;3.7
conda activate py3.7
conda deactivate
# 查看运行环境列表
conda info -e# 针对py3.7运行环境打包
conda pack --n-threads&＃61;&＃96;nproc&＃96; -n py3.7# 还原环境&＃xff1a;将py3.7.tar.gz上传到其它服务器并解压
mkdir /home/test/miniconda3/envs/py3.7
tar -zxvf /tmp/py3.7.tar.gz -C /home/test/miniconda3/envs/py3.7# 查看运行环境列表
conda info -e
# 激活运行环境
conda activate py3.7


mamba安装

mamba是c&＃43;&＃43;中conda包管理器的重新实现&＃xff0c;可以认为是更高级的conda。
有以下特点

• 使用多线程并行下载repository 数据和包文件&＃xff0c;实现更高效的安装
• libsolv可以更快地解决依赖关系&＃xff0c;libsolv是Red Hat、Fedora和OpenSUSE的RPM包管理器中使用的最新库
• mamba的核心部分是用c&＃43;&＃43;实现的&＃xff0c;以获得最大的效率

与此同时&＃xff0c;mamba利用了相同的命令行解析器、包安装和卸载代码以及事务验证例程&＃xff0c;以尽可能保持与conda的兼容性。

安装

conda install mamba -n base -c conda-forge

额外功能

• 查找软件包
  mamba repoquery search "pandas>0.20.3"
• 查看软件包的依赖&＃xff08;已安装软件包&＃xff09;
  mamba repoquery depends --tree six
• 查看软件包被谁依赖
  mamba repoquery whoneeds openssl

切换pip镜像源

切换pip镜像源&＃xff0c;加速软件包【如flink】的安装

创建或修改~/.pip/pip.conf文件
pip.conf文件的内容如下&＃xff1a;

[global]
timeout &＃61; 6000
index-url &＃61; http://mirrors.aliyun.com/pypi/simple/
[install]
use-mirrors &＃61;true
mirrors &＃61;http://mirrors.aliyun.com/pypi/simple/
trusted-host &＃61;mirrors.aliyun.com

参考链接

Centos 安装 Miniconda
miniconda
conda-pack
mamba
Anaconda、Jupyter的安装部署及使用问题总结
python–切换pip镜像源&＃xff0c;加速软件包的安装