使用 Dataflowrunner 时无法使用本机 Python3 库 (ElementTree)

Question

我目前正在使用 Dataflow 处理/按摩来自 PubSub 的 XML 文本字符串。我能够使用 DirectRunner 作为我的 --runner 标志成功运行 Dataflow 作业。但是，我在尝试使用完全相同的 Dataflow 作业以 DataflowRunner 作为我的标志创建 Dataflow 资源时遇到了问题。

从错误日志（使用 DataflowRunner 时）看来，我创建的 Dataflow 模板似乎无法识别：

import xml.etree.ElementTree as ET

每当我在管道中引用 ET 时，我都会收到“NameError: name 'ET' is not defined [while running 'generatedPtransform-419']”。奇怪的是，我的 Dataflow 作业在 DirectRunner 上运行得非常好，这让我相信使用 DataflowRunner 构建我的模板存在问题，因为 xml.etree.ElementTree 是一个简单/原生的 PyPI 库。

对于我的环境，我正在使用：
Python 3.7.7
apache-beam 2.22.0

非常感谢任何帮助/指导，谢谢！

工作directrunner工作：

import apache_beam as beam
import argparse, xmltodict, json
from apache_beam.options.pipeline_options import PipelineOptions, SetupOptions
import xml.etree.ElementTree as ET

class FormatMessage(beam.DoFn):
    def process(self, line):
        xml_msg = ET.fromstring(line)

        # Code to construct XML Object (removed) 

        tree_new_xml = ET.ElementTree(element_msg)
        xml_dict = xmltodict.parse(ET.tostring(tree_new_xml.getroot(), encoding='utf8'))
        json_obj = str.encode(json.dumps(xml_dict), 'utf8')

        yield json_obj

def run(argv=None):
    parser = argparse.ArgumentParser()
    parser.add_argument('--input_topic', help='Input topic read data from.', default='')
    parser.add_argument('--output_topic', help='Output topic to write data to.', default='')
    known_args, pipeline_args = parser.parse_known_args(argv)

    pipeline_options = PipelineOptions()
    pipeline_options.view_as(SetupOptions).save_main_session = True

    # Create and implement PubSub-to-PubSub pipeline
    p = beam.Pipeline(options=pipeline_options)
    (p
     | "Read PubSub Message" >> beam.io.ReadFromPubSub(topic=known_args.input_topic)
     | "Format Msg" >> beam.ParDo(FormatMessage())
     | "Write PubSub Output" >> beam.io.WriteToPubSub(known_args.output_topic)
     )
    p.run().wait_until_finish()

Answer 1

根据这个documentation，您的问题是由于名称在 Dataflow 工作器上不可用。

请注意，如果您的全局命名空间中有不能 被腌制，你会得到一个腌制错误。如果错误是关于 应该在 Python 发行版中可用的模块，您可以 通过在本地导入模块来解决这个问题。

在您的情况下，您必须在 process 函数中导入提到的库，如下所示：

class FormatMessage(beam.DoFn):
    def process(self, line):
        import xml.etree.ElementTree as ET
        xml_msg = ET.fromstring(line)

        # Code to construct XML Object (removed) 

        tree_new_xml = ET.ElementTree(element_msg)
        xml_dict = xmltodict.parse(ET.tostring(tree_new_xml.getroot(), encoding='utf8'))
        json_obj = str.encode(json.dumps(xml_dict), 'utf8')

        yield json_obj