在谷歌云中运行 Apache Beam 作业时找不到错误模块答案

【问题标题】：Error module not found while running Apache beam job in google cloud在谷歌云中运行 Apache Beam 作业时找不到错误模块
【发布时间】：2022-06-18 21:34:31
【问题描述】：

我正在尝试在谷歌云中运行 Apache Beam 作业，但未能成功完成。我尝试了调试和其他故障排除步骤，但每次仍然卡住，这是错误：

  File "/home/avien/.pyenv/versions/dataflow/lib/python3.8/site-packages/apache_beam/transforms/core.py", line 1730, in <lambda>
    wrapper = lambda x: [fn(x)]
  File "xmlload.py", line 59, in <lambda>
NameError: name 'parse_into_dict' is not defined [while running 'parse-ptransform-73']

在没有 lamda 函数的情况下运行并直接将其传递给 beam.Map() 它更改为：

File "/home/avien/.pyenv/versions/dataflow/lib/python3.8/site-packages/apache_beam/transforms/core.py", line 1730, in <lambda>
    wrapper = lambda x: [fn(x)]
  File "xmlload.py", line 36, in parse_into_dict
ModuleNotFoundError: No module named 'xmltodict' [while running 'parse-ptransform-73']

我已经设置了 pyenv 并安装了 xmltodict：

Requirement already satisfied: xmltodict in ./.pyenv/versions/3.8.13/envs/dataflow/lib/python3.8/site-packages (0.13.0)

管道正在尝试运行：

import argparse
import logging
import apache_beam as beam
import xmltodict

def parse_into_dict(xmlfile):
    import xmltodict
    import apache_beam as beam
    with open(xmlfile) as ifp:
        doc = xmltodict.parse(ifp.read())
        return doc

table_schema = {
    'fields': [
        {'name' : 'CustomerID', 'type': 'STRING', 'mode': 'NULLABLE'},
        {'name' : 'EmployeeID', 'type': 'STRING', 'mode': 'NULLABLE'},
        {'name' : 'OrderDate', 'type': 'STRING', 'mode': 'NULLABLE'},
        {'name' : 'RequiredDate', 'type': 'STRING', 'mode': 'NULLABLE'},
        {'name' : 'ShipInfo', 'type': 'RECORD', 'mode': 'NULLABLE', 'fields': [
            {'name' : 'ShipVia', 'type': 'STRING', 'mode': 'NULLABLE'},
            {'name' : 'Freight', 'type': 'STRING', 'mode': 'NULLABLE'},
            {'name' : 'ShipName', 'type': 'STRING', 'mode': 'NULLABLE'},
            {'name' : 'ShipAddress', 'type': 'STRING', 'mode': 'NULLABLE'},
            {'name' : 'ShipCity', 'type': 'STRING', 'mode': 'NULLABLE'},
            {'name' : 'ShipRegion', 'type': 'STRING', 'mode': 'NULLABLE'},
            {'name' : 'ShipPostalCode', 'type': 'STRING', 'mode': 'NULLABLE'},
            {'name' : 'ShipCountry', 'type': 'STRING', 'mode': 'NULLABLE'},
            {'name' : 'ShippedDate', 'type': 'STRING', 'mode': 'NULLABLE'},
        ]},
    ]
}

def cleanup(x):
    import copy
    y = copy.deepcopy(x)
    if '@ShippedDate' in x['ShipInfo']: # optional attribute
        y['ShipInfo']['ShippedDate'] = x['ShipInfo']['@ShippedDate']
        del y['ShipInfo']['@ShippedDate']
    print(y)
    return y

def get_orders(doc):
    for order in doc['Root']['Orders']['Order']:
        yield cleanup(order)

def run(argv=None):
    parser = argparse.ArgumentParser()
    parser.add_argument(
      '--output',
      required=True,
      help=(
          'Specify text file orders.txt or BigQuery table project:dataset.table '))

    known_args, pipeline_args = parser.parse_known_args(argv)
    with beam.Pipeline(argv=pipeline_args) as p:
        orders = (p
             | 'files' >> beam.Create(['orders.xml'])
             | 'parse' >> beam.Map(parse_into_dict)
             | 'orders' >> beam.FlatMap(get_orders))

        if '.txt' in known_args.output:
             orders | 'totxt' >> beam.io.WriteToText(known_args.output)
        else:
             orders | 'tobq' >> beam.io.WriteToBigQuery(known_args.output,
                                       schema=table_schema,
                                       write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND, #WRITE_TRUNCATE
                                       create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED)


if __name__ == '__main__':
    logging.getLogger().setLevel(logging.INFO)
    run()

到目前为止，我已经尝试了以下步骤：

尝试将所有函数包含在管道本身中，但结果相同。
在每个函数中包含所有导入

此外，当在独立的 python 文件中运行 parse_into_dict 时，它根本不会抛出任何错误，并且能够成功地将 xml 转换为 dict。

非常感谢任何帮助，提前致谢！

【问题讨论】：

您是否有包含xmltodict 的requirements.txt 文件？
不，但我在 cloudshell 中使用 pip 安装了它，无论如何，即使在我包含要求并使用 --requirements_file requirements.txt 运行之后，唯一的改变是错误为：NameError: name 'parse_into_dict' is not定义[在运行 'parse-ptransform-73' 时]
您的管道中是否定义了parse_into_dict？因为在您的示例代码中它不是
这是初始代码，之后我将所有函数移动到管道中，并在每个函数中包含所有导入以确保安全，即使使用 --save_main_session 标记但仍然不能作为应该是的，当通过 directrunner 本地运行时，一切正常。

标签： python google-cloud-platform apache-beam

【解决方案1】：

尝试在函数和管道定义中导入模块；或使用--save_main_session。 NameErrors 很常见，因为工作人员不知道全局命名空间中定义的对象。

【讨论】：

wrapper = lambda x: [fn(x)] File "xmlload1.py", line 38, in parse_into_dict RuntimeError: FileNotFoundError: [Errno 2] No such file or directory: 'orders.xml' [while running 'parse-ptransform-73'] 现在它已转换为此错误，在本地运行（即通过 Directrunner）时不会发生这些错误。

【解决方案2】：

除了@ningk 答案，您还必须为数据流提供您的orders.xml 文件。您正尝试在管道的第一步 (beam.Create['orders.xml']) 中加载此文件，但是，数据流在执行您的管道时不知道/拥有此文件。

尝试添加具有以下内容的MANIFEST.in 文件（注意大写）

include path/to/xml/orders.xml

在管道代码的源文件夹中。有关示例文件，请参阅 here。

【讨论】：