谷歌云平台：发布/订阅到 Bigtable

Question

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

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

Answer 1

山姆， 我不确定我们如何在 Python 中做到这一点。但我已经在 J​​ava 中做到了这一点。希望这个想法可以帮助您解决问题。

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

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。