Flink readFile API如何维护状态？从Savepoint恢复重复读文件是否正常

你观察到的从savepoint恢复后重新处理未修改S3文件的现象，在Flink 1.13.2版本中属于预期行为。

原因说明 #

Flink 1.13及更早版本中，readFile API的PROCESS_CONTINUOUSLY处理模式，没有将已处理文件的元信息（包括文件路径、最后修改时间、已读取偏移量等）持久化到算子状态中。因此当你从savepoint恢复任务时，连续文件读取算子没有历史处理记录，会将目标路径下的所有文件判定为未处理的新文件，触发全量重新读取。

你对运行逻辑的理解是正确的：任务正常运行不重启的前提下，未修改的文件不会被重复处理。但这个判断逻辑只在任务单次运行的内存上下文里生效，没有持久化到可恢复的状态中，因此重启后会失效。

适配场景的解决方案 #

你的场景是导入固定历史启动数据，同时需要保持算子处于RUNNING状态可生成savepoint，可以参考以下方案：

• 方案1：升级Flink版本到1.14及以上。该版本开始官方优化了连续文件读取源的实现，已处理文件的元数据会持久化到状态中，从savepoint恢复后不会重复处理未修改的文件，完全匹配需求。
• 方案2：如果暂时无法升级，可以自行补充去重逻辑：在文件处理算子前增加一层判断，将已经处理完成的文件路径+最后修改时间存入ListState，每次扫描到新文件时先和状态中的记录比对，确认未处理过再往下游发送数据。
• 方案3：如果不需要监控后续新增文件，可以将PROCESS_CONTINUOUSLY模式的扫描间隔调整为一个极大值（比如86400000L即24小时），既可以保持算子RUNNING状态，也能减少不必要的S3扫描请求开销。

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