GCP运行Apache Beam作业报ModuleNotFoundError错误求助

两个报错都来自Apache Beam分布式运行机制和本地单机运行的差异，和你本地pyenv环境配置、函数在代码里的位置没有直接关系：

• NameError: name 'parse_into_dict' is not defined：Beam提交作业时需要把自定义函数序列化后分发到worker节点，这个报错是函数序列化/分发失败导致的，常见原因是主代码文件没有被正确上传到worker、或者函数定义在非全局作用域导致序列化时找不到引用。
• ModuleNotFoundError: No module named 'xmltodict'：你本地pyenv安装的依赖只存在于你的开发机上，GCP Dataflow的worker节点是独立的运行环境，默认不会预装你本地装的第三方包，运行时自然找不到xmltodict。
• 还有一个你暂时没碰到的隐藏bug：当前代码里用open(xmlfile)读本地路径的orders.xml，等作业跑到Dataflow worker上时会直接报文件不存在——worker节点上根本没有你开发机里的本地文件。

1. 配置worker节点依赖 #

提交Dataflow作业时必须显式指定worker需要安装的第三方包，不要依赖函数内部的import语句解决依赖问题——包本身不存在的话，写在哪import都会报错。
最简便的方式是在项目根目录新建requirements.txt，写入：

xmltodict==0.13.0

提交作业时追加参数，让Dataflow自动在所有worker节点安装依赖：

--requirements_file=./requirements.txt

2. 修复函数序列化问题 #

• 确认parse_into_dict/cleanup/get_orders这几个自定义函数全部定义在Python文件的全局作用域，不要嵌套在run()函数内部或者if __name__ == '__main__'判断块里。
• 提交作业时确保命令执行的当前目录就是xmlload.py所在的目录，Beam会自动把主代码文件分发到所有worker；如果是多文件项目，需要额外通过--setup_file参数配置打包规则，把所有本地依赖代码一起上传。
• 保持beam.Map(parse_into_dict)这种直接传函数引用的写法，不要额外套lambda，避免序列化时出现引用找不到的问题。

3. 修复文件读取逻辑 #

不要用原生open()读文件，换成Beam内置的文件系统接口，同时先把orders.xml上传到GCS存储桶：

1. 先把本地的orders.xml上传到你自己的GCS路径，比如gs://your-project-bucket/orders.xml
2. 把代码里的文件读取部分替换成跨环境兼容的写法，修改parse_into_dict函数：

from apache_beam.io import filesystems

def parse_into_dict(xmlfile_path):
    import xmltodict
    with filesystems.FileSystems.open(xmlfile_path) as ifp:
        doc = xmltodict.parse(ifp.read())
        return doc

3. 把beam.Create(['orders.xml'])里的路径替换成你上传后的GCS路径即可。这个接口在本地用DirectRunner跑的时候会自动识别本地文件路径，提交到Dataflow的时候会自动识别GCS路径，不会出现跨环境文件找不到的问题。

4. 提交前本地验证 #

不要直接提交Dataflow作业，先本地用默认的DirectRunner跑通全流程：

• 输出路径指定本地txt文件，确认能正常解析XML、输出结果无报错
• 确认本地运行时调用的是你配置的pyenv环境，避免本地环境和提交环境的版本差异
  本地验证通过后，再带上依赖参数、GCS文件路径提交Dataflow作业即可。

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