google-cloud-dataflow - 从管道中的 GCS 文件名的 PCollection 中读取文件？

Question

我有一个连接到 pub/sub 的流管道，用于发布 GCS 文件的文件名。从那里我想读取每个文件并解析每一行上的事件(这些事件是我最终想要处理的)。

我可以使用 TextIO 吗？当文件名在执行期间定义时，您可以在流管道中使用它吗(而不是使用 TextIO 作为源并且文件名在构造时是已知的)。如果不是，我正在考虑执行以下操作:

从发布/订阅获取主题
ParDo 读取每个文件并获取行
处理文件的行...

在这种情况下，我可以使用 FileBasedReader 或类似的东西来读取文件吗？文件不太大，所以我不需要并行读取单个文件，但我需要读取很多文件。

Answer 1

您可以使用 TextIO.readAll()变换，最近已在 #3443 中添加到 Beam .例如:

PCollection<String> filenames = p.apply(PubsubIO.readStrings()...);
PCollection<String> lines = filenames.apply(TextIO.readAll());

这将读取通过 pubsub 到达的每个文件中的所有行。