版本 | 日期 | 备注 |
---|---|---|
1.0 | 2022.2.2 | 文章首发 |
1.1 | 2022.2.14 | 更新3.4部分,增强注释部分 |
1.2 | 2022.2.27 | 更新3.6部分,删除部分对于1.14版本不适的描述 |
1.3 | 2022.3.8 | fix typo |
将Flink应用至生产已有一段时间,刚上生产的时候有幸排查过因数据倾斜引起的Checkpoint超时问题——当时简单的了解了相关机制,最近正好在读Flink源码,不如趁这个机会搞清楚。
在这里,我们首先要搞清楚两种Exactly-Once的区别:
本文基于Flink 1.14代码进行分析。
做Checkpoint时算子会有一个barrier的对齐机制(为何一定要对齐后面会讲到)。以下图为例讲解对齐过程:
当两条边下发barrier时,barrier1比barrier2先到达了算子,那么算子会将一条边输入的元素缓存起来,直到barrier2到了做Checkpoint以后才会下发元素。
每个算子对齐barrier后,会进行异步状态存储,然后下发barrier。每个算子做完Checkpoint时,会通知CheckpointCoordinator
。当CheckpointCoordinator
得知所有算子的Checkpoint都做完时,认为本次Checkpoint完成。
而在我们的应用程序中,有一个map算子接受了大量数据,导致barrier一直没有下发,最终整个Checkpoint超时。
其具体原理可以参考Flink团队的论文:Lightweight Asynchronous Snapshots for Distributed Dataflow。简单来说,早期流计算的容错方案都是周期性做全局状态的快照,但这有两个缺点:
而Flink是基于Chandy-Lamport 算法来扩展的——该算法异步地执行快照,同时要求数据源可重放,但仍然会存储上游数据。而Flink的方案提出的方案在无环图中并不会存储数据。
在Flink中(无环有向图),会周期性的插入Barrier这个标记,告知下游算子开始做快照。这个算法基于以下前提:
blocked
和 unblocked
操作,如果一个算子是blocked
,它会把从上游通道接收到的所有数据缓存起来,直接收到unblocked
的信号才发送。block
, unblock
, send messages
, broading messages
。Nil
输入通道。在Flink中,做Checkpoint大致由以下几步组成:
接下来,让我们跟着源码来看一下里面的具体实现。
参考代码:CheckpointCoordinator#startTriggeringCheckpoint
。
CheckpointPlanCalculator#calculateCheckpointPlan
)。CheckpointPlanCalculator#createPendingCheckpoint
)。CheckpointPlanCalculator#snapshotMasterState
)。CheckpointPlanCalculator#triggerCheckpointRequest
)。在CheckpointPlanCalculator#triggerCheckpointRequest
中,会通过triggerTasks
方法调用到Execution#triggerCheckpoint
方法。Execution对应了一个Task实例,因此JobMaster可以通过里面的Slot引用找到其TaskManagerGateway
,发送远程请求触发Checkpoint。
TaskManager在代码中的体现为TaskExecutor
。当JobMaster触发远程请求至TaskExecutor时,handle的方法为TaskExecutor#triggerCheckpoint
,之后便会调用Task#triggerCheckpointBarrier
来做:
CheckpointableTask#triggerCheckpointAsync
CheckpointableTask#triggerCheckpointAsync
。以StreamTask
实现为例,这里会考虑上游已经Finish时如何触发下游Checkpoint的情况——通过塞入CheckpointBarrier
来触发;如果任务没有结束,则调用StreamTask#triggerCheckpointAsyncInMailbox
。最终都会走入SubtaskCheckpointCoordinator#checkpointState
来触发Checkpoint。OperatorChain#broadcastEvent
:保存OperatorState与KeyedState。SubtaskCheckpointCoordinatorImpl#finishAndReportAsync
,:异步的汇报当前快照已完成。|-- RpcCheckpointResponder
\-- acknowledgeCheckpoint
|-- JobMaster
\-- acknowledgeCheckpoint
|-- SchedulerBase
\-- acknowledgeCheckpoint
|-- ExecutionGraphHandler
\-- acknowledgeCheckpoint
|-- CheckpointCoordinator
\-- receiveAcknowledgeMessage
在3.1中,我们提到过PendingCheckpoint。这里面维护了一些状来确保Task全部Ack、Master全部Ack。当确认完成后,CheckpointCoordinator
将会通知所有的Checkpoint已经完成。
|-- CheckpointCoordinator
\-- receiveAcknowledgeMessage
\-- sendAcknowledgeMessages //通知下游Checkpoint已经完成。如果Sink实现了TwoPhaseCommitSinkFunction,将会Commit;如果因为一些原因导致Commit没有成功,则会抛出一个FlinkRuntimeException,而pendingCommitTransactions中的将会继续保存失败的CheckpointId,当检查点恢复时将会重新执行。
该部分代码较为简单,有兴趣的同学可以根据相关调用栈自行阅读代码。
|-- Task
\-- run
\-- doRun
|-- StreamTask
\-- invoke
\-- restoreInternal
\-- restoreGates
|-- OperatorChain
\-- initializeStateAndOpenOperators
|-- StreamOperator
\-- initializeState
|-- StreamOperatorStateHandler
\-- initializeOperatorState
|-- AbstractStreamOperator
\-- initializeState
|-- StreamOperatorStateHandler
\-- initializeOperatorState
|-- CheckpointedStreamOperator
\-- initializeState #调用用户代码
端到端的精准一次实现其实是比较困难的——考虑一个Source对N个Sink的场景。故此Flink设计了相应的接口来保障端到端的精准一次,分别是:
目前Source和Sink全部ExactlyOnce实现的只有Kafka——其上游支持断点读取,下游支持回滚or幂等。有兴趣的同学可以阅读该接口的相关实现。
可能有同学会好奇为什么JDBC Sink没有实现ExactlyOnce。本质和这个接口的执行方式无法兼容JDBC的事务使用方式——当一个算子意味退出时,是无法再对之前的事务进行操作的。因此TwoPhaseCommitSinkFunction
中的retryCommit以及retryRollback是无法进行的——见https://github.com/apache/flink/pull/10847。JDBC的Sink是基于XA实现的,尽可能保证一致性。这里可能又有同学会问了为什么不用Upset类的语句,因为这个方式并不通用——对于Upset需要一个唯一键,不然性能极差。
本文以问题视角切入Checkpoint的原理与实现,并对相关源码做了简单的跟踪。其实代码的线路是比较清晰的,但涉及大量的类——有心的同学可能已经发现,这是单一职责原则的体现。TwoPhaseCommitSinkFunction
中的实现也是典型的模版方法设计模式。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。