Apache Beam Java SDK处理同Bucket多文件时如何定位错误记录所属文件名

默认使用TextIO.read().from(通配路径)的方式批量读取文件时，返回的PCollection仅包含行文本内容，不会携带文件元信息，因此无法直接关联错误行所属文件，可通过先匹配文件元数据、再逐文件读行绑定文件名的方式实现需求，具体步骤如下：

• 替换原有直接读取文本的逻辑，先通过FileIO匹配路径下所有CSV文件，拿到带完整元信息的文件句柄：

PCollection<FileIO.ReadableFile> matchedFiles = pipeline
  .apply("匹配目标路径CSV文件", FileIO.match()
    .filepattern(options.getInputFilePattern()))
  .apply("生成可读文件句柄", FileIO.readMatches());

• 编写逐文件读行的ParDo逻辑，读取每个文件的每一行时，将当前行和所属文件名绑定为KV结构输出：

PCollection<KV<String, String>> sourceDataWithMeta = matchedFiles
  .apply("逐文件读取行并绑定文件名", ParDo.of(new DoFn<FileIO.ReadableFile, KV<String, String>>() {
    @ProcessElement
    public void process(@Element FileIO.ReadableFile file, OutputReceiver<KV<String, String>> out) throws IOException {
      // 仅获取文件名用getFilename()，需要完整GCS路径可替换为file.getMetadata().resourceId().toString()
      String sourceFile = file.getMetadata().resourceId().getFilename();
      try (BufferedReader br = new BufferedReader(Channels.newReader(file.open(), StandardCharsets.UTF_8))) {
        String line;
        // 若需要跳过CSV表头，可加计数器判断第一行直接跳过
        while ((line = br.readLine()) != null) {
          out.output(KV.of(sourceFile, line));
        }
      }
    }
  }));

• 后续做数据类型校验、转换BigQuery记录的逻辑中，处理每条KV数据时如果触发类型不匹配异常，将KV的key（源文件名）和错误行内容一同写入invalidDataTag对应的侧输出集合即可，收集到的错误记录就会自带所属源文件信息。

注意：如果单文件体积较大，上述自行实现的读行逻辑不会自动拆分文件分片，若需要优化大文件读取性能，可使用2.29.0及以上版本Beam内置的TextIO.readFiles()，配置withOutputMetadata(true)参数，可直接读取得到带文件元信息的行记录，无需自行实现读行逻辑。

内容的提问来源于stack exchange，提问作者raj