java - Apache Beam TextIO.ReadAll()，处理丢失的文件名？

Question

我有一个管道步骤 block ，用于从 GCS 存储桶中的特定文件读取数据。代码如下所示:

List<String> filepaths = new ArrayList<String>("filepath1", "filepath2", "filepathMissing");

return pipeline
    .apply("GatherFiles", Create.of(filepaths)).setCoder(StringUtf8Coder.of())
    .apply("GatherFileData", TextIO.readAll())        
    .apply("ApplyCustomDoFn", ParDo.of(new CustomDoFn()))
    .apply("Group", GroupByKey.<String, String>create())
    .apply("AnotherCustomDoFn", ParDo.of(new AnotherCustomDoFn()));

如果 GCS 中缺少其中一个文件路径(例如，缺少“filepathMissing”)，整个管道就会崩溃。我正在考虑在这组管道步骤周围添加一个 try/catch，但我不确定这样做的后果。

我的问题:

1. 使用 TextIO.readAll() 时，try/catch 是对可能丢失的文件进行错误处理的正确方法吗？
2. 使用 try/catch，如果无法从 GCS 找到一个文件，上述整套步骤是否会失败？

如果有任何关于如何执行此操作的具体文档，请将其与您的答案链接起来:)

Answer 1

我最终找到了上面问题的答案。

我必须在 TextIO.readAll() 之后添加代码 .withEmptyMatchTreatment(EmptyMatchTreatment.ALLOW)，如下所示。

List<String> filepaths = new ArrayList<String>("filepath1", "filepath2", "filepathMissing");

return pipeline
    .apply("GatherFiles", Create.of(filepaths)).setCoder(StringUtf8Coder.of())
    .apply("GatherFileData", TextIO.readAll().withEmptyMatchTreatment(EmptyMatchTreatment.ALLOW))        
    .apply("ApplyCustomDoFn", ParDo.of(new CustomDoFn()))
    .apply("Group", GroupByKey.<String, String>create())
    .apply("AnotherCustomDoFn", ParDo.of(new AnotherCustomDoFn()));

此修复的不幸问题是，如果文件丢失，您的管道将在管道步骤中输出一条 INFO 消息，但不会在管道外部抛出任何错误。如果您期望该文件存在，则除非您在数据流中显式检查管道步骤的日志，否则您可能不知道该文件尚未包含在内。