Python SDK 流式数据流运行器答案

【问题标题】：Python SDK Streaming Dataflow RunnerPython SDK 流式数据流运行器
【发布时间】：2018-10-30 22:18:54
【问题描述】：

我有一个用 Python 编写的数据流，我试图在 GCP 上运行。数据流不断终止：

工作流程失败。原因：未知的消息代码。

我的数据流管道中的主要代码是：

schema = 'Member_ID:INTEGER,First_Name:STRING,Last_Name:STRING,Gender:STRING,Age:INTEGER,Height:STRING,weight:INTEGER,Hours_Sleep:INTEGER,Calories_Consumed:INTEGER,Calories_Burned:INTEGER,Evt_Date:DATE,Height_Inches:INTEGER,Min_Sleep_Hours:INTEGER,Max_Sleep_Hours:INTEGER,Enough_Sleep:BOOL'

# read, transform and local source data
p = beam.Pipeline(options=options)

# Read from PubSub into a PCollection.
events = (p | 'Read PubSub' >> beam.io.ReadFromPubSub (topic='projects/prefab-envoy-220213/topics/health_event')
            | 'Parse CSV' >> beam.ParDo(getCSVFields())
            | 'Convert Types' >> beam.ParDo(ConvDataTypes())
            | 'Convert Height' >> beam.ParDo(ConvHeight())
            | 'Join CDC Sleep' >> beam.ParDo(CDCSleepJoin(), cdcsleep)
            | 'Create Row' >> beam.ParDo(CreateRow()) 
            | 'Write to BQ' >> beam.io.Write(beam.io.BigQuerySink(
                    'prefab-envoy-220213:nhcdata.nhcevents', schema=schema,
                    write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
                    create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED))
)

results = p.run()
results.wait_until_finish()

如果我删除了

            | 'Write to BQ' >> beam.io.Write(beam.io.BigQuerySink(
                    'prefab-envoy-220213:nhcdata.nhcevents', schema=schema,
                    write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
                    create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED))

然后数据流开始正常。

【问题讨论】：

标签： python google-bigquery google-cloud-dataflow apache-beam

【解决方案1】：

beam.io.Write(beam.io.BigQuerySink()) 是本机 Dataflow 接收器，它仅适用于批处理作业。对于流式传输管道，您应该改用beam.io.WriteToBigQuery()。用法：

beam.io.WriteToBigQuery(table ='table_name',dataset='dataset',project='project_id')

您可以在answer 中找到一个很好的例子。

【讨论】：