本文重點介紹下如何在 Zeppelin 里高效的開發 PyFlink Job，特別是解決 PyFlink 的環境問題，

大家都知道PyFlink 的開發環境不容易搭建，稍有不慎，PyFlink 環境就會亂掉，而且很難排查原因
本文使用miniconda、conda-pack、mamba分別制作JobManager 上的 PyFlink Conda 環境、TaskManager 上的 PyFlink Conda 環境，然后在 Zeppelin 里使用 PyFlink 以及指定 Conda 環境，這樣就可以在一個 Yarn 集群里同時使用多個版本的 PyFlink，

本文使用環境：flink 1.13.2， zeppelin：0.10.0-bin-all

需要改進的地方：

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

1. 搭建 PyFlink 環境

1.1.制作 JobManager 上的 PyFlink Conda 環境

• 注意dependencies所列的第三方包是在 PyFlink 客戶端 (JobManager) 需要的包，比如 Matplotlib ，并且確保至少安裝了所列的這些包：
• jupyter，grpcio，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=3.7
  - pip
  - pip:
    - apache-flink==1.13.2
  - 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

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

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=3.7
  - pip
  - pip:
    - apache-flink==1.13.2
  - pandas
 " > pyflink_tm_env.yml

mamba env remove -n pyflink_tm_env   
mamba env create -f pyflink_tm_env.yml

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

1.3.安裝本地的flink

下載 flink 1.13并解壓，然后：

• 把 opt目錄下的flink-python-*.jar 這個 jar 包 copy 到flink 的 lib 檔案夾下；
• 把 opt/python 這個檔案夾 copy 到flink 的 lib 檔案夾下，

1.4. 在 PyFlink 中使用 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_HOME	/data/flink/flink-1.13.2	
HADOOP_CONF_DIR	/etc/hadoop/conf	
HIVE_CONF_DIR	/etc/hive/conf
flink.execution.mode	yarn-application

zeppelin.pyflink.python	python

yarn.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	2048

zeppelin.interpreter.connect.timeout	600000

1.5. 使用示例

%flink.ipyflink
%matplotlib inline
import matplotlib.pyplot as plt
plt.plot([1,2,3,4])
plt.ylabel('numbers')
plt.show()


%flink.ipyflink
import time
class PandasVersion(ScalarFunction):
  def eval(self,s):
    import pandas as pd
    return pd.__version__ + " " + s
bt_env.register_funtion("pandas_version",udf(PandasVersion(),DataTypes.STRING,DataTypes.STRING)


%bsql
select pandas_version('hello world')

附錄

準備conda環境

在centos 7.x上安裝miniconda

安裝miniconda

miniconda是一個免費的conda最小安裝程式，
它是Anaconda的一個小型的引導版本，只包含conda、Python、它們所依賴的包，以及少量其他有用的包，包括pip、zlib和其他一些包

下載并安裝

根據實際需要，下載對應的版本，latest表示最新的python版本，是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

# 安裝：按照指示輸入一次回車，接受協議，遇到選擇yes，然后就安裝好了
bash Miniconda3-latest-Linux-x86_64.sh

檢查環境配置

• 檢查是否在檔案尾添加如下內容 vi /root/.bashrc
• 加載環境變數 source /root/.bashrc

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

修改鏡像地址

將鏡像地址修改為國內源
通過如下命令檢查配置
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是一個命令列工具，用于創建conda環境的存檔檔案【archives 】，這些存檔檔案可以安裝在其他系統上，這對于在一致的環境中部署代碼非常有用，

安裝

conda install -c conda-forge conda-pack

使用conda pack進行環境遷移

# 創建pyhon環境
conda create -n py3.7 python=3.7
conda activate py3.7
conda deactivate
# 查看運行環境串列
conda info -e

# 針對py3.7運行環境打包
conda pack --n-threads=`nproc` -n py3.7

# 還原環境：將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++中conda包管理器的重新實作，可以認為是更高級的conda，
有以下特點

• 使用多執行緒并行下載repository 資料和包檔案，實作更高效的安裝
• libsolv可以更快地解決依賴關系，libsolv是Red Hat、Fedora和OpenSUSE的RPM包管理器中使用的最新庫
• mamba的核心部分是用c++實作的，以獲得最大的效率

與此同時，mamba利用了相同的命令列決議器、包安裝和卸載代碼以及事務驗證例程，以盡可能保持與conda的兼容性，

安裝

conda install mamba -n base -c conda-forge

額外功能

• 查找軟體包
  mamba repoquery search "pandas>0.20.3"
• 查看軟體包的依賴（已安裝軟體包）
  mamba repoquery depends --tree six
• 查看軟體包被誰依賴
  mamba repoquery whoneeds openssl

切換pip鏡像源

切換pip鏡像源，加速軟體包【如flink】的安裝

創建或修改~/.pip/pip.conf檔案
pip.conf檔案的內容如下：

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

參考鏈接

Centos 安裝 Miniconda
miniconda
conda-pack
mamba
Anaconda、Jupyter的安裝部署及使用問題總結
python–切換pip鏡像源，加速軟體包的安裝