GCP Dataflow作业运行报ModuleNotFoundError找不到utils模块如何解决

报错根因 #

DataflowRunner是分布式运行模式，作业逻辑会分发到GCP托管的远程工作节点执行，工作节点默认只有基础运行环境，不会自动同步你本地项目中的自定义utils模块，所以会触发模块找不到的报错。DirectRunner是本地运行模式，直接读取本地路径下的项目文件，所以运行正常。
你之前加--setup_file后作业卡住，是因为启动命令写错了，参数没有被正确解析导致作业初始化失败。 #

修复步骤 #

1. 修正启动命令 #

你原有命令里多了一个多余的横杠，且setup_file的路径不需要加尖括号，直接在项目根目录（root_dir）下执行以下命令即可：

python3 main.py --setup_file ./setup.py

2. 可选：提前打包本地依赖 #

如果上述命令还是运行异常，可以先本地把项目打成源码包再提交作业：

• 先在项目根目录执行打包命令：

  python setup.py sdist
  

  执行完成后会在dist目录下生成dataflow_example-1.0.tar.gz包
• 再用打包好的文件提交作业：

  python3 main.py --sdk_location ./dist/dataflow_example-1.0.tar.gz
  

3. 可选：优化setup.py配置 #

如果还是识别不到utils模块，可以修改setup.py明确指定要打包的自定义模块，同时确认首行导入了setuptools：

import setuptools # 确保首行有这行导入
setuptools.setup(
   name='dataflow_example',
   version='1.0',
   install_requires=[
        "google-cloud-tasks==2.2.0",
        "google-cloud-pubsub>=0.1.0",
        "google-cloud-storage==1.39.0",
        "google-cloud-bigquery==2.6.2",
        "google-cloud-secret-manager==2.0.0",
        "google-api-python-client==2.3.0",
        "oauth2client==4.1.3",
        "apache-beam[gcp]>=2.20.0",
        "wheel>=0.36.2"
   ],
   packages=["utils"], # 替换原来的setuptools.find_packages()
   include_package_data=True
)

4. 可选：规避序列化问题 #

你代码中设置了save_main_session=True，如果主模块导入的对象无法序列化也会导致作业异常，可以把自定义模块的导入移到DoFn的process方法内部：

class PrintCount(beam.DoFn):
    def process(self, row_count, batch_size):
        from utils.log_util import LogUtil # 移到方法内部导入
        logger = LogUtil()
        # 剩下的原有逻辑保持不变

内容的提问来源于stack exchange，提问作者Gopinath S