python - 谷歌云平台 : Pub/Sub to Bigtable

Question

我正在使用 Python 在 Google Cloud Platform 上构建管道。我在 Cloud Pub/Sub 中有我的数据。我想使用 Dataflow 将它存储到 Bigtable 中。到目前为止，我有一些使用 Java 将数据从 Pub/Sub 流式传输到 Bigtable 的示例。

谁能帮我提供一些资源或链接，了解如何使用 Python 中的 Dataflow 将我的数据从 Pub/Sub 流式传输到 Bigtable？

Answer 1

山姆，我不确定我们如何在 Python 中执行此操作。但我是用 Java 完成的。希望这个想法可以帮助您解决问题。

执行此操作时应牢记的步骤是

1. 从 Pub/Sub 读取，将流设置为 true

   PubsubIO.readStrings().fromTopic(PUBSUB_SUBSCRIPTION))
   

2. 用常量键对集合进行分组

   PCollection<KV<String, String>> keyedStream = streamData
           .apply(WithKeys.of(new SerializableFunction<String, String>() {
               /**
                * serial version id
                */
               private static final long serialVersionUID = 1L;
   
               public String apply(String s) {
                   return CONSTANT_KEY;
               }
           }));
   

3. 创建转换以将 PCollection 中的元素划分到窗口中，并在输出这些元素时触发控件。

   Window.<String>into(new GlobalWindows())
           .triggering(Repeatedly
                   .forever(AfterProcessingTime
                           .pastFirstElementInPane()
                           .plusDelayOf(Duration.standardSeconds(30))
                       )).withAllowedLateness(Duration.standardDays(1)).discardingFiredPanes()
   

4. 将PCollection写入Bigtable。