Dataflow 适用于 directrunner 但不适用于 dataflowrunner（PubSub 到 GCS）答案

【问题标题】：Dataflow works with directrunner but not with dataflowrunner (PubSub to GCS)Dataflow 适用于 directrunner 但不适用于 dataflowrunner（PubSub 到 GCS）
【发布时间】：2023-03-29 05:09:01
【问题描述】：

我正在用数据流做一个非常简单的管道。

它从 pubsub 获取原始数据并添加时间戳，然后写入原始文件（我先尝试了 parquet）。

代码：

class GetTimestampFn(beam.DoFn):
    """Prints element timestamp"""
    def process(self, element, timestamp=beam.DoFn.TimestampParam):
        timestamp_utc = float(timestamp)
        yield {'raw':str(element),"inserted_at":timestamp_utc}


options = PipelineOptions(streaming=True)
p = beam.Pipeline(DirectRunner(), options=options)
parser = argparse.ArgumentParser()
parser.add_argument('--input_topic',required=True)
parser.add_argument('--output_parquet',required=True)
known_args, _ = parser.parse_known_args(argv)


raw_data = p | 'Read'  >> beam.io.ReadFromPubSub(subscription=known_args.input_topic) 
raw_with_timestamp = raw_data | 'Getting Timestamp' >> beam.ParDo(GetTimestampFn())
_ = raw_with_timestamp | 'Write' >> beam.io.textio.WriteToText(known_args.output_parquet,append_trailing_newlines=True
    ,file_name_suffix='.gzip'
)

p.run().wait_until_finish()

它适用于直接运行器，但它在 dataflowrunner 上失败并显示此消息“工作流失败”。

职位编号：2021-04-14_17_11_02-16453427249129279174

我是如何开展这项工作的：

python  real_time_events.py \
--region us-central1 \
--input_topic 'projects/{project}/subscriptions/{subscription}' \
--output_parquet 'gs://{bucket}/stream/' \
--project "{project}" \
--temp_location "gs://{bucket}/tmp" \
--staging_location "gs://{bucket}/stage"

关于如何解决的任何想法？

【问题讨论】：

尝试在参数中添加--save_main_session
谢谢，刚刚试了还是一样的错误。

标签： python google-cloud-dataflow dataflow

【解决方案1】：

WriteToText 不支持流式传输管道。我建议您尝试改用fileio.WriteToFiles，以获得支持流式传输的转换。请注意，您可能需要在它之前进行分组。

【讨论】：