文章/答案/技术大牛

发布

问从发布/订阅流到BigQuery
EN

Stack Overflow用户

提问于 2017-09-07 05:08:45

回答 1查看 2K关注 0票数 5

我正在尝试使用python数据流将一些数据从google PubSub传输到BigQuery中。出于测试目的，我已经将以下代码https://github.com/GoogleCloudPlatform/DataflowSDK-examples/blob/master/python/dataflow_examples/cookbook/bigquery_schema.py修改为流管道，方法是设置

options.view_as(StandardOptions).streaming = True

因此，我将record_ids管道更改为从发布/订阅读取

# ADDED THIS
lines = p | 'Read PubSub' >> beam.io.ReadStringsFromPubSub(INPUT_TOPIC) | beam.WindowInto(window.FixedWindows(15))
# CHANGED THIS # record_ids = p | 'CreateIDs' >> beam.Create(['1', '2', '3', '4', '5'])
record_ids = lines | 'Split' >> (beam.FlatMap(split_fn).with_output_types(unicode))
records = record_ids | 'CreateRecords' >> beam.Map(create_random_record)
records | 'Write' >> beam.io.Write(
    beam.io.BigQuerySink(
        OUTPUT,
        schema=table_schema,
        create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
        write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE))

注意:我已经被谷歌列入了运行代码的白名单( alpha)

现在，当我尝试它的时候，我有一个错误

工作流失败。原因：(f215df7c8fcdbb00)：未知流接收器: bigquery

你可以在这里找到完整的代码：https://github.com/marcorigodanzo/gcp_streaming_test/blob/master/my_bigquery_schema.py

我认为这与流水线现在是流类型有关，有人能告诉我如何在流水线中写bigQuery吗？

python

google-bigquery

google-cloud-platform

google-cloud-dataflow

google-cloud-pubsub

回答 1

Stack Overflow用户

发布于 2017-09-08 10:40:52

Beam Python不支持从流式管道写入BigQuery。现在，你需要使用Beam -你可以分别使用PubsubIO.readStrings()和BigQueryIO.writeTableRows()。

票数 2

页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持

原文链接：

https://stackoverflow.com/questions/46084361

复制

相似问题

问从发布/订阅流到BigQuery
EN

回答 1

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问从发布/订阅流到BigQueryEN

回答 1

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问从发布/订阅流到BigQuery
EN