Google Dataflow：从Google Cloud Storage读取未绑定的PCollection

Question

我有一个管道，可将JSON消息从PubSub（未绑定PCollection）流传输到Google Cloud Storage。每个文件应包含多个JSON对象，每行一个。

我想创建另一个管道，该管道应该从此GCS存储桶中读取所有JSON对象，以进行进一步的[[stream-processing。最重要的是，第二个管道应该作为流而不是批处理。意味着我希望它“监听”存储桶并处理写入其中的每个JSON对象。 Unbound PCollection。

有没有办法实现这种行为？

谢谢

Answer 1

流传输过程仅适用于PubSub数据源。但是不用担心，您可以实现自己的管道。

创建a notification subscription on the bucket to publish event on PubSub
• [创建用于侦听PubSub消息的管道。
• 消息到达时，请阅读并获取文件URI
• 使用存储API读取文件并在管道中插入每一行
• 继续每条解码的行的管道。