PySpark 依赖管理集群环境下如何分发 Python 包

1. 为什么 PySpark 需要做依赖管理当 PySpark 程序运行在集群上时Executor 也要执行 Python 代码。如果你的程序依赖pandas、pyarrow之类的第三方库那么这些库必须在每个 Executor 上都能被找到否则就会报错例如ModuleNotFoundError: No module named pyarrow这类问题在使用 Pandas UDF 时尤其常见因为底层依赖pyarrow。2. 一个典型示例下面这段官方示例代码使用了 Pandas UDF因此需要在 Driver 和 Executor 上都具备pandas与pyarrow相关依赖。importpandasaspdfrompyspark.sql.functionsimportpandas_udffrompyspark.sqlimportSparkSessiondefmain(spark):dfspark.createDataFrame([(1,1.0),(1,2.0),(2,3.0),(2,5.0),(2,10.0)],(id,v))pandas_udf(double)defmean_udf(v:pd.Series)-float:returnv.mean()print(df.groupby(id).agg(mean_udf(df[v])).collect())if__name____main__:main(SparkSession.builder.getOrCreate())3. PySpark 依赖管理的几种方式官方文档总结了几种主要方案PySpark 原生方式CondaVirtualenvPEX它们的核心差别在于能否打包解释器、能否打包带原生代码的依赖、是否适合大规模集群环境。4. 方式一使用 PySpark 原生能力PySpark 可以通过以下方式把 Python 代码分发到 Executor配置spark.submit.pyFiles使用--py-files在程序中调用SparkContext.addPyFile()4.1 常见用法spark-submit --py-files deps.zip app.py或者在代码里spark.sparkContext.addPyFile(deps.zip)4.2 适用场景这种方式适合分发单个.py文件.zip打包的 Python 包.egg文件4.3 局限性它不能分发 Wheel也不适合携带带有本地编译代码的依赖因此像某些依赖 C/C 扩展的包就不太适合这种方式。官方明确指出这种方式不支持包含原生代码依赖的场景。5. 方式二使用 CondaConda 是最常见的 Python 包管理工具之一。官方推荐使用conda-pack把整个 Conda 环境打包后分发到 Driver 和 Executor。这样不仅带上依赖还能带上 Python 解释器本身。5.1 构建环境conda create-y-npyspark_conda_env-cconda-forge pyarrow pandas conda-pack conda activate pyspark_conda_env conda pack-f-opyspark_conda_env.tar.gz5.2 使用 spark-submit 提交exportPYSPARK_DRIVER_PYTHONpythonexportPYSPARK_PYTHON./environment/bin/python spark-submit--archivespyspark_conda_env.tar.gz#environment app.py5.3 在普通 Python Shell 或 Notebook 中使用importosfrompyspark.sqlimportSparkSessionfromappimportmain os.environ[PYSPARK_PYTHON]./environment/bin/pythonsparkSparkSession.builder.config(spark.archives,pyspark_conda_env.tar.gz#environment).getOrCreate()main(spark)5.4 在 pyspark shell 中使用exportPYSPARK_DRIVER_PYTHONpythonexportPYSPARK_PYTHON./environment/bin/python pyspark--archivespyspark_conda_env.tar.gz#environment5.5 注意事项官方特别说明在 YARN 或 Kubernetes 的 cluster mode 下不要设置PYSPARK_DRIVER_PYTHON。6. 方式三使用 VirtualenvVirtualenv 也是常见的 Python 隔离环境方案。PySpark 可以配合venv-pack打包虚拟环境并通过--archives或spark.archives分发到 Executor。6.1 构建虚拟环境python-mvenv pyspark_venvsourcepyspark_venv/bin/activate pipinstallpyarrow pandas venv-pack venv-pack-opyspark_venv.tar.gz6.2 spark-submit 提交方式exportPYSPARK_DRIVER_PYTHONpythonexportPYSPARK_PYTHON./environment/bin/python spark-submit--archivespyspark_venv.tar.gz#environment app.py6.3 Notebook / Python Shell 用法importosfrompyspark.sqlimportSparkSessionfromappimportmain os.environ[PYSPARK_PYTHON]./environment/bin/pythonsparkSparkSession.builder.config(spark.archives,pyspark_venv.tar.gz#environment).getOrCreate()main(spark)6.4 pyspark shell 用法exportPYSPARK_DRIVER_PYTHONpythonexportPYSPARK_PYTHON./environment/bin/python pyspark--archivespyspark_venv.tar.gz#environment6.5 注意事项官方说明venv-pack打包的 Python 解释器是符号链接因此集群中所有节点都需要安装相同的 Python 解释器。也就是说Virtualenv 方式对节点环境一致性要求更高。7. 方式四使用 PEXPEX 是另一种常见方案它会把依赖打成一个.pex可执行文件。与 Conda 和 Virtualenv 不同PEX 文件本身不包含 Python 解释器因此集群上所有节点也要有相同版本的 Python。7.1 构建 PEX 文件pipinstallpyarrow pandas pex pex pyspark pyarrow pandas-opyspark_pex_env.pex7.2 验证 PEX 文件./pyspark_pex_env.pex-cimport pandas; print(pandas.__version__)7.3 spark-submit 提交方式exportPYSPARK_DRIVER_PYTHONpythonexportPYSPARK_PYTHON./pyspark_pex_env.pex spark-submit--filespyspark_pex_env.pex app.py7.4 Notebook / Python Shell 用法importosfrompyspark.sqlimportSparkSessionfromappimportmain os.environ[PYSPARK_PYTHON]./pyspark_pex_env.pexsparkSparkSession.builder.config(spark.files,pyspark_pex_env.pex).getOrCreate()main(spark)7.5 pyspark shell 用法exportPYSPARK_DRIVER_PYTHONpythonexportPYSPARK_PYTHON./pyspark_pex_env.pex pyspark--filespyspark_pex_env.pex7.6 注意事项因为.pex是普通文件而不是目录或压缩包所以它需要通过--files或spark.files传输而不是--archives。8. 四种方案怎么选如果只是分发你自己的少量 Python 代码原生--py-files最简单。如果需要完整、可迁移的 Python 运行环境Conda 会更稳。如果团队已经统一使用 virtualenv也可以用 Virtualenv venv-pack。如果你想把依赖收敛成一个单独文件PEX 会更方便。更直白一点只分发纯 Python 代码优先--py-files依赖复杂包含解释器和环境优先 Conda已有 virtualenv 体系可选 Virtualenv希望单文件分发可选 PEX9. 总结PySpark 在集群环境中的依赖管理本质上是在解决一个问题如何让 Driver 和所有 Executor 拥有一致的 Python 运行环境。官方文档给出的四种方案中没有绝对最好的选项只有更适合当前场景的方案。简单场景可以用原生方式复杂依赖环境更适合 Conda 或 Virtualenv追求单文件交付则可以考虑 PEX。只要这个思路想清楚PySpark 集群部署时的大多数依赖问题都能提前规避。