★ FEATURED ARTICLE 做流计算的人迟早会被一个问题逼到墙角作业跑了两周状态积了几十GB某个节点突然挂了怎么恢复才能做到既不停太久、又不丢数据、还不重复影响下游Flink给出的答案是定期Checkpoint而这套机制背后的分布式快照主干正是经典的Chandy-Lamport算法。这篇文章不打算把官方文档复述一遍而是从算法原理到生产调优把Flink Checkpoint这条线彻底捋清楚。阅读时你会发现Flink对Chandy-Lamport不是照搬而是做了大量面向流式计算场景的改造理解这些改造动机比记住几个参数名有用得多。1. Checkpoint到底在解决什么问题流的“记忆”与灾难恢复1.1 状态流计算的命根子流计算里状态是用来记住“到目前为止我算到哪了”的。最典型的例子是五分钟滚动窗口的PV统计数据来一条加一条窗口不关闭之前累计数必须持续存在再比如去重计数你要记住所有见过的用户ID还有风控规则某个账号过去一分钟的点击频次本质上也是一份状态。没有状态流计算和普通的管道透传没区别所有的聚合、关联、窗口计算都无从谈起。但状态恰恰也是最麻烦的东西。数据在流里冲状态在内存里不断更新一旦进程崩了这部分记忆就跟着丢了。批处理的世界里任务死了重跑一遍就行最多慢一点流计算是7x24小时在线的跑了两周的业务你说重跑就得把两周的数据再吃进来一次下游还要忍受重复推送这在实时链路里基本上是不可接受的。所以流处理框架必须回答一个问题怎么把活的、不断更新的状态在不停机的前提下周期性地固化下来1.2 恢复的两种思路至少一次与精确一次先想个朴素方案每来一条数据就把整个状态存一次。这条路走不通状态动不动GB级每条都存性能直接崩。再退一步定期存比如每三十秒存一次。那么故障后最多丢失三十秒的增量数据这叫“至少一次”语义意思是不保证不丢只保证不重复花时间去算已经算过的部分。很多实时场景其实能接受丢几秒但真正被下游依赖的精确计算场景比如金额统计、对账报表丢一个数都麻烦。Flink想要的是“精确一次”状态恢复到某个一致的、完整的时间点既不丢数据也不重复计算而且还要保证分布式环境下多个节点之间的状态彼此协调。这就是Checkpoint的核心价值。注意这里的“精确一次”是流处理内部语义真正落到外部系统比如把结果写进数据库或消息队列光靠Checkpoint还不够后面讲到端到端一致性的时候我会专门展开。1.3 为什么在分布式流里做快照这么难单机进程做快照很简单暂停、复制内存、恢复。分布式流作业里的难处在于作业是分散在几十上百个并行子任务里的每个子任务有自己的状态子任务之间通过网络传输数据传输过程当中还有大量“在途”数据也就是上游已经发出、下游还没处理到的记录。如果每个节点各自随便挑个时间点存一份状态拼起来的全局快照多半是不一致的这条数据可能已经被下游加了两次或者上游的状态里还留有这条数据而下游已经把它算过了。Chandy-Lamport算法解决的就是这个问题在分布式系统里不借助全局时钟不需要暂停整个系统也能拿到一份一致的全局快照。Flink的Checkpoint本质上就是把这个算法搬到了流计算场景里然后针对流的特殊性做了工程化改造。2. Chandy-Lamport算法核心思想一条Marker走天下2.1 算法的前提与目标这个算法由两位研究者在1985年前后提出解决的问题听起来朴素却很本质一个分布式系统里每个节点各自有状态节点之间的通信链路上还可能飘着消息怎么在不停止业务的条件下得到一份“一致”的全局状态所谓一致指的是如果一条消息的接收者已经把它的效果算进去了那么这条消息就不能再出现在发送者的状态里反过来如果发送者已经把它发出去了而接收者还没处理那它必须被记录在这次快照当中。算法的前提是两条通信信道是可靠的、有序的。可靠意味着消息不会丢失有序意味着某个节点从同一条链路上收到的消息先后顺序和发送方发出的顺序完全一致。这两条对后来Flink的Barrier对齐设计有直接关系。2.2 核心机制记录本地状态加标记消息算法的操作过程不复杂。整个系统里会选出一个发起节点由它发起一次快照发起节点先把自己的本地状态存一份然后往自己所有的出站链路上发送一条特殊的“标记消息”相当于告诉对端我这边已经记录完了。系统里的其他节点在第一次收到标记消息的时候做两件事先把当前本地状态记录下来然后把这个标记消息继续转发给所有出站链路让它一路扩散到全网。节点需要记录每条入站链路的状态。具体规则是如果一条链路上在收到标记消息之前还有没有处理完的消息这些消息要被完整记录下来因为它们在快照发生的瞬间处于“在途”状态在标记消息之后到达的消息则属于“下一阶段”不用记。用生活里的事情来类比一整条流水线上的工位每个工位配一个传送带。领导喊“现在停产盘点”第一道工序停下来记录自己手头的半成品同时往传送带上放了一个红色标牌。后面每个工位看到标牌传过来先停下记录自己的工作台上的半成品再把标牌传给下一个工位。而两个工位之间传送带上已经运出来、但还没落到下一个工位手里的半成品也要单独数一遍这部分就是“在途消息”。所有工位记录完成的时间点可能都不一样但合起来这份数据是一致且完整的既没有重复计算哪件半成品也没有哪件半成品被凭空丢掉。2.3 这张快照为什么是一致的一致的割集把时间想象成一条轴每个工位记录状态的那一刻在时间轴上划一刀。由于没有全局时钟这一刀在每台机器上的位置不一样把这些点连起来就形成一条“割线”。如果不做任何协同割线是歪歪扭扭的可能正好切在一条消息传播的中间发送方已经记完、做了“已发送”标记接收方却还没把它算进去这条消息就从快照里蒸发了。Chandy-Lamport用标记消息这件事妙就妙在它把“消息与快照的关系”在逻辑上锁死了发起节点发Marker前发出的业务消息必然在接收节点第一次收到Marker之前到达因为通信有序所以接收节点会把它们记进链路状态而接收节点收到Marker后就已经冻结了状态再到的消息属于后续时间片。整条割线上既没有消息被算丢也没有消息被重复算这就是算法的正确性所在。理解这一层后面看Flink的Barrier对齐就全是顺理成章。3. Flink对Chandy-Lamport的工程改造从Marker到Barrier3.1 Barrier为流式作业改造过的MarkerFlink直接把Chandy-Lamport的标记消息翻译成了自己的术语Checkpoint Barrier。Barrier是一个带着编号的轻量级事件由JobManager触发从数据源算子注入到每条输入流里然后顺着数据流一路向下游传递。它和普通数据一样走网络和算子处理链但不参与业务计算纯粹是个路标告诉你“属于第N次Checkpoint的数据在这里截止了”。为什么不能照搬原始的Marker机制因为流计算作业和一般的分布式系统有个重要区别数据不是沿着固定链路传递的而是经过各种算子重分区、聚合、广播的。下游一个并行子任务可能同时从上游好几个并行子任务收数据每条连接上都会来Barrier。原始算法没有考虑这种多对多的复杂拓扑Flink需要对Barrier的汇合过程做额外的协调这就是接下来要说的对齐。3.2 Barrier对齐精确一次的钥匙在一个并行度大于1的作业里比如上游6个并行度、下游6个并行度下游某个子任务会同时收到来自6条输入流的Barrier。问题来了这6个Barrier几乎不可能同时到达因为每条路径上的数据量、网络延迟都不一样。Flink的做法是对齐当下游算子收到第一条Barrier时不再处理这个输入通道上后续的数据而是把它们先堵在缓冲区里同时继续处理其他还没到Barrier的通道上的数据。直到所有输入通道的Barrier都到齐了才把当前的本地状态做一次快照然后放开缓冲区让阻塞的数据继续流过。换句话说状态保存的是“所有Barrier都到了”这个整齐时刻的状态之前已经处理的数据都算在本次快照里之后的数据都算下一次。这样做每条数据在快照边界上的归属就被厘清了一次。这个过程可以理解为把“每台机器各自任选时刻”改成“所有机器等一条水平的对齐线”。代价是延迟和背压——快的那条通道要等慢的那条这段时间数据被积压算子往下游发送的速度也会被拖慢。在背压严重的作业里对齐等待时间可能超过Checkpoint的完成时限这是生产环境最常见的故障之一。为此Flink保留了一个降级开关如果不需要严格精确一次把语义切成至少一次Source端Checkpoint做完就放行不用等所有Barrier到齐全程不产生对齐等待恢复时最多丢一些尾部数据。3.3 未对齐Checkpoint用空间换时间的激进方案对齐造成的延迟在带宽紧张、消息积压严重的场景里非常难受。Flink后来引入了未对齐CheckpointUnaligned Checkpoint不等Barrier对齐每个算子收到属于自己的Barrier就立即把本地状态存起来同时把输入缓冲区里所有尚未处理的消息连同它们所属的通道信息一并写入快照。代价相当直白快照的体积会明显变大因为那些在途数据被完整打包进去了恢复的时候算子要从快照里把在途数据重新灌进缓冲区然后按顺序继续处理。好处是Checkpoint的完成时间不再被对齐等待绑架即使下游严重背压也能快速完成一次快照。我的建议是只在共享带宽紧张、对齐延迟持续高企的作业上用而且要监控快照大小的变化如果体积大到拖慢存储和恢复宁可回去对齐。快照再快大了一圈也是个隐患。这也是理解Chandy-Lamport算法很有价值的一个地方原始算法本来就要求记录在途链路状态未对齐Checkpoint某种程度上更接近原始算法对齐反而是在流计算场景里加的优化。3.4 端到端精确一次两阶段提交的补完严格说Checkpoint只保证Flink作业内部的状态和计算结果是精确一次的。结果要写进Kafka、数据库、HDFS如果外部系统写入动作不配合故障恢复后还是可能重复写。比如结果已经写进MySQL了但对应作业的Checkpoint还没完成作业挂了重启从上一个Checkpoint恢复就会把这条结果再算一次、再写一遍下游看到的对账数字就会多一笔。Flink的做法是两阶段提交Sink算子先把写入动作“预提交”例如把数据写进外部系统的临时区或者标记为未确认等到这次Checkpoint完成JobManager通知Sink提交第一阶段的事务如果作业在中间挂了恢复时会根据Checkpoint里保存的事务信息做回滚或重放。Kafka的Sink能配合事务API做到这一点因为Kafka本身提供事务性写入。而没有事务支持的外部系统只能靠幂等写入或者WAL预写日志来兜底。所以聊端到端一致性时我一直跟人说Process的精确有一个前提它解决的是发生在Flink内部的语义别把外部系统自动当成一样可靠设计时就要想好Sink侧怎么配合。4. Checkpoint的完整生命周期从触发到恢复的全链路4.1 三种状态存储内存、文件系统与RocksDB状态是Checkpoint的内容而状态本身放在哪里由State Backend决定。Flink有几种可选堆内存后端状态直接放在JVM堆里读写快但容量受堆大小限制Checkpoint时要把状态序列化到外部存储堆大的话Full GC会频繁。文件系统后端状态对象放在堆内Checkpoint写入配置的分布式文件系统适合状态量中等、希望堆上操作更简单的作业。RocksDB后端状态放在内嵌的RocksDB里实际上是基于本地磁盘的KV存储状态可以做得很大几乎不占JVM堆但读写要走序列化性能比纯堆内低一个量级。选型的核心判断标准就一条状态量级。几个GB以内、对性能要求极高的作业用文件系统后端合适几十GB甚至上百GB、希望稳定跑不爆内存的作业老老实实上RocksDB。RocksDB还支持增量Checkpoint只把上次之后变更过的SST文件上传在状态大、变更比例小的场景下Checkpoint时间能省一大截。4.2 一次Checkpoint的完整流转把整个流程串起来看JobManager里的Checkpoint Coordinator按周期向所有Source算子发送Barrier创建的指令同时生成一个全局唯一的Checkpoint编号。Source算子收到指令后把自己的状态保存一份把Barrier随着数据流注入到下游。Barrier沿图传递每个算子按前面说的对齐或未对齐逻辑完成自己的状态快照快照写进配置好的存储。每个算子完成快照后向Coordinator上报完成消息。Coordinator收到所有算子的确认把这次Checkpoint标记为Completed。任何一个算子失败、超时或者上报失败整次Checkpoint都会被打上Failed标记。作业正常运行不等待Coordinator继续按周期触发下一次Checkpoint。这里有个容易被忽略的细节一个正在做快照的算子它的快照只是一个指向状态的引用或者一份不阻塞的异步拷贝。为了不让快照期间新的数据更新污染快照内容Flink作业里每个算子会先把状态持有一个屏障要么用CopyOnWrite要么用RocksDB的快照句柄保证拍照的那一刻冻结了一个一致的视图。我们自己调优时一定要心里有数异步快照期间如果状态更新频繁对存储IO的瞬时压力是很大的。4.3 恢复从快照里拉回一条平行世界作业挂了之后Flink按照配置的重启策略拉起整个作业图每个算子从最近一次Completed的Checkpoint去读取自己的状态。这份状态里包含两部分算子的键控状态Keyed State和算子状态Operator State前者是按Key分区的状态比如每个用户ID的计数器后者是整个算子共享的状态比如分区分配信息。恢复时还有一个数据重放的细节Checkpoint对应的是某个时刻的一致性截断这个时刻之后的数据需要从Source重新读一遍。Kafka等Source通过保存消费位点来实现这一点Checkpoint里记录了每个分区的消费位点恢复时从位点接着读刚才漏掉的数据就重新进了一遍计算。因为状态已经回滚到了位点对应的一致时刻重放的数据和恢复后的状态是吻合的不会出现重复累加。这就是为什么Sink如果不在事务上配合这层“吻合”会破功——外部系统的写入没法跟着一起回滚。5. 生产环境Checkpoint配置与调优实战5.1 必须吃透的几个核心参数先给一份生产上最常见的参数清单再逐个说怎么写、为什么这么写。参数含义我常用的倾向checkpoint.intervalCheckpoint触发周期30秒到5分钟看业务恢复时效要求checkpoint.timeout单次Checkpoint超时上限默认10分钟生产上设1~3分钟更稳妥checkpoint.min-pause两次Checkpoint的最短间隔保证至少数十秒防止连续频繁触发checkpoint.max-concurrent允许同时进行的最大Checkpoint数生产保持1降低资源竞争checkpoint.unaligned是否启用未对齐Checkpoint默认关按需开rocksdb.incrementalRocksDB增量Checkpoint开关大状态强烈建议开设置interval时不能拍脑袋。Checkpoint周期越短故障恢复时丢的数据越少但快照本身也需要时间和资源。一个经验是interval设成累积状态快照耗时的3倍以上。比如你的作业完成一次Checkpoint平均要15秒那interval设45秒以上才比较健康。有些作业要求秒级恢复那你先把State Backend换轻量级别靠狂压interval硬撑。5.2 一次真实调优经历被背压拖垮的对齐记忆比较深的一次某实时聚合作业用的是Kafka到Flink再到Kafka的链路状态大约10GB落在RocksDB里。上线后监控里Checkpoint频繁失败日志报超时。排查思路是这样走的先看Checkpoint的耗时拆分发现瓶颈几乎全在Barrier对齐阶段而不是状态快照本身。接着看反压监测发现Sink到下游Kafka的写入经常触发背压导致上游堆积Barrier跟着被堵在后面。第一反应是想换未对齐Checkpoint但考虑到状态大在途数据打包体积会很明显没有立刻下手。我们先把Sink的batch.size调大、linger.ms调大让写入更宽松背压缓解了不少再把interval从30秒调整到1分钟让每次Checkpoint之间有更充分的排空窗口失败率立刻降下来了。这次经历给了一个很重要的教训Checkpoint失败往往不是Checkpoint本身的配置问题而是整个数据链路健康度的镜子。你先看反压、看GC、看存储IO再反过来调Checkpoint参数顺序千万别搞反。单纯把timeout调大、把重试次数调高只是把肾虚的病给掩盖了。5.3 监控指标怎么盯Flink Web UI和Metrics系统里有几个指标值得一直盯着Checkpoint完成的耗时曲线重点看对齐耗时和快照上传耗时的比例。若对齐耗时长期占比过高说明Barrier被数据挤在后面重点排查上游吞吐和背压。完成与失败的Checkpoint数量统计失败率升高通常意味着存储、网络或GC出问题。RocksDB的Compaction指标RocksDB在做Compaction时IO占满会明显拖慢快照写入这种情况可以考虑单独给RocksDB挂独立磁盘。上游Kafka消费位点的Lag。这能间接反映整个链路是否健康——如果消费位点一路高升Barrier的推进也会被连累。另外尽量把监控聚合做得粗一点不用精确到每个子任务先看Job整体趋势再逐层下钻。我见过不少团队在指标面板上堆了上百个曲线真正告警时谁也说不清先看哪个。盯核心三张图Checkpoint耗时折线、完成失败计数、反压比例。6. 常见问题与排查技巧实录6.1 Checkpoint一直超时这是遇见概率最高的问题。先别急着加timeout按下面的顺序排查看Barrier对齐耗时如果占了总耗时的大头甚至反复触发背压优先解决链路瓶颈比如调下游写入参数、加并行度。看状态量。如果状态膨胀明显想想是不是窗口没有清理、Key设计导致状态无限增长。一个KeyBy的维度字段基数大到离谱时状态大小会失控快照写到一半就超时。看存储写入性能。分布式文件系统出现抖动、写入带宽被其他作业抢占都会把快照上传阶段拖垮。给Checkpoint路径换独立的命名空间或者换本地盘做Staging再异步上传都能缓解。如果上述全没问题再考虑开未对齐Checkpoint但要监控快照体积。6.2 恢复后数据对不上账这个问题的根子大概率不在Flink而在Sink。Checkpoint只回滚Flink内部MySQL或者其他外部系统不会跟着回滚已经提交的事务。排查时先确认Sink是不是用的两阶段提交不是的话下游有没有做幂等。幂等写入是很多场景下的兜底方案比依赖分布式事务简单可靠得多。Kafka Sink用事务写、外部存储用幂等键两个配合好精确一次才真正落到端到端。还有一种“恢复后看似重复”的情况其实是Source重放位点和状态恢复顺序不匹配。Checkpoint完成时刻和Source位点保存时刻在语义上必须一致——上面的机制保证了这一点但如果你改了Source端的实现或者手动调整位点这个一致性就会被打破。我们自己做过一次位点手工纠正结果恢复时重复计算了一批数据后来再也不敢碰运行中的位点了。6.3 快照文件数量爆炸文件系统后端默认是完整快照每次都会生成一批新文件长期运行下来存储里的文件会堆积。解决手段有三板斧一是不用自动清理在保存周期策略里设置保留最近N次即可Flink会做清理磁盘压力能缓解二是换成RocksDB后端并开增量三是如果业务对恢复速度容忍度高把有状态的算子改成无状态或尽量减小状态量。这里特别提醒不要图省事把自动清理关了也不设保留数量我见过有作业跑半年快照目录里躺了几十万个文件恢复时HDFS的列表操作直接卡到超时。6.4 Savepoint与Checkpoint别搞混Checkpoint是连续自动触发的设计目标是快速恢复实现细节可能随版本演进而变。Savepoint是手动触发的设计目标是为作业升级、扩容、迁移提供稳定基线Flink会为它保证格式兼容。团队里做发布流程时版本升级前手动触发Savepoint升级失败回滚用Savepoint起正常运行期靠Checkpoint。两种东西混为一谈踩坑的案例我在多个团队都见过有人拿Checkpoint目录去做版本迁移结果新版本不再兼容旧Checkpoint里的序列化格式直接恢复失败。收尾几个忘不掉的细节说实话调Checkpoint这件事实战中最好的老师就是故障本身。我自己的体会是文档里写得再明白都不如一次线上超时让你记得牢。三个印象最深的点分享给读者第一Checkpoint不是独立于数据链路的旁路它就是链路里的一部分背压、GC、磁盘IO任何一环出问题它都会第一个报警第二精确一次不是Flink单方面能送的礼物Sink侧的事务配合或者幂等兜底缺一不可第三排查问题时先看链路再看状态最后才动参数顺序错了一步往往是在给症状吃药。把Chandy-Lamport和Flink的Checkpoint放到一起看其实就是一个优雅的算法如何在真实的大规模工程里被反复打磨的样本。理解了Marker与Barrier的对应关系理解了为什么需要对齐、什么时候可以不对齐你在调参和排查时就有了理论的锚点不会再被监控面板上的数字牵着鼻子走。愿你下次看见Barrier堵在路上的时候能比这次更容易笑出来。
阅读完成 · 觉得有帮助？
