前言

上一篇，我们了解了 checkpoint 全流程，现在我们具体讲解一下 checkpoint 时 snapshot 的全过程。现在我们具体看一下 checkpoint 时是如何做 snapshot 的

正文

由 checkpoint 全流程 我们可以知道

public void executeCheckpointing() throws Exception {startSyncPartNano = System.nanoTime();try {// 调用 StreamOperator 进行 snapshotState 的入口方法// 先 sourceOperator (flatMap -> source) 再 sinkOperator (sink -> filter)for (StreamOperator<?> op : allOperators) {//对每一个算子进行 snapshotInProgress 并存储至 operatorSnapshotsInProgress// (存储 是异步checkpoint的一个引用) 然后分别进行本地 checkpoint store and jobManager ack// 捕获 barrier 的过程其实就是处理 input 数据的过程，对应着 StreamInputProcessor.processInput() 方法checkpointStreamOperator(op);}
......}

是做 snapshot 逻辑，具体如下( AbstractStreamOperator.snapshotState )

@Override// 由此处统一持久化public final OperatorSnapshotFutures snapshotState(long checkpointId, long timestamp, CheckpointOptions checkpointOptions,CheckpointStreamFactory factory) throws Exception {KeyGroupRange keyGroupRange = null != keyedStateBackend ?keyedStateBackend.getKeyGroupRange() : KeyGroupRange.EMPTY_KEY_GROUP_RANGE;OperatorSnapshotFutures snapshotInProgress = new OperatorSnapshotFutures();try (StateSnapshotContextSynchronousImpl snapshotContext = new StateSnapshotContextSynchronousImpl(checkpointId,timestamp,factory,keyGroupRange,getContainingTask().getCancelables())) {snapshotState(snapshotContext);snapshotInProgress.setKeyedStateRawFuture(snapshotContext.getKeyedStateStreamFuture());snapshotInProgress.setOperatorStateRawFuture(snapshotContext.getOperatorStateStreamFuture());// state 持久化if (null != operatorStateBackend) {snapshotInProgress.setOperatorStateManagedFuture(// 触发一个异步的 snapshot 至 DefaultOperatorStateBackend(内部的)operatorStateBackend.snapshot(checkpointId, timestamp, factory, checkpointOptions));}// source -> flatMap --> rebance --> filter --> keyby --> sink// 只有当 sink 的时候，keyedStateBackend 才不为 null , 才会执行 backend 的 snapshotif (null != keyedStateBackend) {snapshotInProgress.setKeyedStateManagedFuture(// 触发一个异步的 snapshot 至 StateBackedkeyedStateBackend.snapshot(checkpointId, timestamp, factory, checkpointOptions));}} catch (Exception snapshotException) {try {snapshotInProgress.cancel();} catch (Exception e) {snapshotException.addSuppressed(e);}String snapshotFailMessage = "Could not complete snapshot " + checkpointId + " for operator " +getOperatorName() + ".";if (!getContainingTask().isCanceled()) {LOG.info(snapshotFailMessage, snapshotException);}throw new Exception(snapshotFailMessage, snapshotException);}return snapshotInProgress;}

由此可以知道，如果是 null != operatorStateBackend 则 operatorStateBackend.snapshot，如果 null != keyedStateBackend 则 keyedStateBackend.snapshot。
此处，我们以 RocksDBIncrementalSnapshotOperation 为例 ( operatorStateBackend.snapshot 的代码注释已经很清楚了 )

@Nonnull@Overrideprotected RunnableFuture<SnapshotResult<KeyedStateHandle>> doSnapshot(long checkpointId,long checkpointTimestamp,@Nonnull CheckpointStreamFactory checkpointStreamFactory,@Nonnull CheckpointOptions checkpointOptions) throws Exception {final SnapshotDirectory snapshotDirectory = prepareLocalSnapshotDirectory(checkpointId);LOG.trace("Local RocksDB checkpoint goes to backup path {}.", snapshotDirectory);//  RocksDBIncrementalRestoreOperation 中 kvStateInformation 赋值//  kvStateInformation.put(columnFamilyName, registeredColumn(RocksDBKeyedStateBackend.RocksDbKvStateInfo));final List<StateMetaInfoSnapshot> stateMetaInfoSnapshots = new ArrayList<>(kvStateInformation.size());final Set<StateHandleID> baseSstFiles = snapshotMetaData(checkpointId, stateMetaInfoSnapshots);// 对 rocksdb 做 checkpoint 为 RocksDBIncrementalSnapshotOperation.uploadSstFiles 做准备takeDBNativeCheckpoint(snapshotDirectory);// snapshotfinal RocksDBIncrementalSnapshotOperation snapshotOperation =new RocksDBIncrementalSnapshotOperation(checkpointId,checkpointStreamFactory,snapshotDirectory,baseSstFiles,stateMetaInfoSnapshots);return snapshotOperation.toAsyncSnapshotFutureTask(cancelStreamRegistry);}

进入 RocksDBIncrementalSnapshotOperation 内部

@Overrideprotected SnapshotResult<KeyedStateHandle> callInternal() throws Exception {boolean completed = false;// Handle to the meta data fileSnapshotResult<StreamStateHandle> metaStateHandle = null;// Handles to new sst files since the last completed checkpoint will go herefinal Map<StateHandleID, StreamStateHandle> sstFiles = new HashMap<>();// Handles to the misc files in the current snapshot will go herefinal Map<StateHandleID, StreamStateHandle> miscFiles = new HashMap<>();try {// 写 meta (全量) 到 hdfsmetaStateHandle = materializeMetaData();// Sanity checks - they should never failPreconditions.checkNotNull(metaStateHandle, "Metadata was not properly created.");Preconditions.checkNotNull(metaStateHandle.getJobManagerOwnedSnapshot(),"Metadata for job manager was not properly created.");//  将新产生的 sst file、misc file upload to checkpointFsuploadSstFiles(sstFiles, miscFiles);synchronized (materializedSstFiles) {materializedSstFiles.put(checkpointId, sstFiles.keySet());}final IncrementalRemoteKeyedStateHandle jmIncrementalKeyedStateHandle =new IncrementalRemoteKeyedStateHandle(backendUID,keyGroupRange,checkpointId,sstFiles,miscFiles,metaStateHandle.getJobManagerOwnedSnapshot());//PermanentSnapshotDirectoryfinal DirectoryStateHandle directoryStateHandle = localBackupDirectory.completeSnapshotAndGetHandle();final SnapshotResult<KeyedStateHandle> snapshotResult;if (directoryStateHandle != null && metaStateHandle.getTaskLocalSnapshot() != null) {// 增量的 localSnapshotIncrementalLocalKeyedStateHandle localDirKeyedStateHandle =new IncrementalLocalKeyedStateHandle(backendUID,checkpointId,directoryStateHandle,keyGroupRange,metaStateHandle.getTaskLocalSnapshot(),sstFiles.keySet());//  localSnapshot report to local state manager,//  jobManagerState(jmIncrementalKeyedStateHandle) report to job managersnapshotResult = SnapshotResult.withLocalState(jmIncrementalKeyedStateHandle, localDirKeyedStateHandle);} else {//jobManagerState(jmIncrementalKeyedStateHandle) report to job managersnapshotResult = SnapshotResult.of(jmIncrementalKeyedStateHandle);}completed = true;return snapshotResult;} finally {if (!completed) {final List<StateObject> statesToDiscard =new ArrayList<>(1 + miscFiles.size() + sstFiles.size());statesToDiscard.add(metaStateHandle);statesToDiscard.addAll(miscFiles.values());statesToDiscard.addAll(sstFiles.values());cleanupIncompleteSnapshot(statesToDiscard);}}}

元数据是全部持久化，而数据仅仅将新产生的 sst file、misc file upload to checkpointFs