无法将值从 pubsub 插入到 GCP 中的大查询

Question

我是 Python Apache Beam 的新手，正在尝试执行一项任务以从 pubsub 中提取值并使用 Google 云数据流将其传递给大查询。请在下面找到我针对此场景执行的步骤。

1) 数据作为 json 推送到 pubsub 主题中。 {-----record1 ----}{---record2---} 等等

2) 大查询 table 遵循与推送到 pubsub 的 json 完全相同的键，所有列都是字符串类型。

3) 下面是场景的代码

pipeline_options.view_as(StandardOptions).streaming = True
p_bq = beam.Pipeline(options=pipeline_options)
logging.info('Start')

BQ_data1 = (p_bq 
            | 'readPubSub' >> beam.io.ReadFromPubSub(subscription='subscriptionname') 
            | 'writeBQ' >> beam.io.WriteToBigQuery(table='table name', 
                                                   dataset='dataset',
                                                   project='projectname'))
result_bq = p_bq.run()

预期结果：将值从 pubsub 传递到大查询

实际结果： AttributeError: 'str' 对象没有属性 'items'

Answer 1

您可能从 pubsub 中获取原始字符串，而不是 python 对象。在将它们写入 BQ 之前，您可能需要将 json 字符串解析为 python 对象（或字典）。看这个例子：https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/windowed_wordcount.py#L71

无法将值从 pubsub 插入到 GCP 中的大查询

Failing to insert values from pubsub to Big query in GCP

python

google-cloud-pubsub

google-cloud-dataflow

apache-beam