在上一篇文章中，我们对 Flink 状态管理相关的代码逻辑进行了分析，但为了实现任务的故障恢复以及数据一致性的效果，还需要借助于检查点（Checkpoint）机制。
简单地说，Checkpoint 是一种分布式快照：在某一时刻，对一个 Flink 作业所有的 task 做一个快照（snapshot），并且将快照保存在 memory / file system 等存储系统中。这样，在任务进行故障恢复的时候，就可以还原到任务故障前最近一次检查点的状态，从而保证数据的一致性。当然，为了保证 exactly-once / at-least-once 的特性，还需要数据源支持数据回放。
概述
Flink 的 checkpoint 机制基于 chandy-lamda 算法，具体的实现可以参考 Flink 官方的文档以及 Flink 团队发表的论文 State Management in Apache Flink。这里先做一下概要性的介绍。
Flink 分布式快照的核心在与 stream barrier，barrier 是一种特殊的标记消息，会和正常的消息记录一起在数据流中向前流动。Checkpoint Coordinator 在需要触发检查点的时候要求数据源向数据流中注入 barrie， barrier 和正常的数据流中的消息一起向前流动，相当于将数据流中的消息切分到了不同的检查点中。当一个 operator 从它所有的 input channel 中都收到了 barrier，则会触发当前 operator 的快照操作，并向其下游 channel 中发射 barrier。当所有的 sink 都反馈完成了快照之后，Checkpoint Coordinator 认为检查点创建完毕。
Checkpoint 的发起流程
CheckpointCoordinator 是 Flink 分布式快照流程的“协调者”，它主要负责：
- 发起 checkpoint 触发的消息，并接收不同 task 对 checkpoint 的响应信息（Ack）
- 维护 Ack 中附带的状态句柄（state-handle）的全局视图
在 StreamingJobGraphGenerator 中，生成 JobGraph 之后会调用 configureCheckpointing 方法进行 Checkpoint 相关的配置。
这其中会有三个列表：
List<JobVertexID> triggerVerticesList<JobVertexID> ackVerticesList<JobVertexID> commitVertices
其中， triggerVertices 只包含那些作为 source 的节点，ackVertices 和 commitVertices 均包含所有的节点：
| ```
class StreamingJobGraphGenerator {
private void configureCheckpointing() {
// ………
// collect the vertices that receive “trigger checkpoint” messages.
// currently, these are all the sources
List
|| --- |在 `ExecutionGraphBuilder#buildGraph` 中，如果作业开启了 checkpoint，则会调用 `ExecutionGraph.enableCheckpointing()` 方法, 这里会创建 `CheckpointCoordinator` 对象，并注册一个作业状态的监听 `CheckpointCoordinatorDeActivator`, `CheckpointCoordinatorDeActivator` 会在作业状态发生改变时得到通知。|
class ExecutionGraph {
public void enableCheckpointing(
long interval,
long checkpointTimeout,
long minPauseBetweenCheckpoints,
int maxConcurrentCheckpoints,
CheckpointRetentionPolicy retentionPolicy,
List
|| --- |当状态变为 RUNNING 时，`CheckpointCoordinatorDeActivator`会得到通知，并且通过 `CheckpointCoordinator.startCheckpointScheduler` 启动 checkpoint 的定时器。|
class CheckpointCoordinatorDeActivator implements JobStatusListener { @Override public void jobStatusChanges(JobID jobId, JobStatus newJobStatus, long timestamp, Throwable error) { if (newJobStatus == JobStatus.RUNNING) { // start the checkpoint scheduler coordinator.startCheckpointScheduler(); } else { // anything else should stop the trigger for now coordinator.stopCheckpointScheduler(); } } }
|| --- |定时任务被封装为 `ScheduledTrigger`, 运行时会调用 `CheckpointCoordinator.triggerCheckpoint()` 触发一次 checkpoint。`CheckpointCoordinator.triggerCheckpoint` 方法代码逻辑很长，概括地说，包括以下几个步骤：- 检查是否可以触发 checkpoint，包括是否需要强制进行 checkpoint，当前正在排队的并发 checkpoint 的数目是否超过阈值，距离上一次成功 checkpoint 的间隔时间是否过小等，如果这些条件不满足，则当前检查点的触发请求不会执行- 检查是否所有需要触发 checkpoint 的 Execution 都是 `RUNNING` 状态- 生成此次 checkpoint 的 checkpointID（id 是严格自增的），并初始化 `CheckpointStorageLocation`，`CheckpointStorageLocation` 是此次 checkpoint 存储位置的抽象，通过 `CheckpointStorage.initializeLocationForCheckpoint()` 创建（`CheckpointStorage` 目前有两个具体实现，分别为 `FsCheckpointStorage` 和 `MemoryBackendCheckpointStorage`），`CheckpointStorage` 则是从 `StateBackend` 中创建- 生成 `PendingCheckpoint`，这表示一个处于中间状态的 checkpoint，并保存在 `checkpointId -> PendingCheckpoint` 这样的映射关系中- 注册一个调度任务，在 checkpoint 超时后取消此次 checkpoint，并重新触发一次新的 checkpoint- 调用 `Execution.triggerCheckpoint()` 方法向所有需要 trigger 的 task 发起 checkpoint 请求savepoint 和 checkpoint 的处理逻辑基本一致，只是 savepoint 是强制触发的，需要调用 `Execution.triggerSynchronousSavepoint()` 进行触发。<br />在CheckpointCoordinator 内部也有三个列表：- `ExecutionVertex[] tasksToTrigger`;- `ExecutionVertex[] tasksToWaitFor`;- `ExecutionVertex[] tasksToCommitTo`;这就对应了前面 `JobGraph` 中的三个列表，在触发 checkpoint 的时候，只有作为 source 的 Execution 会调用 `Execution.triggerCheckpoint()` 方法。会通过 RPC 调用通知对应的 `RpcTaskManagerGateway` 调用 `triggerCheckpoint`。<a name="e1883852"></a>## [](https://blog.jrwang.me/2019/flink-source-code-checkpoint/#checkpoint-%E7%9A%84%E6%89%A7%E8%A1%8C)Checkpoint 的执行<a name="0da19580"></a>### [](https://blog.jrwang.me/2019/flink-source-code-checkpoint/#barrier-%E7%9A%84%E6%B5%81%E5%8A%A8)barrier 的流动`CheckpointCoordinator` 发出触发 checkpoint 的消息，最终通过 RPC 调用 `TaskExecutorGateway.triggerCheckpoint`，即请求执行 `TaskExecutor.triggerCheckpoin()`。 因为一个 `TaskExecutor` 中可能有多个 `Task` 正在运行，因而要根据触发 checkpoint 的 `ExecutionAttemptID` 找到对应的 `Task`，然后调用 `Task.triggerCheckpointBarrier()` 方法。只有作为 source 的 Task 才会触发 `triggerCheckpointBarrier()` 方法的调用。<br />在 `Task` 中，checkpoint 的触发被封装为一个异步任务执行，|
class Task { public void triggerCheckpointBarrier( final long checkpointID, final long checkpointTimestamp, final CheckpointOptions checkpointOptions, final boolean advanceToEndOfEventTime) { …… if (executionState == ExecutionState.RUNNING && invokable != null) { // build a local closure final String taskName = taskNameWithSubtask; final SafetyNetCloseableRegistry safetyNetCloseableRegistry = FileSystemSafetyNet.getSafetyNetCloseableRegistryForThread(); Runnable runnable = new Runnable() { @Override public void run() { // set safety net from the task’s context for checkpointing thread FileSystemSafetyNet.setSafetyNetCloseableRegistryForThread(safetyNetCloseableRegistry); try { //真正的调用逻辑 boolean success = invokable.triggerCheckpoint(checkpointMetaData, checkpointOptions, advanceToEndOfEventTime); if (!success) { checkpointResponder.declineCheckpoint( getJobID(), getExecutionId(), checkpointID, new CheckpointDeclineTaskNotReadyException(taskName)); } } catch (Throwable t) { if (getExecutionState() == ExecutionState.RUNNING) { failExternally(new Exception( “Error while triggering checkpoint “ + checkpointID + “ for “ + taskNameWithSubtask, t)); } else { LOG.debug(“Encountered error while triggering checkpoint {} for “ + “{} ({}) while being not in state running.”, checkpointID, taskNameWithSubtask, executionId, t); } } finally { FileSystemSafetyNet.setSafetyNetCloseableRegistryForThread(null); } } }; //异步执行 executeAsyncCallRunnable( runnable, String.format(“Checkpoint Trigger for %s (%s).”, taskNameWithSubtask, executionId), checkpointOptions.getCheckpointType().isSynchronous()); } else { // send back a message that we did not do the checkpoint checkpointResponder.declineCheckpoint(jobId, executionId, checkpointID, new CheckpointDeclineTaskNotReadyException(taskNameWithSubtask)); } } }
|| --- |`Task` 执行 checkpoint 的真正逻辑被封装在 `AbstractInvokable.triggerCheckpoint(...)` 中，`AbstractInvokable` 中有两个触发 checkpoint 的方法：- `triggerCheckpoint`- `triggerCheckpointOnBarrier`其中 `triggerCheckpoint` 是触发 checkpoint 的源头，会向下游注入 `CheckpointBarrier`；而下游的其他任务在收到 `CheckpointBarrier` 后调用 `triggerCheckpointOnBarrier` 方法。这两个方法的具体实现有一些细微的差异，但主要的逻辑是一致的，在 `StreamTask.performCheckpoint()` 方法中： 1）先向下游发送 barrier， 2）存储检查点快照。<br />一旦 `StreamTask.triggerCheckpoint()` 或 `StreamTask.triggerCheckpointOnBarrier()` b被调用，就会通过 `OperatorChain.broadcastCheckpointBarrier()` 向下游发送 barrier：|
class OperatorChain { public void broadcastCheckpointBarrier(long id, long timestamp, CheckpointOptions checkpointOptions) throws IOException { //创建一个 CheckpointBarrier CheckpointBarrier barrier = new CheckpointBarrier(id, timestamp, checkpointOptions); //向所有的下游发送 for (RecordWriterOutput<?> streamOutput : streamOutputs) { streamOutput.broadcastEvent(barrier); } } }
|| --- |我们已经知道，每一个 Task 的通过 `InputGate` 消费上游 Task 产生的数据，而实际上在 `StreamInputProcessor` 和 `StreamTwoInputProcessor` 中会创建 `CheckpointBarrierHandler`, `CheckpointBarrierHandler` 是对 `InputGate` 的一层封装，增加了对 `CheckpointBarrier` 等事件的处理。`CheckpointBarrierHandler` 有两个具体的实现，即 `BarrierTracker` 和 `BarrierBuffer`，分别对应 AT_LEAST_ONCE 和 EXACTLY_ONCE 这两种模式。<br />`StreamInputProcessor` 和 `StreamTwoInputProcessor` 循环调用 `CheckpointBarrierHandler.getNextNonBlocked()` 获取新数据，因而在 `CheckpointBarrierHandler` 获得 `CheckpointBarrier` 后可以及时地进行 checkpoint 相关的操作。<br />我们先来看一下 AT_LEAST_ONCE 模式下的 `BarrierTracker`，它仅仅追踪从每一个 input channel 接收到的 barrier，当所有 input channel 的 barrier 都被接收时，就可以触发 checkpoint 了：|
public class BarrierTracker implements CheckpointBarrierHandler {
@Override
public BufferOrEvent getNextNonBlocked() throws Exception {
while (true) {
Optional
|| --- |而对于 EXACTLY_ONCE 模式下的 `BarrierBuffer`，它除了要追踪每一个 input channel 接收到的 barrier 之外，在接收到所有的 barrier 之前，先收到 barrier 的 channel 要进入阻塞状态。当然为了避免进入“反压”状态，`BarrierBuffer` 会继续接收数据，但会对接收到的数据进行缓存，直到所有的 barrier 都到达。|
public class BarrierBuffer implements CheckpointBarrierHandler { / To utility to write blocked data to a file channel. */ private final BufferBlocker bufferBlocker; //用于缓存被阻塞的channel接收的数据 /
* The sequence of buffers/events that has been unblocked and must now be consumed before* requesting further data from the input gate.*/private BufferOrEventSequence currentBuffered; //当前缓存的数据@Overridepublic BufferOrEvent getNextNonBlocked() throws Exception {while (true) {// process buffered BufferOrEvents before grabbing new ones// 先处理缓存的数据Optional<BufferOrEvent> next;if (currentBuffered == null) {next = inputGate.getNextBufferOrEvent();}else {next = Optional.ofNullable(currentBuffered.getNext());if (!next.isPresent()) {completeBufferedSequence();return getNextNonBlocked();}}if (!next.isPresent()) {if (!endOfStream) {// end of input stream. stream continues with the buffered dataendOfStream = true;releaseBlocksAndResetBarriers();return getNextNonBlocked();}else {// final end of both input and buffered datareturn null;}}BufferOrEvent bufferOrEvent = next.get();if (isBlocked(bufferOrEvent.getChannelIndex())) {// 如果当前 channel 是 block 状态，先写入缓存// if the channel is blocked, we just store the BufferOrEventbufferBlocker.add(bufferOrEvent);checkSizeLimit();}else if (bufferOrEvent.isBuffer()) {return bufferOrEvent;}else if (bufferOrEvent.getEvent().getClass() == CheckpointBarrier.class) {if (!endOfStream) {// process barriers only if there is a chance of the checkpoint completingprocessBarrier((CheckpointBarrier) bufferOrEvent.getEvent(), bufferOrEvent.getChannelIndex());}}else if (bufferOrEvent.getEvent().getClass() == CancelCheckpointMarker.class) {processCancellationBarrier((CancelCheckpointMarker) bufferOrEvent.getEvent());}else {if (bufferOrEvent.getEvent().getClass() == EndOfPartitionEvent.class) {processEndOfPartition();}return bufferOrEvent;}}}.......
}
|| --- |除了 `CheckpointBarrier` 消息以外，在 checkpoint 发生异常或取消 checkpoint 的时候，会向下游发送 `CancelCheckpointMarker` 消息。<a name="ffcde92f"></a>### [](https://blog.jrwang.me/2019/flink-source-code-checkpoint/#%E5%AD%98%E5%82%A8%E6%A3%80%E6%9F%A5%E7%82%B9%E7%8A%B6%E6%80%81%E5%BF%AB%E7%85%A7)存储检查点状态快照在触发了 checkpoint 之后，对于一个 Task 而言，最重要的就是将当前 Task 中所有算子的状态快照（state snapshot）储存到外部存储系统的。外部存储系统可能是一个分布式文件系统，也可能是 JobManager 内存中。<br />在 `StreamTask.performCheckpoint` 方法中，开始进行 checkpoint 操作，这里主要分为三部分：1）checkpoint的准备操作，这里通常不进行太多操作；2）发送 CheckpointBarrier；3）存储检查点快照：|
class StreamTask {
private boolean performCheckpoint(
CheckpointMetaData checkpointMetaData,
CheckpointOptions checkpointOptions,
CheckpointMetrics checkpointMetrics,
boolean advanceToEndOfTime) throws Exception {
final long checkpointId = checkpointMetaData.getCheckpointId();
final boolean result;
synchronized (lock) {
if (isRunning) {
if (checkpointOptions.getCheckpointType().isSynchronous()) {
syncSavepointLatch.setCheckpointId(checkpointId);
if (advanceToEndOfTime) {
advanceToEndOfEventTime();
}
}
// All of the following steps happen as an atomic step from the perspective of barriers and
// records/watermarks/timers/callbacks.
// We generally try to emit the checkpoint barrier as soon as possible to not affect downstream
// checkpoint alignments
// Step (1): Prepare the checkpoint, allow operators to do some pre-barrier work.
// The pre-barrier work should be nothing or minimal in the common case.
operatorChain.prepareSnapshotPreBarrier(checkpointId);
// Step (2): Send the checkpoint barrier downstream
operatorChain.broadcastCheckpointBarrier(
checkpointId,
checkpointMetaData.getTimestamp(),
checkpointOptions);
// Step (3): Take the state snapshot. This should be largely asynchronous, to not
// impact progress of the streaming topology
checkpointState(checkpointMetaData, checkpointOptions, checkpointMetrics);
result = true;
}
else {
// we cannot perform our checkpoint - let the downstream operators know that they
// should not wait for any input from this operator
// we cannot broadcast the cancellation markers on the ‘operator chain’, because it may not
// yet be created
final CancelCheckpointMarker message = new CancelCheckpointMarker(checkpointMetaData.getCheckpointId());
Exception exception = null;
for (RecordWriter
|| --- |在介绍如何存储检查点快照之前，先简单了解一下和 checkpoint 存储相关的一些类。简单地来说，`CheckpointStorage` 是对状态存储系统的抽象，它有两个不同的实现，分别是 `MemoryBackendCheckpointStorage` 和 `FsCheckpointStorage`。`MemoryBackendCheckpointStorage` 会将所有算子的检查点状态存储在 JobManager 的内存中，通常不适合在生产环境中使用；而 `FsCheckpointStorage` 则会把所有算子的检查点状态持久化存储在文件系统中。`CheckpointStorageLocation` 是对检查点状态存储位置的一个抽象，它能够提供获取检查点输出流的方法，通过输出流将状态和元数据写入到存储系统中。输出流关闭时可以获得状态句柄（StateHandle），后面可以使用句柄重新读取写入的状态。<br />[](https://blog.jrwang.me/img/flink/checkpoint-storage.svg)<br />接着我们来看看进行快照操作的主要逻辑。|
class StreamTask { private void checkpointState( CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions, CheckpointMetrics checkpointMetrics) throws Exception { //1. 解析得到 CheckpointStorageLocation CheckpointStreamFactory storage = checkpointStorage.resolveCheckpointStorageLocation( checkpointMetaData.getCheckpointId(), checkpointOptions.getTargetLocation()); //2. 将存储过程封装为 CheckpointingOperation，开始进行检查点存储操作 CheckpointingOperation checkpointingOperation = new CheckpointingOperation( this, checkpointMetaData, checkpointOptions, storage, checkpointMetrics); checkpointingOperation.executeCheckpointing(); } }
|| --- |每一个算子的快照被抽象为 `OperatorSnapshotFutures`，包含了 operator state 和 keyed state 的快照结果：|
public class OperatorSnapshotFutures {
@Nonnull
private RunnableFuture
|| --- |检查点快照的过程被封装为 `CheckpointingOperation`，由于每一个 `StreamTask` 可能包含多个算子，因而内部使用一个 Map 维护 `OperatorID -> OperatorSnapshotFutures` 的关系。`CheckpointingOperation` 中，快照操作分为两个阶段，第一阶段是同步执行的，第二阶段是异步执行的：|
class StreamTask {
private static final class CheckpointingOperation {
//OperatorID -> OperatorSnapshotFutures
private final Map
|| --- |在同步执行阶段，会依次调用每一个算子的 `StreamOperator.snapshotState`，返回结果是一个 runnable future。根据 checkpoint 配置成同步模式和异步模式的区别，这个 future 可能处于完成状态，也可能处于未完成状态：|
interface StreamOperator
* Called to draw a state snapshot from the operator.** @return a runnable future to the state handle that points to the snapshotted state. For synchronous implementations,* the runnable might already be finished.** @throws Exception exception that happened during snapshotting.*/OperatorSnapshotFutures snapshotState(long checkpointId,long timestamp,CheckpointOptions checkpointOptions,CheckpointStreamFactory storageLocation) throws Exception;
}
public abstract class AbstractStreamOperator
* Stream operators with state, which want to participate in a snapshot need to override this hook method.** @param context context that provides information and means required for taking a snapshot*/public void snapshotState(StateSnapshotContext context) throws Exception {final KeyedStateBackend<?> keyedStateBackend = getKeyedStateBackend();//TODO all of this can be removed once heap-based timers are integrated with RocksDB incremental snapshots// 所有的 timer 都作为 raw keyed state 写入if (keyedStateBackend instanceof AbstractKeyedStateBackend &&((AbstractKeyedStateBackend<?>) keyedStateBackend).requiresLegacySynchronousTimerSnapshots()) {KeyedStateCheckpointOutputStream out;try {out = context.getRawKeyedOperatorStateOutput();} catch (Exception exception) {throw new Exception("Could not open raw keyed operator state stream for " +getOperatorName() + '.', exception);}try {KeyGroupsList allKeyGroups = out.getKeyGroupList();for (int keyGroupIdx : allKeyGroups) {out.startNewKeyGroup(keyGroupIdx);timeServiceManager.snapshotStateForKeyGroup(new DataOutputViewStreamWrapper(out), keyGroupIdx);}} catch (Exception exception) {throw new Exception("Could not write timer service of " + getOperatorName() +" to checkpoint state stream.", exception);} finally {try {out.close();} catch (Exception closeException) {LOG.warn("Could not close raw keyed operator state stream for {}. This " +"might have prevented deleting some state data.", getOperatorName(), closeException);}}}}
}
public abstract class AbstractUdfStreamOperator
|| --- |现在我们已经看到 checkpoint 操作是如何同用户自定义函数建立关联的了，接下来我们来看看由 Flink 托管的状态是如何写入存储系统的，即：|
1 2
|
operatorStateBackend.snapshot(checkpointId, timestamp, factory, checkpointOptions); _keyedStateBackend.snapshot(checkpointId, timestamp, factory, checkpointOptions); //写入 keyed state
|| --- | --- |首先来看看 operator state。`DefaultOperatorStateBackend` 将实际的工作交给 `DefaultOperatorStateBackendSnapshotStrategy` 完成。首先，会为对当前注册的所有 operator state（包含 list state 和 broadcast state）做深度拷贝，然后将实际的写入操作封装在一个异步的 FutureTask 中，这个 FutureTask 的主要任务包括： 1）打开输出流 2）写入状态元数据信息 3）写入状态 4）关闭输出流，获得状态句柄。如果不启用异步checkpoint模式，那么这个 FutureTask 在同步阶段就会立刻执行。|
class DefaultOperatorStateBackendSnapshotStrategy extends AbstractSnapshotStrategy
|| --- |keyed state 写入的基本流程与此相似，但由于 keyed state 在存储时有多种实现，包括基于堆内存和 RocksDB 的不同实现，此外基于 RocksDB 的实现还包括支持[增量 checkpoint](https://flink.apache.org/features/2018/01/30/incremental-checkpointing.html)，因而相比于 operator state 要更复杂一些。另外，Flink 自 1.5.0 版本还引入了一个[本地状态存储](https://issues.apache.org/jira/browse/FLINK-8360)的优化，支持在 TaskManager 的本地保存一份 keyed state，试图优化状态恢复的速度和网络开销。<br />至此，我们介绍了快照操作的第一个阶段，即同步执行的阶段。异步执行阶段被封装为 `AsyncCheckpointRunnable`，主要的操作包括 1）执行同步阶段创建的 FutureTask 2）完成后向 CheckpointCoordinator 发送 Ack 响应。|
class StreamTask {
protected static final class AsyncCheckpointRunnable implements Runnable, Closeable {
@Override
public void run() {
FileSystemSafetyNet.initializeSafetyNetForThread();
try {
TaskStateSnapshot jobManagerTaskOperatorSubtaskStates =
new TaskStateSnapshot(operatorSnapshotsInProgress.size());
TaskStateSnapshot localTaskOperatorSubtaskStates =
new TaskStateSnapshot(operatorSnapshotsInProgress.size());
// 完成每一个 operator 的状态写入
// 如果是同步 checkpoint，那么在此之前状态已经写入完成
// 如果是异步 checkpoint，那么在这里才会写入状态
for (Map.Entry
|| --- |<a name="5896b785"></a>### [](https://blog.jrwang.me/2019/flink-source-code-checkpoint/#%E6%9C%AC%E5%9C%B0%E7%8A%B6%E6%80%81%E5%AD%98%E5%82%A8)本地状态存储所谓本地状态存储，即在存储检查点快照时，在 `Task` 所在的 TaskManager 本地文件系统中存储一份副本，这样在进行状态恢复时可以优先从本地状态进行恢复，从而减少网络数据传输的开销。本地状态存储仅针对 keyed state，我们以较为简单的 `HeapKeyedStateBackend` 为例，看看本地状态存储时如何实现的。|
class HeapSnapshotStrategy
|| --- |其中关键的一点在于，根据是否启用本地状态恢复创建不同的 `CheckpointStreamWithResultProvider`。|
public interface CheckpointStreamWithResultProvider extends Closeable { @Nonnull static CheckpointStreamWithResultProvider createSimpleStream( @Nonnull CheckpointedStateScope checkpointedStateScope, @Nonnull CheckpointStreamFactory primaryStreamFactory) throws IOException { CheckpointStreamFactory.CheckpointStateOutputStream primaryOut = primaryStreamFactory.createCheckpointStateOutputStream(checkpointedStateScope); return new CheckpointStreamWithResultProvider.PrimaryStreamOnly(primaryOut); } @Nonnull static CheckpointStreamWithResultProvider createDuplicatingStream( @Nonnegative long checkpointId, @Nonnull CheckpointedStateScope checkpointedStateScope, @Nonnull CheckpointStreamFactory primaryStreamFactory, @Nonnull LocalRecoveryDirectoryProvider secondaryStreamDirProvider) throws IOException { CheckpointStreamFactory.CheckpointStateOutputStream primaryOut = primaryStreamFactory.createCheckpointStateOutputStream(checkpointedStateScope); try { File outFile = new File( secondaryStreamDirProvider.subtaskSpecificCheckpointDirectory(checkpointId), String.valueOf(UUID.randomUUID())); Path outPath = new Path(outFile.toURI()); CheckpointStreamFactory.CheckpointStateOutputStream secondaryOut = new FileBasedStateOutputStream(outPath.getFileSystem(), outPath); //有两个输出流，primary 和 secondary，secondary 对应本地存储 return new CheckpointStreamWithResultProvider.PrimaryAndSecondaryStream(primaryOut, secondaryOut); } catch (IOException secondaryEx) { LOG.warn(“Exception when opening secondary/local checkpoint output stream. “ + “Continue only with the primary stream.”, secondaryEx); } return new CheckpointStreamWithResultProvider.PrimaryStreamOnly(primaryOut); } }
|| --- |所以在启用本地状态存储的情况下，会创建两个输出流，其中 `primaryOut` 对应外部存储，而 `secondaryOut` 对应本地存储。状态会输出两份。本地状态句柄会存储在 `TaskLocalStateStore` 中。<a name="a48941e2"></a>## [](https://blog.jrwang.me/2019/flink-source-code-checkpoint/#%E5%AF%B9-checkpoint-%E7%9A%84%E7%A1%AE%E8%AE%A4)对 Checkpoint 的确认`Task` 对 checkpoint 的响应是通过 `CheckpointResponder` 接口完成的：|
public interface CheckpointResponder { /**
* Acknowledges the given checkpoint.*/void acknowledgeCheckpoint(JobID jobID,ExecutionAttemptID executionAttemptID,long checkpointId,CheckpointMetrics checkpointMetrics,TaskStateSnapshot subtaskState);/*** Declines the given checkpoint.*/void declineCheckpoint(JobID jobID,ExecutionAttemptID executionAttemptID,long checkpointId,Throwable cause);
}
|| --- |`RpcCheckpointResponder` 作为 `CheckpointResponder` 的具体实现，主要是通过 RPC 调用通知 `CheckpointCoordinatorGateway`，即通知给 `JobMaster`, `JobMaster` 调用 `CheckpointCoordinator.receiveAcknowledgeMessage()` 和 `CheckpointCoordinator.receiveDeclineMessage()` 进行处理。<a name="4cff70c6"></a>### [](https://blog.jrwang.me/2019/flink-source-code-checkpoint/#%E7%A1%AE%E8%AE%A4%E5%AE%8C%E6%88%90)确认完成在一个 `Task` 完成 checkpoint 操作后，`CheckpointCoordinator` 接收到 Ack 响应，对 Ack 响应的处理流程主要如下：- 根据 Ack 的 checkpointID 从 `Map<Long, PendingCheckpoint> pendingCheckpoints` 中查找对应的 `PendingCheckpoint`- 若存在对应的`PendingCheckpoint`- 这个`PendingCheckpoint`没有被丢弃，调用`PendingCheckpoint.acknowledgeTask`方法处理 Ack，根据处理结果的不同：- SUCCESS：判断是否已经接受了所有需要响应的 Ack，如果是，则调用 `completePendingCheckpoint` 完成此次 checkpoint- DUPLICATE：Ack 消息重复接收，直接忽略- UNKNOWN：未知的 Ack 消息，清理上报的 Ack 中携带的状态句柄- DISCARD：Checkpoint 已经被 discard，清理上报的 Ack 中携带的状态句柄- 这个 `PendingCheckpoint` 已经被丢弃，抛出异常- 若不存在对应的 `PendingCheckpoint`，则清理上报的 Ack 中携带的状态句柄相应的代码如下：|
class CheckpointCoordinator { public boolean receiveAcknowledgeMessage(AcknowledgeCheckpoint message) throws CheckpointException { if (shutdown || message == null) { return false; } if (!job.equals(message.getJob())) { LOG.error(“Received wrong AcknowledgeCheckpoint message for job {}: {}”, job, message); return false; } final long checkpointId = message.getCheckpointId(); synchronized (lock) { // we need to check inside the lock for being shutdown as well, otherwise we // get races and invalid error log messages if (shutdown) { return false; } final PendingCheckpoint checkpoint = pendingCheckpoints.get(checkpointId); if (checkpoint != null && !checkpoint.isDiscarded()) { switch (checkpoint.acknowledgeTask(message.getTaskExecutionId(), message.getSubtaskState(), message.getCheckpointMetrics())) { case SUCCESS: LOG.debug(“Received acknowledge message for checkpoint {} from task {} of job {}.”, checkpointId, message.getTaskExecutionId(), message.getJob()); if (checkpoint.isFullyAcknowledged()) { completePendingCheckpoint(checkpoint); } break; case DUPLICATE: LOG.debug(“Received a duplicate acknowledge message for checkpoint {}, task {}, job {}.”, message.getCheckpointId(), message.getTaskExecutionId(), message.getJob()); break; case UNKNOWN: LOG.warn(“Could not acknowledge the checkpoint {} for task {} of job {}, “ + “because the task’s execution attempt id was unknown. Discarding “ + “the state handle to avoid lingering state.”, message.getCheckpointId(), message.getTaskExecutionId(), message.getJob()); discardSubtaskState(message.getJob(), message.getTaskExecutionId(), message.getCheckpointId(), message.getSubtaskState()); break; case DISCARDED: LOG.warn(“Could not acknowledge the checkpoint {} for task {} of job {}, “ + “because the pending checkpoint had been discarded. Discarding the “ + “state handle tp avoid lingering state.”, message.getCheckpointId(), message.getTaskExecutionId(), message.getJob()); discardSubtaskState(message.getJob(), message.getTaskExecutionId(), message.getCheckpointId(), message.getSubtaskState()); } return true; } else if (checkpoint != null) { // this should not happen throw new IllegalStateException( “Received message for discarded but non-removed checkpoint “ + checkpointId); } else { boolean wasPendingCheckpoint; // message is for an unknown checkpoint, or comes too late (checkpoint disposed) if (recentPendingCheckpoints.contains(checkpointId)) { wasPendingCheckpoint = true; LOG.warn(“Received late message for now expired checkpoint attempt {} from “ + “{} of job {}.”, checkpointId, message.getTaskExecutionId(), message.getJob()); } else { LOG.debug(“Received message for an unknown checkpoint {} from {} of job {}.”, checkpointId, message.getTaskExecutionId(), message.getJob()); wasPendingCheckpoint = false; } // try to discard the state so that we don’t have lingering state lying around discardSubtaskState(message.getJob(), message.getTaskExecutionId(), message.getCheckpointId(), message.getSubtaskState()); return wasPendingCheckpoint; } } } }
|| --- |对于一个已经触发但还没有完成的 checkpoint，即 `PendingCheckpoint`，它是如何处理 Ack 消息的呢？在 `PendingCheckpoint` 内部维护了两个 Map，分别是：- `Map<OperatorID, OperatorState> operatorStates;` : 已经接收到 Ack 的算子的状态句柄- `Map<ExecutionAttemptID, ExecutionVertex> notYetAcknowledgedTasks;`: 需要 Ack 但还没有接收到的 Task每当接收到一个 Ack 消息时，`PendingCheckpoint` 就从 `notYetAcknowledgedTasks` 中移除对应的 Task，并保存 Ack 携带的状态句柄保存。当 `notYetAcknowledgedTasks` 为空时，表明所有的 Ack 消息都接收到了。<br />其中 `OperatorState` 是算子状态句柄的一层封装：|
class OperatorState implements CompositeStateHandle {
/ handles to non-partitioned states, subtaskindex -> subtaskstate */
private final Map
|| --- |一旦 `PendingCheckpoint` 确认所有 Ack 消息都已经接收，那么就可以完成此次 checkpoint 了，具体包括：- 调用`PendingCheckpoint.finalizeCheckpoint()`将`PendingCheckpoint`转化为`CompletedCheckpoint`- 获取 `CheckpointMetadataOutputStream`，将所有的状态句柄信息通过 `CheckpointMetadataOutputStream` 写入到存储系统中- 创建一个 `CompletedCheckpoint` 对象- 将`CompletedCheckpoint`保存到`CompletedCheckpointStore`中- `CompletedCheckpointStore` 有两种实现，分别为 `StandaloneCompletedCheckpointStore` 和 `ZooKeeperCompletedCheckpointStore`- `StandaloneCompletedCheckpointStore` 简单地将 `CompletedCheckpointStore` 存放在一个数组中- `ZooKeeperCompletedCheckpointStore` 提供高可用实现：先将 `CompletedCheckpointStore` 写入到 `RetrievableStateStorageHelper` 中（通常是文件系统），然后将文件句柄存在 ZK 中- 保存的 `CompletedCheckpointStore` 数量是有限的，会删除旧的快照- 移除被越过的 `PendingCheckpoint`，因为 `CheckpointID` 是递增的，那么所有比当前完成的 `CheckpointID` 小的 `PendingCheckpoint` 都可以被丢弃了- 依次调用`Execution.notifyCheckpointComplete()`通知所有的 Task 当前 Checkpoint 已经完成- 通过 RPC 调用 `TaskExecutor.confirmCheckpoint()` 告知对应的 Task<a name="0c109f38"></a>### [](https://blog.jrwang.me/2019/flink-source-code-checkpoint/#%E6%8B%92%E7%BB%9D)拒绝在 Task 进行 checkpoint 的过程，可能会发生异常导致 checkpoint 失败，在这种情况下会通过 `CheckpointResponder` 发出回绝的消息。当 `CheckpointCoordinator` 接收到 `DeclineCheckpoint` 消息后会移除 `PendingCheckpoint`，并尝试丢弃已经接收到的 Ack 消息中已完成的状态句柄：|
class CheckpointCoordinator { public void receiveDeclineMessage(DeclineCheckpoint message) { if (shutdown || message == null) { return; } if (!job.equals(message.getJob())) { throw new IllegalArgumentException(“Received DeclineCheckpoint message for job “ + message.getJob() + “ while this coordinator handles job “ + job); } final long checkpointId = message.getCheckpointId(); final String reason = (message.getReason() != null ? message.getReason().getMessage() : “”); PendingCheckpoint checkpoint; synchronized (lock) { // we need to check inside the lock for being shutdown as well, otherwise we // get races and invalid error log messages if (shutdown) { return; } checkpoint = pendingCheckpoints.remove(checkpointId); if (checkpoint != null && !checkpoint.isDiscarded()) { LOG.info(“Decline checkpoint {} by task {} of job {}.”, checkpointId, message.getTaskExecutionId(), job); discardCheckpoint(checkpoint, message.getReason()); } else if (checkpoint != null) { // this should not happen throw new IllegalStateException( “Received message for discarded but non-removed checkpoint “ + checkpointId); } else if (LOG.isDebugEnabled()) { if (recentPendingCheckpoints.contains(checkpointId)) { // message is for an unknown checkpoint, or comes too late (checkpoint disposed) LOG.debug(“Received another decline message for now expired checkpoint attempt {} of job {} : {}”, checkpointId, job, reason); } else { // message is for an unknown checkpoint. might be so old that we don’t even remember it any more LOG.debug(“Received decline message for unknown (too old?) checkpoint attempt {} of job {} : {}”, checkpointId, job, reason); } } } } }
|| --- |<a name="1a0fa53d"></a>## [](https://blog.jrwang.me/2019/flink-source-code-checkpoint/#%E7%8A%B6%E6%80%81%E6%81%A2%E5%A4%8D)状态恢复当 Flink 作业失败重启或者从指定 SavePoint 启动时，需要将整个作业恢复到上一次成功 checkpoint 的状态。这里主要分为两个阶段：- `CheckpointCoordinator` 加载最近一次成功的 `CompletedCheckpoint`，并将状态重新分配到不同的 `Execution`（`Task`）中- `Task` 启动时进行状态初始化<a name="50d79edb"></a>### [](https://blog.jrwang.me/2019/flink-source-code-checkpoint/#%E7%8A%B6%E6%80%81%E5%88%86%E9%85%8D)状态分配首先，`JobMaster` 在创建 `ExecutionGraph` 后会尝试恢复状态到最近一次成功的 checkpoint，或者加载 SavePoint，最终都会调用 `CheckpointCoordinator.restoreLatestCheckpointedState()` 方法：|
class CheckpointCoordinator {
public boolean restoreLatestCheckpointedState(
Map
|| --- |状态的分配过程被封装在 `StateAssignmentOperation` 中。在状态恢复的过程中，假如任务的并行度发生变化，那么每个子任务的状态和先前必然是不一致的，这其中就涉及到状态的平均重新分配问题，关于状态分配的细节，可以参考 Flink 团队的博文 [A Deep Dive into Rescalable State in Apache Flink](https://flink.apache.org/features/2017/07/04/flink-rescalable-state.html#reassigning-operator-state-when-rescaling)，里面给出了 operator state 和 keyed state 重新分配的详细介绍。<br />最终，每个 `Task` 分配的状态被封装在 `JobManagerTaskRestore` 中，并通过 `Execution.setInitialState()` 关联到 `Execution` 中。`JobManagerTaskRestore` 回作为 `TaskDeploymentDescriptor` 的一个属性下发到 `TaskExecutor` 中。<a name="4a0de354"></a>### [](https://blog.jrwang.me/2019/flink-source-code-checkpoint/#task-%E7%8A%B6%E6%80%81%E5%88%9D%E5%A7%8B%E5%8C%96)Task 状态初始化当 `TaskDeploymentDescriptor` 被提交给 `TaskExecutor` 之后，`TaskExecutor` 会 `TaskStateManager` 用于管理当前 `Task` 的状态，`TaskStateManager` 对象会基于分配的 `JobManagerTaskRestore` 和本地状态存储 `TaskLocalStateStore` 进行创建：|
class TaskExecutor {
@Override
public CompletableFuture
|| --- |在 `Task` 启动后，`StreamTask` 会先调用 `initializeState` 方法，这样每一个算子都会调用 `StreamOperator.initializeState()` 进行状态的初始化：|
public abstract class AbstractStreamOperator
|| --- |状态恢复的关键操作在于通过 `StreamTaskStateInitializer.streamOperatorStateContext()` 生成 `StreamOperatorStateContext`, 通过 `StreamOperatorStateContext` 可以获取 state backend，timer service manager 等：|
public interface StreamOperatorStateContext {
// Returns true, the states provided by this context are restored from a checkpoint/savepoint.
boolean isRestored();
// Returns the operator state backend for the stream operator.
OperatorStateBackend operatorStateBackend();
// Returns the keyed state backend for the stream operator. This method returns null for non-keyed operators.
AbstractKeyedStateBackend<?> keyedStateBackend();
// Returns the internal timer service manager for the stream operator. This method returns null for non-keyed operators.
InternalTimeServiceManager<?> internalTimerServiceManager();
// Returns an iterable to obtain input streams for previously stored operator state partitions that are assigned to this stream operator.
CloseableIterable
|| --- |为了生成 `StreamOperatorStateContext`，首先要通过 `TaskStateManager.prioritizedOperatorState()` 方法获得每个 Operator 需要恢复的状态句柄；然后使用获得的状态句柄创建并还原 state backend 和 timer。这里引入了 `PrioritizedOperatorSubtaskState`, 它封装了多个备选的 OperatorSubtaskState (快照)，这些快照相互之间是可以（部分）替换的，并按照优先级排序。列表中的最后一项是包含了这个子任务的所有状态，但是优先级最低。在进行状态恢复的时候，优先从高优先级的状态句柄中读取状态。|
public class PrioritizedOperatorSubtaskState {
/ List of prioritized snapshot alternatives for managed operator state. */
private final List
|| --- |在获得了 `PrioritizedOperatorSubtaskState` 之后就可以进行状态的恢复了：|
public class StreamTaskStateInitializerImpl implements StreamTaskStateInitializer {
@Override
public StreamOperatorStateContext streamOperatorStateContext(
@Nonnull OperatorID operatorID,
@Nonnull String operatorClassName,
@Nonnull KeyContext keyContext,
@Nullable TypeSerializer<?> keySerializer,
@Nonnull CloseableRegistry streamTaskCloseableRegistry,
@Nonnull MetricGroup metricGroup) throws Exception {
TaskInfo taskInfo = environment.getTaskInfo();
OperatorSubtaskDescriptionText operatorSubtaskDescription =
new OperatorSubtaskDescriptionText(
operatorID,
operatorClassName,
taskInfo.getIndexOfThisSubtask(),
taskInfo.getNumberOfParallelSubtasks());
final String operatorIdentifierText = operatorSubtaskDescription.toString();
//先获取用于恢复状态的 PrioritizedOperatorSubtaskState
final PrioritizedOperatorSubtaskState prioritizedOperatorSubtaskStates =
taskStateManager.prioritizedOperatorState(operatorID);
AbstractKeyedStateBackend<?> keyedStatedBackend = null;
OperatorStateBackend operatorStateBackend = null;
CloseableIterable
状态恢复和创建创建 state backend 耦合在一起，借助 BackendRestorerProcedure 来完成，具体的逻辑在 BackendRestorerProcedure.createAndRestore 方法中。
小结
Flink 的 checkpoint 机制是实现故障恢复和数据一致性的重要保障。本文首先对 Flink 的分布式快照的核心思想进行了概述，接着从源码的角度对 checkpoint 的发起、执行、以及状态恢复等流程进行了分析。
参考
- State Management in Apache Flink
- A Deep Dive into Rescalable State in Apache Flink
- Managing Large State in Apache Flink: An Intro to Incremental Checkpointing
-EOF-
若有收获，就点个赞吧
0 人点赞
