在Google Dataproc运行Beam Pipeline时遇ModuleNotFoundError，如何修复？

问题分析 #

报错的核心原因是Spark Executor节点无法找到你的自定义Python模块或依赖包——官方文档的单文件示例仅打包主脚本，并未处理多目录结构下的子模块和外部依赖分发问题。

具体解决方案 #

1. 打包完整Python项目 #

如果你的项目是多目录结构（示例如下）：

my_project/
├── main.py
└── utils/
    ├── __init__.py
    └── helper.py

需要将整个项目打包成wheel包（推荐）或zip包（需确保含__init__.py让Python识别为合法模块）：

• 用setuptools打包：
  在项目根目录创建setup.py文件：

  from setuptools import setup, find_packages
  
  setup(
      name="my_beam_project",
      version="0.1",
      packages=find_packages(),
      install_requires=[
          # 列出所有外部依赖，比如apache-beam、pandas等
          "apache-beam[spark]",
          "pandas==2.0.3"
      ]
  )
  

  执行打包命令生成包文件：

  python setup.py sdist bdist_wheel
  

  生成的包会存放在dist/目录下。

2. 提交作业时分发依赖 #

提交Dataproc作业时，需将自定义包和外部依赖同步到所有集群节点：

• 方式一：通过--py-files参数指定
  将打包好的wheel/zip包、外部依赖包（若集群未预装）通过该参数传递：

  gcloud dataproc jobs submit pyspark main.py \
      --cluster=your-cluster-name \
      --region=your-region \
      --py-files=dist/my_beam_project-0.1-py3-none-any.whl,requests-2.31.0-py3-none-any.whl
  

  若依赖可通过pip安装，也可启用Spark虚拟环境自动安装：

  gcloud dataproc jobs submit pyspark main.py \
      --cluster=your-cluster-name \
      --region=your-region \
      --properties spark.pyspark.python=python3,spark.pyspark.virtualenv.enabled=true,spark.pyspark.virtualenv.type=native,spark.pyspark.virtualenv.bin.path=/usr/bin/virtualenv \
      --requirements-file=requirements.txt
  

• 方式二：集群初始化时预安装依赖
  针对长期运行的集群，可通过初始化脚本在集群创建时预装依赖：

  gcloud dataproc clusters create your-cluster-name \
      --region=your-region \
      --initialization-actions=gs://your-bucket/init-script.sh
  

  init-script.sh内容示例：

  #!/bin/bash
  pip3 install apache-beam[spark] pandas requests gs://your-bucket/my_beam_project-0.1-py3-none-any.whl
  

3. 配置PipelineOptions参数 #

在Beam代码中设置SparkRunner相关参数，确保与集群环境匹配：

from apache_beam.options.pipeline_options import PipelineOptions, SparkRunnerOptions

options = PipelineOptions()
spark_options = options.view_as(SparkRunnerOptions)
spark_options.spark_python_version = '3.8'  # 需与集群Python版本一致
# 若使用虚拟环境，指定路径
# spark_options.spark_virtual_env = '/path/to/virtualenv'

4. 关键注意事项 #

• 保证集群节点Python版本与本地开发环境一致，避免版本兼容问题。
• 自定义模块使用绝对导入（如from utils.helper import func），避免相对导入在集群环境中失效。
• 确认打包时所有子目录都包含__init__.py，确保Python能识别为模块。

内容的提问来源于stack exchange，提问作者Asket Agarwal