使用Apache Beam FileIO.matchAll()时如何避免丢弃原结构体其余字段

完全可以在保留全部元数据的前提下完成文件匹配，这里提供两种常用实现方案：

方案一：自定义DoFn调用底层文件匹配API（推荐，无额外Shuffle开销） #

你不需要强制使用FileIO.matchAll()，可以直接在自定义DoFn中调用Beam底层的FileSystems匹配API，匹配逻辑和FileIO.matchAll()完全一致，同时可以保留原结构体的所有字段：

import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.fs.MatchResult;

// 自定义输出结构，同时保留原元数据和匹配结果
public class MatchedFileWithMeta {
    // 原始结构体的所有字段都保留
    public MyStruct originalInfo;
    // 当前pattern匹配到的所有文件元数据
    public List<MatchResult.Metadata> matchedFiles;
}

PCollection<MatchedFileWithMeta> result = myStructsPCollection
    .apply(ParDo.of(new DoFn<MyStruct, MatchedFileWithMeta>() {
        @ProcessElement
        public void process(@Element MyStruct input, OutputReceiver<MatchedFileWithMeta> out) throws IOException {
            // 执行文件匹配，逻辑和FileIO.matchAll完全一致
            List<MatchResult> matchResults = FileSystems.match(Collections.singletonList(input.filePattern));
            List<MatchResult.Metadata> files = new ArrayList<>();
            for (MatchResult mr : matchResults) {
                files.addAll(mr.metadata());
            }
            // 组装输出结果，原始数据完全保留
            MatchedFileWithMeta output = new MatchedFileWithMeta();
            output.originalInfo = input;
            output.matchedFiles = files;
            out.output(output);
        }
    }));

方案二：通过关联操作配合FileIO.matchAll()使用 #

如果你必须使用FileIO.matchAll()的封装能力（比如需要配置匹配参数、连续调用FileIO.readMatches()等），可以通过按文件模式关联的方式把原始元数据加回来：

// 第一步：将原始数据按filePattern分组，生成KV
PCollection<KV<String, MyStruct>> patternToStruct = myStructsPCollection
    .apply(MapElements.into(TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptor.of(MyStruct.class)))
        .via(struct -> KV.of(struct.filePattern, struct)));

// 第二步：提取filePattern做匹配
PCollection<MatchResult> matchResults = myStructsPCollection
    .apply(MapElements.into(TypeDescriptors.strings()).via(MyStruct::getFilePattern))
    .apply(FileIO.matchAll());

// 第三步：将匹配结果也转成以filePattern为key的KV
PCollection<KV<String, MatchResult>> patternToMatch = matchResults
    .apply(MapElements.into(TypeDescriptors.kvs(TypeDescriptors.strings(), TypeDescriptor.of(MatchResult.class)))
        .via(mr -> KV.of(mr.inputPattern(), mr)));

// 第四步：按key关联两个PCollection，就能同时拿到原始数据和匹配结果
// 后续可以根据需要处理关联后的结果

方案选择建议 #

• 单pattern对应唯一MyStruct的场景优先选方案一，不需要额外的Shuffle开销，实现更简单
• 多个MyStruct共享同一个filePattern的场景可以选方案二，避免重复执行匹配逻辑

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