从主题 PubSub 读取数据流并写入 Bigquery（多个表）答案

【问题标题】：Dataflow read from a topic PubSub and write to Bigquery (multiples tables)从主题 PubSub 读取数据流并写入 Bigquery（多个表）
【发布时间】：2020-03-18 02:34:59
【问题描述】：

有人在 Dataflow 中使用过 DynamicDestination，他有一个简单且描述的示例。我厌倦了在 git (https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/master/src/main/java/com/google/cloud/teleport/templates/DLPTextToBigQueryStreaming.java) 中看到示例传送，成为 apache Beam 的新手让我很痛苦。顺便说一句，我需要做的是从 Pubsub 读取消息并通过 Dataflow 作业写入 BigQuery 数据集中的不同目标（表）。我有一个非常适合 Bigquery 表的自定义项目，但 Pubsub 主题将包含来自同一数据集的多个目的地。此外，消息是 JSON 格式，并包含一个带有目标表名称的字段。

这是我最有代表性的代码

TopicToBigQueryOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().as(TopicToBigQueryOptions.class);
Pipeline p = Pipeline.create(options);
p.apply(Constants.READ_PUBSUB, PubsubIO.readStrings().fromSubscription(options.getInputSubscription()))
         .apply(Constants.LINE_TO_CHAMP, new PubSubToTableRowTransform())
         .apply(Constants.WRITE_CHAMPBAN, BigQueryIO.writeTableRows()
                .to(options.getTableStagingFileLines())
                .withSchema(AmplaChangeLogSchema.getTableSchema())
                .withCreateDisposition(CREATE_IF_NEEDED)
             .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));

有什么建议吗？

最好的问候

【问题讨论】：

看看这个example，它可能与当前用例有关。
是否需要将每条pubsub消息内容写在一张表中（可以根据pubsub消息内容变化）？还是需要在多个表中写入相同的 pubsub 消息？
更正每条消息对应一个表，但我有25种不同类型的消息@guillaumeblaquiere
@mk_sta 该示例非常接近我的需要，不幸的是，该示例需要从生产者（在我的情况下为 kafka）设置 tableNameAttr，我无法获得。在我的例子中，我在 JSON Paylod 中有一个名为 TableName 的属性。
这个例子正是你所需要的。这里的表名是根据 PubSub 属性发现的。但是您可以在getTableDestination 函数中添加预处理以提取 PubSub 消息负载，根据需要对其进行解析并获得正确的值！

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

【解决方案1】：

正如我在评论中提到的，原作者 (@Ryan McDowell) 正在解释几乎相同的用户 scenario，使用 GCP Pub/Sub 消息队列中的 JSON 有效负载，执行到 Bigquery 表的动态路由，提取某些来自 Pub/Sub 消息的特定属性中的表名称。

在example的管道中，我们看到getTableDestination()方法，继承自DynamicDestinations类，用于从包含Bigquery表名的消息中提取特定属性（tableNameAttr），最终识别目标对象TableDestination()。

【讨论】：

是的，现在我也需要更新和删除，这更复杂。
我想这与真正的问题无关，考虑创建一个单独的线程。但是，如果您发现上述解决方案有用，那么您可以接受/投票，因此这将有助于进一步的贡献者研究。