无法创建支持架构的Pcollection以在Google数据流中运行

Question

出于某种原因，每次我尝试在我的管道（ApacheBeamPython）中创建schema ware PCollection时，使用Beam。选择（）。Row（）并尝试在数据流上运行它，我得到一个如下所示的致命错误，该错误导致工作进程在VM上死亡。奇怪的是，当我使用Directrunner在本地运行时，我没有遇到这个错误。

我想确保这不是我在自定义代码中犯的愚蠢错误。我尝试了谷歌的例子，比如下面的片段简化自：https://github.com/apache/beam/blob/034ccdf93a0c5dfe6629501b456105ac47047e44/sdks/python/apache_beam/examples/sql_taxi.py

import json
import logging

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

PROJECT_ID = <my_project_id>

def run(pipeline_args):
  pipeline_options = PipelineOptions(
      pipeline_args, project=PROJECT_ID, save_main_session=True, streaming=True)

  with beam.Pipeline(options=pipeline_options) as pipeline:
    _ = (
        pipeline
        | beam.io.ReadFromPubSub(
            topic='projects/pubsub-public-data/topics/taxirides-realtime',
            timestamp_attribute="ts").with_output_types(bytes)
        | "Parse JSON payload" >> beam.Map(json.loads)
        # Use beam.Row to create a schema-aware PCollection
        # ** This gives me trouble everytime when run through Dataflow **
        | "Create beam Row" >> beam.Map(
            lambda x: beam.Row(
                ride_status=str(x['ride_status']),
                passenger_count=int(x['passenger_count']))))

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

  parser = argparse.ArgumentParser()
  _, pipeline_args = parser.parse_known_args()

  run(pipeline_args)

完全相同的问题。在Directrunner中本地运行时非常好；在数据流中运行时崩溃。我使用的是ApacheBeam2.31。0 python SDK

我为此伤了好几天脑筋。任何帮助或指导都将不胜感激！！

Answer 1

多亏了@大ドア東, 我能够使用NamedTuple让它在Google数据流上工作。

然而，我仍然对那束光束感到震惊。行（）或梁。Select（）在Apache Beam python SDK的数据流上根本不起作用，我在stackoverflow上找不到类似的抱怨。