PyFlink是Fl??ink的Python语言入口。它的Python语言确实非常简单易学，但是PyFlink的开发环境搭建起来并不容易。很难调查原因。今天给大家介绍一款可以帮助大家解决这些问题的PyFlink开发环境工具：ZeppelinNotebook。主要内容有：1.准备工作2.搭建PyFlink环境3.总结与未来也许你听说过Zeppelin，但之前的几篇文章主要介绍了如何在Zeppelin中开发FlinkSQL。今天，我将介绍如何在Zeppelin中使用Zeppelin。高效开发PyFlinkJob，特别是解决PyFlink的环境问题。一句话总结这篇文章的主题就是使用Conda在Zeppelinnotebook中创建Pythonenv并自动部署到Yarn集群。不需要在集群上手动安装任何PyFlink包，可以在一个Yarn集群中同时使用。多个版本的PyFlink被隔离。最终可以看到的效果是这样的：1.可以在PyFlink客户端使用第三方Python库，比如matplotlib：2.可以在PyFlinkUDF中使用第三方Python库，比如：下面看看如何实现它。一、准备步骤一、准备搭建最新版本的Zeppelin，这里不再赘述。有问题可以加入FlinkonZeppelin钉钉群（34517043）进行咨询。另外需要注意的是Zeppelin部署集群需要是Linux的。如果是Mac，在Mac机器上创建的Conda环境无法在Yarn集群中使用（因为Conda包在不同系统之间不兼容）。步骤2.下载Flink1.13。需要注意的是，本文中的功能只能在Flink1.13以上的版本中使用。然后：将flink-Python-*.jar这个jar包复制到Flink的lib文件夹下；把opt/Python这个文件夹复制到Flink的lib文件夹下。步骤3.安装以下软件（这些软件用于创建Condaenv）：miniconda：https://docs.conda.io/en/latest/miniconda.htmlcondapack：https://conda.github.io/conda-pack/mamba:https://github.com/mamba-org/mamba2.搭建PyFlink环境接下来就可以在Zeppelin中搭建并使用PyFlink了。Step1.在JobManager上创建PyFlinkConda环境由于Zeppelin天然支持Shell，您可以使用Shell在Zeppelin中创建PyFlink环境。注意这里的Python第三方包是PyFlink客户端（JobManager）需要的包，比如Matplotlib，确保至少安装了以下包：某个版本的Python（这里用的是3.7）apache-flink（这里使用1.13.1)jupyter,grpcio,protobuf（这三个包是Zeppelin需要的）其余包可以根据需要指定：%sh#makesureyouhavecondaandmombainstalled.#installminiconda:https://docs.conda.io/en/最新/miniconda.html#installmamba: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.ymlmambaenvremove-npyflink_envmambaenvcreate-fpyflink_env.yml运行如下代码将PyFlinkConda环境打包上传到HDFS（注意打包好的这里的文件格式是tar.gz):%shrm-rfpyflink_env.tar.gzcondapack--ignore-missing-files-npyflink_env-opyflink_env.tar.gzhadoopfs-rmr/tmp/pyflink_env.tar.gzhadoopfs-putpyflink_env.tar.gz/tmp#Pythoncondatars应该是可公开访问的，需要在这里更改权限。hadoopfs-chmod644/tmp/pyflink_env.tar.gz步骤2。做TaskManager上的PyFlinkConda环境运行以下代码在TaskManager上创建PyFlinkConda环境。TaskManager上的PyFlink环境至少包含以下两个包：某个版本的Python（这里用的是3.7）apache-flink（这里用的是1.13.1）剩下的包就是PythonUDF需要依赖的包。比如这里指定pandas：echo"name:pyflink_tm_envchannels:-conda-forge-defaultsdependencies:-Python=3.7-pip-pip:-apache-flink==1.13.1-pandas">pyflink_tm_env.ymlmambaenvremove-npyflink_tm_envmambaenvcreate-fpyflink_tm_env.yml运行如下代码打包PyFlink的conda环境并上传到HDFS（注意这里使用的是zip格式）%shrm-rfpyflink_tm_env.zipcondapack--ignore-missing-files--zip-symlinks-npyflink_tm_env-opyflink_tm_env.ziphadoopfs-rmr/tmp/pyflink_tm_env.ziphadoopfs-putpyflink_tm_env.zip/tmp#ThePythoncondatarshouldbepublicaccessible,soneedtochangepermissionhere.hadoopfs-chmod644/tmp/pyflink_tm_env.zipStep3.在PyFlink中使用Conda环境接Afterthat,youcanusetheCondaenvironmentcreated上面是齐柏林飞艇。首先，你需要在Zeppelin中配置Flink。主要配置选项有：flink.execution.mode为yarn-application，本文介绍的方法仅适用于yarn-application模式；指定yarn.ship-archives,zeppelin.pyflink.Python和zeppelin.interpreter.conda.env.name用于在JobManager端配置PyFlinkConda环境；指定Python.archives和Python.executable来指定TaskManager端的PyFlinkConda环境；指定其他可选的Flink配置，比如这里的flink.jm.memory和flink.tm.memory.Pythonpyflink_env.tar.gz/bin/Pythonzeppelin.interpreter.conda.env.namepyflink_env.tar.gzPython.archiveshdfs:///tmp/pyflink_tm_env.zipPython.executablepyflink_tm_env.zip/bin/Python3.7flink.jm.memory2048flink.tm。memory2048那么就可以和开头说的一样在Zeppelin中使用PyFlink和指定的Conda环境。有两种场景：下面的例子，PyFlink客户端（JobManager端）可以使用上面创建的JobManager端的Conda环境，例如下面使用Matplotlib。下面的例子是在PyFlinkUDF中使用上面创建的TaskManager端的Conda环境中的库，例如下面在UDF中使用Pandas。3.总结与未来本文内容是在Zeppelinnotebook中使用Conda创建Pythonenv并自动部署到Yarn集群。集群上无需手动安装任何Pyflink包，一个Yarn集群可以同时使用多个版本的PyFlink。.每个PyFlink环境都是隔离的，Conda环境可以随时定制和更改。大家可以下载下面的note导入Zeppelin来复现我今天说的：http://23.254.161.240/#/notebook/2G8N1WTTS另外还有很多可以改进的地方：目前我们需要创建2个condaenv，原因是Zeppelin支持tar.gz格式，而Flink只支持zip格式。后期双方统一后，创建condaenv即可；apache-flink现在包含了Flink的jar包，导致condaenv特别大，yarn容器的初始化时间会比较长。这就需要Flink社区提供一个轻量级的Python包（不包括Flinkjar包），可以大大减少condaenv的体积。
