本文首发于泊浮目标语雀:https://www.yuque.com/17sing版本日期备注1.02022.2.2文章首发0.前言将Flink利用至生产已有一段时间，刚上生产的时候有幸排查过因数据歪斜引起的Checkpoint超时问题——过后简略的理解了相干机制，最近正好在读Flink源码，不如趁这个机会搞清楚。
在这里，咱们首先要搞清楚两种Exactly-Once的区别：
Exactly Once：在计算引擎外部，数据不失落不反复。实质是通过Flink开启检查点进行Barrier对齐，即可做到。End to End Exactly Once：这意味着从数据读取、引擎解决到写入内部存储的整个过程中，数据都是不失落不反复的。这要求数据源可重放，写入端反对事务的复原和回滚或幂等。1. 数据歪斜为什么会引起Checkpoint超时做Checkpoint时算子会有一个barrier的对齐机制（为何肯定要对齐前面会讲到）。以下图为例解说对齐过程：
当两条边下发barrier时，barrier1比barrier2先达到了算子，那么算子会将一条边输出的元素缓存起来，直到barrier2到了做Checkpoint当前才会下发元素。
每个算子对齐barrier后，会进行异步状态存储，而后下发barrier。每个算子做完Checkpoint时，会告诉CheckpointCoordinator。当CheckpointCoordinator得悉所有算子的Checkpoint都做完时，认为本次Checkpoint实现。
而在咱们的应用程序中，有一个map算子承受了大量数据，导致barrier始终没有下发，最终整个Checkpoint超时。
2. Checkpoint的原理其具体原理能够参考Flink团队的论文：Lightweight Asynchronous Snapshots for Distributed Dataflow。简略来说，晚期流计算的容错计划都是周期性做全局状态的快照，但这有两个毛病：
阻塞计算——做快照时是同步阻塞的。会将以后算子未解决以及正在解决的record一起做进快照，因而快照会变得特地大。而Flink是基于Chandy-Lamport 算法来扩大的——该算法异步地执行快照，同时要求数据源可重放，但依然会存储上游数据。而Flink的计划提出的计划在无环图中并不会存储数据。
在Flink中（无环有向图），会周期性的插入Barrier这个标记，告知上游算子开始做快照。这个算法基于以下前提：
网络传输牢靠，能够做到FIFO。这里会对算子进行blocked和 unblocked操作，如果一个算子是blocked，它会把从上游通道接管到的所有数据缓存起来，间接收到unblocked的信号才发送。Task能够对它们的通道进行以下操作：block, unblock, send messages, broading messages。对于Source节点来说，会被形象成Nil输出通道。3. Checkpoint的实现在Flink中，做Checkpoint大抵由以下几步组成：
可行性查看JobMaster告诉Task触发检查点TaskExecutor执行检查点JobMaster确认检查点接下来，让咱们跟着源码来看一下外面的具体实现。
3.1 可行性查看参考代码：CheckpointingCoordinator#startTriggeringCheckpoint。
确保作业不是处于敞开中或未启动的状态（见CheckpointPlanCalculator#calculateCheckpointPlan）。生成新的CheckpointingID，并创立一个PendingCheckpoint——当所有Task都实现了Checkpoint，则会转换成一个CompletedCheckpoint。同时也会注册一个线程去关注是否有超时的状况，如果超时则会Abort以后的Checkpoint（见CheckpointPlanCalculator#createPendingCheckpoint）。触发MasterHook。局部内部零碎在触发检查点之前，须要做一些扩大逻辑，通过该实现MasterHook能够实现告诉机制（见CheckpointPlanCalculator#snapshotMasterState）。反复步骤1，没问题的话告诉SourceStreamTask开始触发检查点（见CheckpointPlanCalculator#triggerCheckpointRequest）。3.2 JobMaster告诉Task触发检查点在CheckpointPlanCalculator#triggerCheckpointRequest中，会通过triggerTasks办法调用到Execution#triggerCheckpoint办法。Execution对应了一个Task实例，因而JobMaster能够通过外面的Slot援用找到其TaskManagerGateway，发送近程申请触发Checkpoint。
3.3 TaskManager执行检查点TaskManager在代码中的体现为TaskExecutor。当JobMaster触发近程申请至TaskExecutor时，handle的办法为TaskExecutor#triggerCheckpoint，之后便会调用Task#triggerCheckpointBarrier来做：
做一些查看，比方Task是否是Running状态触发Checkpoint：调用CheckpointableTask#triggerCheckpointAsync执行检查点：CheckpointableTask#triggerCheckpointAsync。以StreamTask实现为例，这里会思考上游曾经Finish时如何触发上游Checkpoint的状况——通过塞入CheckpointBarrier来触发；如果工作没有完结，则调用StreamTask#triggerCheckpointAsyncInMailbox。最终都会走入SubtaskCheckpointCoordinator#checkpointState来触发Checkpoint。算子保留快照：调用OperatorChain#broadcastEvent：保留OperatorState与KeyedState。调用SubtaskCheckpointCoordinatorImpl#finishAndReportAsync，：异步的汇报以后快照已实现。3.4 JobMaster确认检查点|-- RpcCheckpointResponder \-- acknowledgeCheckpoint|-- JobMaster \-- acknowledgeCheckpoint|-- SchedulerBase \-- acknowledgeCheckpoint|-- ExecutionGraphHandler \-- acknowledgeCheckpoint|-- CheckpointCoordinator \-- receiveAcknowledgeMessage在3.1中，咱们提到过PendingCheckpoint。这外面保护了一些状来确保Task全副Ack、Master全副Ack。当确认实现后， CheckpointCoordinator将会告诉所有的Checkpoint曾经实现。
|-- CheckpointCoordinator \-- receiveAcknowledgeMessage \-- sendAcknowledgeMessages3.5 检查点复原该局部代码较为简单，有趣味的同学能够依据相干调用栈自行浏览代码。
|-- Task \-- run \-- doRun|-- StreamTask \-- invoke \-- restoreInternal \-- restoreGates|-- OperatorChain \-- initializeStateAndOpenOperators|-- StreamOperator \-- initializeState|-- StreamOperatorStateHandler \-- initializeOperatorState|-- AbstractStreamOperator \-- initializeState|-- StreamOperatorStateHandler \-- initializeOperatorState|-- CheckpointedStreamOperator \-- initializeState #调用用户代码3.6 End to End Exactly Once端到端的精准一次实现其实是比拟艰难的——思考一个Source对N个Sink的场景。故此Flink设计了相应的接口来保障端到端的精准一次，别离是：
...