如何在Dataflow批处理作业中识别已处理文件？

这个场景我碰到过不少，Dataflow的批处理默认是无状态的，每次运行都会扫描所有匹配通配符的文件，所以重复读取是正常行为。要实现“只处理未处理过的文件”，可以试试下面这几个方案：

1. 维护已处理文件清单 #

这是最直接的思路：把已经处理完成的文件名记录下来，每次作业启动先加载这个清单，过滤掉已经处理过的文件。

具体步骤： #

• 选择一个持久化存储保存清单：比如GCS的文本文件、BigQuery表或者Cloud Firestore。这里以GCS文本文件为例，每行存储一个已处理的文件路径。
• 先读取清单到Dataflow中，再获取所有匹配通配符的文件元数据，两者对比后只保留未处理的文件。
• 处理完成后，把本次处理的文件名追加到清单里（注意GCS的append模式只对特定存储类生效，也可以先读取旧清单，合并新文件名后再覆盖写入）。

示例代码：

// 1. 读取已处理文件清单
PCollection<String> processedFiles = pipeline.apply(
    "Read processed files list",
    TextIO.read().from("gs://bucketName/processed_files.txt")
);

// 2. 获取所有匹配的文件元数据
PCollection<MatchResult.Metadata> allFiles = pipeline.apply(
    "Match all target files",
    FileIO.match().filepattern("gs://bucketName/TrafficData*.txt")
);

// 3. 过滤出未处理的文件（实际中建议用广播变量或Join操作实现集合对比）
PCollection<MatchResult.Metadata> unprocessedFiles = allFiles.apply(
    "Filter unprocessed files",
    Filter.by(metadata -> {
        String filePath = metadata.resourceId().toString();
        return !processedFiles.contains(filePath);
    })
);

// 4. 读取未处理文件的内容
PCollection<String> fileContent = unprocessedFiles.apply(
    "Read unprocessed file content",
    FileIO.readMatches().withCompression(Compression.AUTO)
);

// 5. 处理完成后，将本次处理的文件名写入清单
unprocessedFiles.apply(
    "Extract file paths",
    MapElements.via((MatchResult.Metadata meta) -> meta.resourceId().toString())
).apply(
    "Append to processed list",
    TextIO.write().to("gs://bucketName/processed_files.txt")
        .withAppend() // 启用追加模式
        .withSuffix(".txt")
);

2. 基于文件元数据过滤（修改时间/命名规则） #

如果你的文件有明确的命名规则（比如按日期命名：TrafficData_20240520.txt）或者不会被修改，那可以通过时间范围来过滤文件，避免重复读取。

两种子方案： #

• 按命名规则过滤：作业启动时传入时间参数（比如昨天的日期），构造精确的文件路径，比如gs://bucketName/TrafficData_${yesterday}.txt，直接读取指定文件。
• 按修改时间过滤：用FileIO.match()的时间范围配置，只处理上次作业运行之后新增/修改的文件：

// 获取上次作业运行的时间戳（可以存在GCS/BigQuery中，作业启动时读取）
long lastRunTimestamp = ...;

PCollection<MatchResult.Metadata> recentFiles = pipeline.apply(
    "Match files modified after last run",
    FileIO.match().filepattern("gs://bucketName/TrafficData*.txt")
        .withMatchConfiguration(MatchConfiguration.create()
            .withModifiedTimeRange(
                Instant.ofEpochMilli(lastRunTimestamp),
                Instant.now()
            ))
);

3. 给已处理文件打GCS对象标签 #

你可以给处理完成的文件打上自定义标签（比如processed: true），然后在读取文件前，先调用GCS API筛选出没有该标签的文件。

注意点： #

• 默认的FileIO.match()不支持标签过滤，需要自己实现文件筛选逻辑：先列出所有匹配通配符的文件，再逐个检查标签，过滤出未标记的文件。
• 这个方案适合文件数量不多的场景，否则频繁调用GCS API可能会有性能开销。

4. 利用外部存储记录处理状态 #

如果你的作业是定期运行的，可以把处理状态（比如最后处理的文件、最后处理的时间戳）存在BigQuery或者Cloud Firestore中：

• 作业启动时读取这个状态值
• 基于状态值过滤文件
• 处理完成后更新状态值为本次作业的结束时间/处理的最后一个文件

这种方式的好处是状态存储更可靠，支持并发读写控制，适合复杂的业务场景。

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