从管道中的 GCS 文件名的 PCollection 中读取文件？答案

【问题标题】：Read files from a PCollection of GCS filenames in Pipeline?从管道中的 GCS 文件名的 PCollection 中读取文件？
【发布时间】：2015-08-28 18:54:06
【问题描述】：

我有一个连接到发布 GCS 文件的文件名的 pub/sub 的流式传输管道。从那里我想读取每个文件并解析出每一行的事件（这些事件是我最终想要处理的）。

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

从 pub/sub 获取主题 ParDo 读取每个文件并获取行处理文件的行...

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

【问题讨论】：

我们即将获得足够的 API 支持来创建高效的实现。请关注issues.apache.org/jira/browse/BEAM-2511TextIO 应该支持读取文件名的 PCollection。
我编辑了答案以反映新的 API。

【解决方案1】：

您可以使用TextIO.readAll() 转换，该转换最近已添加到#3443 中的Beam。例如：

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

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

【讨论】：