flinkcheckpoint源码分析二

概述&＃xff1a;上一篇文文章总体接收了flinkcheckpoint的源码分析的总体概念和流程。并结合代码介绍了checkpoint的发起和任务执行过程
详细参考&＃xff1a;https://blog.csdn.net/weixin_40809627/article/details/108537480

本篇文章将接着上篇文章&＃xff0c;继续介绍 flink checkPoint 的检查点快照、本地状态存储、checkpoint的确认、和状态恢复等过程。

一、存储检查点状态快照

在task 触发了chckpoint 之后&＃xff0c;对于Task而言&＃xff0c;最重要的就是将当前 Task 中所有算子的状态快照&＃xff08;state snapshot&＃xff09;储存到外部存储系统的。外部系统可能是一个分布式文件系统&＃xff0c;也可能是JobManager内存中。

在 StreamTask.performCheckpoint 方法中&＃xff0c;开始进行 checkpoint 操作&＃xff0c;这里主要分为三部分&＃xff1a;1&＃xff09;checkpoint的准备操作&＃xff0c;这里通常不进行太多操作&＃xff1b;2&＃xff09;发送 CheckpointBarrier&＃xff1b;3&＃xff09;存储检查点快照&＃xff1a;

class StreamTask {private boolean performCheckpoint(CheckpointMetaData checkpointMetaData,CheckpointOptions checkpointOptions,CheckpointMetrics checkpointMetrics,boolean advanceToEndOfTime) throws Exception {final long checkpointId &＃61; 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 downstreamoperatorChain.broadcastCheckpointBarrier(checkpointId,checkpointMetaData.getTimestamp(),checkpointOptions);// Step (3): Take the state snapshot. This should be largely asynchronous, to not// impact progress of the streaming topologycheckpointState(checkpointMetaData, checkpointOptions, checkpointMetrics);result &＃61; 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 &＃39;operator chain&＃39;, because it may not// yet be createdfinal CancelCheckpointMarker message &＃61; new CancelCheckpointMarker(checkpointMetaData.getCheckpointId());Exception exception &＃61; null;for (RecordWriter>> recordWriter : recordWriters) {try {recordWriter.broadcastEvent(message);} catch (Exception e) {exception &＃61; ExceptionUtils.firstOrSuppressed(new Exception("Could not send cancel checkpoint marker to downstream tasks.", e),exception);}}if (exception !&＃61; null) {throw exception;}result &＃61; false;}}if (isRunning && syncSavepointLatch.isSet()) {//保存 savepoint&＃xff0c;等待 checkpoint 确认完成final boolean checkpointWasAcked &＃61;syncSavepointLatch.blockUntilCheckpointIsAcknowledged();if (checkpointWasAcked) {finishTask();}}return result;}
}


在介绍如何存储检查点快照之前&＃xff0c;先了解下相关checkpoint 存储相关的一些类&＃xff0c;简单地来说&＃xff0c;CheckpointStorage是对状态存储系统地抽象&＃xff0c;它有两个不同的实现&＃xff0c;分别是MemoryBackendCheckpointStorage 和 FsCheckpointStorage。MemoryBackendCheckpointStorage 会将所有算子的检查点状态存储在 JobManager 的内存中&＃xff0c;通常不适合在生产环境中使用&＃xff1b;而 FsCheckpointStorage 则会把所有算子的检查点状态持久化存储在文件系统中。 CheckpointStorageLocation 是对检查点状态存储位置的一个抽象。它呢能过提供获取检查点输出流的方法&＃xff0c;通过输出流将状态和元数据写入到存储系统中。输出流关闭时&＃xff0c;可以获得状态句柄StateHandle&＃xff09;&＃xff0c;后面可以使用句柄重新读取写入的状态。

下面时执行状态快照主要逻辑

每个算个的快照被抽象为OperatorSnapshotFutures&＃xff0c;包含了operator state 和 keyed state 的快照结果&＃xff1a;
检查点快照的过程被封装为CheckpointingOperation&＃xff0c;由于每一个StreamTask 可能包含多个算子&＃xff0c;因而内部使用一个ap 维护 OperatorID -> OperatorSnapshotFutures 的关系。CheckpointingOperation 中&＃xff0c;快照操作分为两个阶段&＃xff0c;第一个阶段同步执行的&＃xff0c;第二个阶段异步执行的。

class StreamTask {private static final class CheckpointingOperation {//OperatorID -> OperatorSnapshotFuturesprivate final Map operatorSnapshotsInProgress;//执行检查点快照public void executeCheckpointing() throws Exception {startSyncPartNano &＃61; System.nanoTime();try {//1. 同步执行的部分for (StreamOperator op : allOperators) {checkpointStreamOperator(op);}//2. 异步执行的部分// checkpoint 可以配置成同步执行&＃xff0c;也可以配置成异步执行的// 如果是同步执行的&＃xff0c;在这里实际上所有的 runnable future 都是已经完成的状态AsyncCheckpointRunnable asyncCheckpointRunnable &＃61; new AsyncCheckpointRunnable(owner,operatorSnapshotsInProgress,checkpointMetaData,checkpointMetrics,startAsyncPartNano);owner.cancelables.registerCloseable(asyncCheckpointRunnable);owner.asyncOperationsThreadPool.submit(asyncCheckpointRunnable);} catch (Exception ex) {........}}&＃64;SuppressWarnings("deprecation")private void checkpointStreamOperator(StreamOperator op) throws Exception {if (null !&＃61; op) {// 调用 StreamOperator.snapshotState 方法进行快照// 返回的结果是 runnable future&＃xff0c;可能是已经执行完了&＃xff0c;也可能没有执行完OperatorSnapshotFutures snapshotInProgress &＃61; op.snapshotState(checkpointMetaData.getCheckpointId(),checkpointMetaData.getTimestamp(),checkpointOptions,storageLocation);operatorSnapshotsInProgress.put(op.getOperatorID(), snapshotInProgress);}}}
}


在同步执行阶段&＃xff0c;会依次调用每一个算子的StreamOperator.snapshotState&＃xff0c;返回结果是一个 runnable future。根据 checkpoint 配置成同步模式和异步模式的区别&＃xff0c;这个 future 可能处于完成状态&＃xff0c;也可能处于未完成状态&＃xff1a; 具体参考代码snapshotState

现在我们已经看到 checkpoint 操作是如何同用户自定义函数建立关联的了&＃xff0c;接下来我们来看看由 Flink 托管的状态是如何写入存储系统的&＃xff0c;即&＃xff1a;

operatorStateBackend.snapshot(checkpointId, timestamp, factory, checkpointOptions); //写入 operator state
keyedStateBackend.snapshot(checkpointId, timestamp, factory, checkpointOptions); //写入 keyed state


首先来看operator state。DefaultOperatorStateBackend 将实际的工作交给DefaultOperatorStateBackendSnapshotStrategy 完成。首先会对当前注册的所有的operator state&＃xff08;包含 list state 和 broadcast state&＃xff09;做深度拷贝&＃xff0c;然后将实际的写入操作封装在一个异步的 FutureTask 中&＃xff0c;这个 FutureTask 的主要任务包括&＃xff1a; 1&＃xff09;、打开输出流 2&＃xff09;、写入状态元数据信息 3&＃xff09;、写入状态 4&＃xff09;、关闭输出流&＃xff0c;获得状态句柄。如果不启动异步checkpoint模式&＃xff0c;那么这个FutureTask 在同步阶段就会立刻执行。

keyed state 写入的基本流程和此相似&＃xff0c;但由于keyed state 在存储时有多种实现&＃xff0c;包括基于堆内存和RocksDB 的不同实现&＃xff0c;此外基于 RocksDB 的实现还包括支持增量 checkpoint&＃xff0c;因而相比于 operator state 要更复杂一些。另外&＃xff0c;Flink 自 1.5.0 版本还引入了一个本地状态存储的优化&＃xff0c;支持在 TaskManager 的本地保存一份 keyed state&＃xff0c;试图优化状态恢复的速度和网络开销。
至此&＃xff0c;我们介绍快照操作的第一个阶段&＃xff0c;即同步执行的阶段。异步执行阶段被封装为AsyncCheckpointRunnable &＃xff0c;主要的操作包括 1&＃xff09;、执行同步阶段创建FutureTask 2&＃xff09;完成后向 CheckpointCoordinator 发送 Ack 响应。

class StreamTask {protected static final class AsyncCheckpointRunnable implements Runnable, Closeable {&＃64;Overridepublic void run() {FileSystemSafetyNet.initializeSafetyNetForThread();try {TaskStateSnapshot jobManagerTaskOperatorSubtaskStates &＃61;new TaskStateSnapshot(operatorSnapshotsInProgress.size());TaskStateSnapshot localTaskOperatorSubtaskStates &＃61;new TaskStateSnapshot(operatorSnapshotsInProgress.size());// 完成每一个 operator 的状态写入// 如果是同步 checkpoint&＃xff0c;那么在此之前状态已经写入完成// 如果是异步 checkpoint&＃xff0c;那么在这里才会写入状态for (Map.Entry entry : operatorSnapshotsInProgress.entrySet()) {OperatorID operatorID &＃61; entry.getKey();OperatorSnapshotFutures snapshotInProgress &＃61; entry.getValue();// finalize the async part of all by executing all snapshot runnablesOperatorSnapshotFinalizer finalizedSnapshots &＃61;new OperatorSnapshotFinalizer(snapshotInProgress);jobManagerTaskOperatorSubtaskStates.putSubtaskStateByOperatorID(operatorID,finalizedSnapshots.getJobManagerOwnedState());localTaskOperatorSubtaskStates.putSubtaskStateByOperatorID(operatorID,finalizedSnapshots.getTaskLocalState());}final long asyncEndNanos &＃61; System.nanoTime();final long asyncDurationMillis &＃61; (asyncEndNanos - asyncStartNanos) / 1_000_000L;checkpointMetrics.setAsyncDurationMillis(asyncDurationMillis);if (asyncCheckpointState.compareAndSet(CheckpointingOperation.AsyncCheckpointState.RUNNING,CheckpointingOperation.AsyncCheckpointState.COMPLETED)) {//报告 snapshot 完成reportCompletedSnapshotStates(jobManagerTaskOperatorSubtaskStates,localTaskOperatorSubtaskStates,asyncDurationMillis);} else {LOG.debug("{} - asynchronous part of checkpoint {} could not be completed because it was closed before.",owner.getName(),checkpointMetaData.getCheckpointId());}} catch (Exception e) {handleExecutionException(e);} finally {owner.cancelables.unregisterCloseable(this);FileSystemSafetyNet.closeSafetyNetAndGuardedResourcesForThread();}}}private void reportCompletedSnapshotStates(TaskStateSnapshot acknowledgedTaskStateSnapshot,TaskStateSnapshot localTaskStateSnapshot,long asyncDurationMillis) {TaskStateManager taskStateManager &＃61; owner.getEnvironment().getTaskStateManager();boolean hasAckState &＃61; acknowledgedTaskStateSnapshot.hasState();boolean hasLocalState &＃61; localTaskStateSnapshot.hasState();// we signal stateless tasks by reporting null, so that there are no attempts to assign empty state// to stateless tasks on restore. This enables simple job modifications that only concern// stateless without the need to assign them uids to match their (always empty) states.taskStateManager.reportTaskStateSnapshots(checkpointMetaData,checkpointMetrics,hasAckState ? acknowledgedTaskStateSnapshot : null,hasLocalState ? localTaskStateSnapshot : null);}
}public class TaskStateManagerImpl implements TaskStateManager {&＃64;Overridepublic void reportTaskStateSnapshots(&＃64;Nonnull CheckpointMetaData checkpointMetaData,&＃64;Nonnull CheckpointMetrics checkpointMetrics,&＃64;Nullable TaskStateSnapshot acknowledgedState,&＃64;Nullable TaskStateSnapshot localState) {long checkpointId &＃61; checkpointMetaData.getCheckpointId();localStateStore.storeLocalState(checkpointId, localState);//发送 ACK 响应给 CheckpointCoordinatorcheckpointResponder.acknowledgeCheckpoint(jobId,executionAttemptID,checkpointId,checkpointMetrics,acknowledgedState);}
}


二、本地状态存储

所谓本地状态存储&＃xff0c;即在存储检查点快照时&＃xff0c;在Task 所在的TaskManager 本地文件系统中存储一份副本&＃xff0c;这样在进行状态恢复时可以优先从本地状态进行恢复&＃xff0c;从而减少网络数据传输的开销。本地状态存储仅针对 keyed state&＃xff0c;我们以较为简单的 HeapKeyedStateBackend 为例&＃xff0c;看看本地状态存储时如何实现的

class HeapSnapshotStrategyextends AbstractSnapshotStrategy implements SnapshotStrategySynchronicityBehavior {&＃64;Nonnull&＃64;Overridepublic RunnableFuture> snapshot(long checkpointId,long timestamp,&＃64;Nonnull CheckpointStreamFactory primaryStreamFactory,&＃64;Nonnull CheckpointOptions checkpointOptions) throws IOException {......//创建 CheckpointStreamWithResultProviderfinal SupplierWithException checkpointStreamSupplier &＃61;localRecoveryConfig.isLocalRecoveryEnabled() ?() -> CheckpointStreamWithResultProvider.createDuplicatingStream(checkpointId,CheckpointedStateScope.EXCLUSIVE,primaryStreamFactory,localRecoveryConfig.getLocalStateDirectoryProvider()) :() -> CheckpointStreamWithResultProvider.createSimpleStream(CheckpointedStateScope.EXCLUSIVE,primaryStreamFactory);........}
}


其中关键的一点在于&＃xff0c;根据是否启用本地状态恢复创建不同的

CheckpointStreamWithResultProvider。
public interface CheckpointStreamWithResultProvider extends Closeable {&＃64;Nonnullstatic CheckpointStreamWithResultProvider createSimpleStream(&＃64;Nonnull CheckpointedStateScope checkpointedStateScope,&＃64;Nonnull CheckpointStreamFactory primaryStreamFactory) throws IOException {CheckpointStreamFactory.CheckpointStateOutputStream primaryOut &＃61;primaryStreamFactory.createCheckpointStateOutputStream(checkpointedStateScope);return new CheckpointStreamWithResultProvider.PrimaryStreamOnly(primaryOut);}&＃64;Nonnullstatic CheckpointStreamWithResultProvider createDuplicatingStream(&＃64;Nonnegative long checkpointId,&＃64;Nonnull CheckpointedStateScope checkpointedStateScope,&＃64;Nonnull CheckpointStreamFactory primaryStreamFactory,&＃64;Nonnull LocalRecoveryDirectoryProvider secondaryStreamDirProvider) throws IOException {CheckpointStreamFactory.CheckpointStateOutputStream primaryOut &＃61;primaryStreamFactory.createCheckpointStateOutputStream(checkpointedStateScope);try {File outFile &＃61; new File(secondaryStreamDirProvider.subtaskSpecificCheckpointDirectory(checkpointId),String.valueOf(UUID.randomUUID()));Path outPath &＃61; new Path(outFile.toURI());CheckpointStreamFactory.CheckpointStateOutputStream secondaryOut &＃61;new FileBasedStateOutputStream(outPath.getFileSystem(), outPath);//有两个输出流&＃xff0c;primary 和 secondary&＃xff0c;secondary 对应本地存储return new CheckpointStreamWithResultProvider.PrimaryAndSecondaryStream(primaryOut, secondaryOut);} catch (IOException secondaryEx) {LOG.warn("Exception when opening secondary/local checkpoint output stream. " &＃43;"Continue only with the primary stream.", secondaryEx);}return new CheckpointStreamWithResultProvider.PrimaryStreamOnly(primaryOut);}
}


所以在启用本地状态存储的情况下&＃xff0c;会创建两个输出流&＃xff0c;其中primaryOut 对应外部存储&＃xff0c;而secondaryOut 对应本地存储。状态会输出两份。本地状态句柄会存储在 TaskLocalStateStore 中。

三、Checkpoint 的确认

todo 结合源码分析&＃xff1a;https://blog.jrwang.me/2019/flink-source-code-checkpoint/