Google Cloud Dataflow Python管道报错：ModuleNotFoundError: No module named 'package'

1. 验证setup.py的包配置正确性 #

确保setup.py能正确识别你的自定义包，避免手动指定包名导致遗漏：

from setuptools import setup, find_packages

setup(
    name='dataflow-pubsub-parser',
    version='0.1',
    packages=find_packages(),  # 自动扫描所有含__init__.py的目录作为包
    install_requires=[
        'apache-beam[gcp]==2.46.0',  # 必须和本地开发环境版本一致
        'google-cloud-pubsub==2.15.0'
    ],
)

关键检查：本地执行pip install -e .后，在Python交互环境中能正常import你的自定义包，确认包结构无问题。

2. 规范save_main_session的使用方式 #

save_main_session=True仅能序列化主会话中已导入的模块，需调整代码导入逻辑：

• 自定义包的导入必须放在脚本全局顶部，不能在run()函数内部导入：

  # 正确：全局导入自定义模块
  from my_package.parsers import PubSubDataParser
  
  def run():
      pipeline_options = beam.PipelineOptions()
      with beam.Pipeline(options=pipeline_options) as p:
          (p
           | 'Read Pub/Sub' >> beam.io.ReadFromPubSub(topic=TOPIC_PATH)
           | 'Parse Data' >> beam.Map(PubSubDataParser.parse)
           )
  
  if __name__ == '__main__':
      run()
  

• 启动管道时必须指定--setup_file=./setup.py，让Dataflow自动打包依赖：

  python -m apache_beam.runners.dataflow.dataflow_runner \
      --project=your-gcp-project \
      --staging_location=gs://your-bucket/staging \
      --temp_location=gs://your-bucket/temp \
      --setup_file=./setup.py \
      --runner=DataflowRunner \
      --save_main_session=True \
      main.py
  

3. 修正Docker镜像的依赖安装逻辑 #

如果使用Docker镜像部署，确保镜像内正确安装自定义包：

FROM python:3.8-slim

WORKDIR /app

# 优先复制setup.py，利用Docker缓存减少构建时间
COPY setup.py ./
RUN pip install --no-cache-dir -e .

# 再复制项目代码
COPY . .

# 指定管道启动命令
CMD ["python", "main.py"]

注意：必须使用-e .安装可编辑模式，保证Dataflow工作节点能识别到你的包。

4. 检查Cloud Build的构建流程 #

确保cloudbuild.yaml正确生成并上传依赖包到GCS：

steps:
- name: 'python:3.8'
  entrypoint: 'pip'
  args: ['install', '-e', '.', '--user']
- name: 'python:3.8'
  entrypoint: 'python'
  args: ['setup.py', 'sdist']
- name: 'gcr.io/cloud-builders/gsutil'
  args: ['cp', 'dist/*.tar.gz', 'gs://your-bucket/dist/']

关键：构建生成的sdist包必须上传到Dataflow可访问的GCS路径，避免工作节点拉取不到依赖。

5. 调试工作节点的依赖环境 #

如果以上步骤无效，在管道中插入调试逻辑，查看工作节点的Python环境：

import sys
import pkg_resources

def debug_environment(element):
    print(f"Python Path: {sys.path}")
    print(f"Installed Packages: {[pkg.key for pkg in pkg_resources.working_set]}")
    return element

# 在管道中添加调试步骤
(p
 | 'Debug Env' >> beam.Map(debug_environment)
)

通过Dataflow日志查看输出，确认自定义包是否在已安装列表中，以及Python路径是否包含包目录。

内容的提问来源于stack exchange，提问作者Jean Walrave