如何避免Flink PROCESS_CONTINUOUS模式读CSV文件产生重复数据

问题根源 #

你使用的FileSource.PROCESS_CONTINUOUS模式默认不会持久化已处理文件的元数据，作业重启时无历史状态参考，会重新扫描所有匹配路径下的文件并处理，导致重复数据生成。

可落地的解决方案 #

方案1：开启Flink Checkpoint 持久化读取状态（官方推荐） #

• Flink的FileSource原生支持将已处理文件列表、文件读取偏移量等信息作为算子状态持久化到Checkpoint中，只要作业开启Checkpoint，重启时从最近的Checkpoint/Savepoint恢复，就会自动跳过已处理完成的文件。
• Java配置示例：

// 开启Checkpoint，间隔根据业务容错要求调整，示例为10分钟，精准一次模式
env.enableCheckpointing(600000, CheckpointingMode.EXACTLY_ONCE);
// 配置Checkpoint存储到分布式存储路径（如HDFS、对象存储）
env.getCheckpointConfig().setCheckpointStorage("hdfs://flink-checkpoint/your-job-name");

• 注意事项：作业重启时必须指定从最近的成功Checkpoint或Savepoint路径启动，不能无状态启动，否则仍无法读取历史处理状态。

方案2：自定义已处理文件标记逻辑（不依赖Flink状态） #

• 自行维护已处理文件标记库，可存储在Redis、MySQL或分布式存储的专属标记目录下，每完整处理完一个文件后，将文件的唯一标识（文件路径+最后修改时间+文件大小的组合，避免重名文件冲突）写入标记库。
• 自定义文件读取过滤器，每次FileSource扫描到新文件时，先查询标记库确认是否已处理，已处理的文件直接跳过。
• 该方案不受Flink状态保留时间限制，就算清空作业状态重启也不会出现重复处理问题，适合Checkpoint无法长期保留的场景。

方案3：Kafka Sink层幂等兜底 #

• 作为额外的兜底策略，给Kafka Producer开启幂等配置，同时为每条记录生成全局唯一的业务主键作为Kafka消息的Key，同一条数据重复发送时，Kafka会自动覆盖相同Key的历史消息，下游消费侧也可以基于该主键做二次去重。
• Java配置示例：

Properties kafkaProps = new Properties();
// 开启Kafka生产者幂等
kafkaProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-address:9092");
// 其余Kafka配置省略

KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
        .setKafkaProducerConfig(kafkaProps)
        .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                .setTopic("your-business-topic")
                .setKeySerializationSchema(new SimpleStringSchema())
                .setValueSerializationSchema(new SimpleStringSchema())
                .build()
        )
        .build();

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