如何将watchfornewfiles与GCS源存储桶一起用于数据流？

Question

参考项目：在Apache Beam中观察与文件夹匹配的新文件

您可以将其用于简单的用例吗？我的用例是我让用户将数据上传到Cloud Storage-

我想做的是保持管道在流式模式下运行，一旦文件上传到Cloud Storage，它就会通过管道进行处理。watchfornewfile可以做到这一点吗？

我编写的代码如下:

p.apply(TextIO.read().from("<bucketname>")         
    .watchForNewFiles(
        // Check for new files every 30 seconds         
        Duration.standardSeconds(30),                      
        // Never stop checking for new files
        Watch.Growth.<String>never()));

没有任何内容被转发到Big Query，但管道显示它正在流式传输。

Answer 1

你可以在这里使用Google云存储触发器:https://Cloud . Google . com/functions/docs/calling/Storage # functions-calling-Storage-python

这些触发器使用类似于Cloud Pub/Sub的Cloud Functions，如果对象是：创建/删除/存档/或元数据更改，则会在对象上触发。

这些事件是使用来自云存储的发布/订阅通知发送的，但注意不要在同一个bucket上设置许多函数，因为存在一些通知限制。

此外，在文档的结尾有一个示例实现的链接。