当 TextIO 完成写入文件时通知

Question

我正在使用 Google Cloud 中的 DataFlow 处理具有两个管道的场景：

管道A 以流模式运行，基于每小时窗口和一些分片在 Google-Storage 中不断创建文件，如下所示：

data.apply(TextIO.write().to(resource.getCurrentDirectory())
            .withFilenamePolicy(new PerWindowFiles(prefix))
            .withWindowedWrites()
            .withNumShards(42));

管道 B 在批处理模式下工作，定期加载这些文件以进行进一步处理，例如每小时一次。

问题来了：管道B可以从 GS 加载哪些文件？

• 所有这些 -> 可能不是一个好主意，以防 A 没有写完其中一些，我们会得到损坏的文件。

• 基于时间（例如仅加载至少 2 小时前的文件）-> 如果 A 迟到

，也会导致问题

• 在 A 中创建“完成”标志的某种方式告诉 B 哪些文件已完成。

• 当窗口的最终窗格处理完毕时，不知何故会收到通知 -> 还没有找到方法。

我想要第三种方法，但找不到一种方法来确定 TextIO 何时真正完成写入文件而不等待管道完成。

TextIO 的 Writer 不返回另一个 PCollection。一种方法是覆盖FileBasedSink.WriteOperation 的finalize 方法，该方法是在TextIO 中的某处创建的，需要复制整个类并最终构建一个自定义接收器。在我看来，这太过分了。

任何人有更简单的解决方案的想法或经验如何实现这一目标？

Answer 1

TextIO.write() 会将数据写入临时文件，然后以原子方式将每个成功写入的临时文件重命名为其最终位置。您可以安全地使用与管道 B 中的“前缀”匹配的文件，因为临时文件将以与前缀不匹配的方式命名（我们在决定如何命名临时文件时明确考虑了您的用例），因此所有文件管道 B 看到的将是完整的。

另外，我们about to add (link to pull request) 是 TextIO.read() 的一个版本，它以流模式持续摄取新文件；准备就绪后，您可以在管道 B 中使用它。另请参阅 http://s.apache.org/textio-sdf 和链接的 JIRA。