从主题PubSub读取数据流并写入Bigquery（multiples tables）

Question

有一个在Dataflow中使用过DynamicDestination的人，他有一个简单的描述示例。在git（https://github.com/googleCloudPlatform/dataflowTemplates/blob/master/src/main/Java/com/google/cloud/teleport/templates/dlpTextToBigQueryStreaming.)中看到示例teleport已经让我感到厌烦了，作为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));

有什么建议吗？

最诚挚的问候

Answer 1

正如我在评论中提到的，原作者(@Ryan McDowell)解释了非常相同的用户场景，使用GCP pub/sub消息队列中的JSON有效负载，执行到Bigquery表的动态路由，从pub/sub消息中提取特定属性中的特定表名。

在示例的管道中，我们看到从DynamicDestinations类继承的GetTableDestination()方法，该方法用于从包含Bigquery表名的消息中提取特定属性(TableNameAttr)，最后标识目标对象TableDestination()。