Kafka到Flink流任务重启后丢数

产线设备事件先进 Kafka，再由 Flink 按 5 分钟窗口统计产量和停机次数，最后写进下游宽表。平时跑得好好的，某天任务重启一次，复核时发现某个窗口的产量少了几十件，另一天重启后，同一个窗口又多出了一批重复计数。

这种时丢时重的情况，第一反应往往是怀疑 Kafka 位点乱了，或者窗口触发器写错了。但在翻消费组位点之前，有一个更前置的问题值得先确认：这个任务到底开没开 checkpoint。Apache Flink 官方文档写得很直接，checkpoint 默认是关闭的，不调用 enableCheckpointing 就等于没有容灾快照，重启后的行为根本不是“从断点续跑”。

一、没开 checkpoint，重启后从哪开始并不由 Flink 记住

Checkpoint 是 Flink 对整个流图的一次一致性快照，里面同时包括算子的状态和 source 的读取位置。窗口里已经攒了一半的聚合结果、Kafka 每个分区的消费位点，都要靠这次快照才能在故障后原样恢复。没有快照，任务重启后窗口的中间状态就没了，source 也只能按启动时配好的起始位点重新开始：配成从最早开始就重放，配成从最新开始就丢弃重启前还没处理完的那一段。

更隐蔽的一点是，即使开了 checkpoint，存储位置没配对也等于白开。官方文档指出，默认情况下 checkpoint 存在 JobManager 的堆内存里。JobManager 本身挂掉或者整个集群重启，这份快照也跟着消失，谈不上恢复。所以“开了 checkpoint”这句话要拆成两句才算数：快照有没有定时生成，快照有没有落到 JobManager 之外的持久存储上。

上线前先到 Flink Web UI 的 Checkpoints 一栏看一眼：有没有 Completed 的快照、最近一次成功是什么时候。如果这一栏长期是空的，后面所有关于丢数、重复的排查都是空中楼阁。

二、一份能直接用的最小配置

下面这段 DataStream API 配置可以作为起点，checkpoint 间隔 60 秒、一致性模式显式写成 exactly-once、快照落到分布式文件系统、取消任务后仍保留外部化快照：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 每 60 秒做一次 checkpoint，间隔单位是毫秒
env.enableCheckpointing(60000);

// exactly-once 本就是默认模式，这里显式写出来，避免后面被人误改
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 一次 checkpoint 超过 60 秒未完成就丢弃，避免旧快照一直占着
env.getCheckpointConfig().setCheckpointTimeout(60000);
// 两次 checkpoint 之间至少留 500 毫秒给正常处理，避免一直忙着做快照
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
// 最多允许 2 次 checkpoint 连续失败，超过就让任务失败而不是悄悄裸奔
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);
// 同一时间只做一个 checkpoint，避免快照排队积压
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
// 快照落到分布式文件系统，不要留在 JobManager 堆内存里
env.getCheckpointConfig().setCheckpointStorage("hdfs:///flink/checkpoints/device_job");
// 任务取消后仍保留快照，升级、迁移时还能从它恢复
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

间隔怎么定，我的经验起点是 60 秒：间隔越短，故障后需要重放的数据越少，但做快照本身的开销越大；状态大、写入慢的任务可以先放宽到几分钟，跑稳后再往下压，不要一上来就设成几秒。窗口状态很大的任务，状态后端建议用 RocksDB 而不是默认的堆内后端，这是经验性选择，具体要看状态量和 TaskManager 的内存余量。

另外要区分 checkpoint 和 savepoint 的分工：checkpoint 是系统自动做的、用来故障恢复的快照，默认在任务取消时会被删除；上面把外部化清理策略设成取消后保留，就是为了让计划内的停机升级也能从快照恢复。临时手动触发一次 savepoint 再升级，同样能达到这个目的，关键是升级前手里必须有一个明确可用的恢复点，而不是默认“任务取消了快照还在”。

三、开了 checkpoint，不等于端到端 exactly-once

这是最容易被一句话带偏的地方。Flink 文档里对 exactly-once 的界定是：每个事件对 Flink 内部管理状态的影响恰好生效一次，而不是每个事件被下游系统恰好写了一次。快照能保证窗口聚合结果本身不丢不重，但数据写出下游的那一步，还要看 sink 支不支持事务或者幂等写入。

对照着检查 sink 就行：Kafka Sink 的 exactly-once 模式会把写入放进事务、随 checkpoint 提交，恢复时不会重复提交；普通的 JDBC 写入或者 HTTP 调用则没有这个机制，故障后从快照位点重放的那一段数据会被再写一遍，除非写入本身按业务主键做 upsert、或者下游按主键去重。窗口计数这种场景，主键一般就是窗口时间加设备编号，写下游时按这个主键覆盖写入，比事后对账删重复行要省事得多。

还有一个前提容易被忽略：checkpoint 恢复依赖 source 能重放。Kafka 这种能按位点回放的 source 才撑得起这套机制，如果上游是只能消费一次、不支持回放的队列，或者 Kafka 的数据保留期已经短于故障发现的时间，快照里记的位点就算还在，数据也已经取不回来了。保留期和故障发现时长要一起算，不要只算 checkpoint 间隔。

四、上线前，做一次真实的重启演练

配置写完不要直接当成已验证，跑一遍下面的演练再上线：

1. 任务正常跑起来后，记下某一个完整窗口的产量和停机次数；
2. 在窗口还没结束时重启一次任务，让它从最近的 checkpoint 恢复；
3. 等窗口闭合后，再对一次数：窗口总量应该和重启前一致，既不多也不少；
4. 同时确认 Flink Web UI 里有 Completed 的 checkpoint，并且快照目录里确实落了文件。

这一步如果发现重复，大概率是 sink 没有幂等；如果发现丢失，先看恢复用的是不是最近的 checkpoint，再看 source 的起始位点和保留期。演练花不了多少时间，却能把“默认关闭”和“存储位置不对”这类问题提前暴露出来，而不是等生产故障时再复盘。

小结

记住三个默认值和一个界限就够了：checkpoint 默认是关闭的，不调用 enableCheckpointing 就没有快照；快照默认存在 JobManager 堆内存里，不配持久存储等于没容灾；任务取消时 checkpoint 默认会被删除，计划内升级要靠外部化保留或 savepoint。而 exactly-once 的界限是 Flink 内部状态的影响恰好一次，写出下游是否重复，还要看 sink 是否支持事务或幂等写入。先把这几处对上，再去查 Kafka 位点和窗口逻辑，排查方向才不会一开始就走偏。