Checkpoint and Savepoint

The source notes for this track are currently maintained in Chinese.

预计阅读时间: 90 分钟 前置阅读: doc-01(JobMaster), doc-02(Barrier介绍) 下一次阅读: doc-04(StateBackend), doc-13(恢复)

1. Chandy-Lamport 分布式快照算法

1.1 算法原理

Flink 的 Checkpoint 基于 Chandy-Lamport（1985）分布式快照算法。该算法的核心思想是：通过在数据流中注入特殊的标记（Marker），让每个进程在收到标记后记录自己的状态，从而获得全局一致快照，而不需要停止整个系统。

Chandy-Lamport 算法的 4 个关键步骤：

1. 发起者（Initiator）: 记录自己的状态，然后向所有出边发送 Marker
2. 接收者（Receiver）:
   - 如果是第一次收到 Marker（来自任意入边），立即记录自己的状态，
     并向所有出边转发 Marker
   - 开始记录该入边之后到达的所有消息（Channel 状态）
3. Channel 状态: 从收到 Marker 的那个入边开始，到该入边再次收到 Marker 前，
   该 Channel 上到达的所有消息都属于 Channel 状态
4. 终止: 当所有进程都收到所有入边的 Marker 后，快照完成

1.2 Flink 的适配：Barrier

Flink 将这个算法中的 Marker 具体化为 CheckpointBarrier——在数据流中传播的特殊事件。Barrier 携带的关键信息：

// CheckpointBarrier.java — Barrier 结构
public class CheckpointBarrier extends RuntimeEvent {
    private final long id;                    // 单调递增的 Checkpoint ID
    private final long timestamp;             // Barrier 创建时间戳 (JM 端)
    private final CheckpointOptions checkpointOptions;  // 选项: aligned/unaligned, timeout 等

    // Barrier 是 RuntimeEvent，与 StreamRecord 在同一 Channel 中顺序传输
}

Barrier 与数据的顺序保证：Barrier 被注入到数据流中，与 StreamRecord 在同一个 Channel（ResultSubpartition → InputChannel）中传输，因此 Barrier 之前的数据一定在 Barrier 之前被处理，不会被"越过"。

1.3 Barrier 对齐（Aligned Checkpoint）

当算子有多个上游输入 Channel 时，它在收到第一个 Channel 的 Barrier 后需要等待其他 Channel 的 Barrier：

算子有 3 个输入 Channel:
  Channel-A:  data1 → data2 → Barrier(#1) → data3 → ...
  Channel-B:  data4 → Barrier(#1) → data5 → ...
  Channel-C:  data6 → data7 → data8 → Barrier(#1) → ...

Barrier 对齐过程:
  1. Channel-A 先到达 Barrier → 算子停止消费 Channel-A，记录 "Barrier received from A"
  2. Channel-B 随后到达 Barrier → 算子停止消费 Channel-B，记录 "received from B"
  3. Channel-C 的 Barrier 到达前，Channel-C 的数据 data6/data7/data8 继续被处理（正常消费）
  4. Channel-C 的 Barrier 到达 → 3 个输入都收到 Barrier → 对齐完成
  5. 算子执行 snapshotState() → 向下游转发 Barrier

对齐的性能代价：在等待慢 Channel 的 Barrier 期间，快 Channel 被阻塞，数据堆积在 Channel 的缓冲区中。这段时间称为 Barrier Alignment Time（对齐时间），是 Checkpoint 性能的关键指标。

1.4 非对齐 Checkpoint（Unaligned Checkpoint）

非对齐 Checkpoint 解决了 Barrier 对齐导致的阻塞问题：

Aligned（对齐）:
  - 等待所有 Channel 的 Barrier 对齐 → 做状态快照
  - 快照内容 = 算子状态 + 所有 Barrier 之前的数据已处理
  - 阻塞: 快 Channel 等待慢 Channel 时暂停消费
  - 延迟代价高但快照体积小

Unaligned（非对齐）:
  - 第一个 Channel 的 Barrier 到达后，立即开始做快照
  - 快照内容 = 算子状态 + in-flight data（未处理但已读取的数据）
  - 不阻塞: 所有 Channel 继续消费数据
  - 延迟低但快照体积大（需要持久化 in-flight buffer）

源码中的实现：

// CheckpointBarrierHandler.java — aligned vs unaligned 的分支
public abstract class CheckpointBarrierHandler implements Closeable {
    // Aligned 实现: 阻塞快 Channel，等待 Barrier 对齐
    // Unaligned 实现: 不阻塞，将 in-flight buffer 直接持久化

    public abstract void processBarrier(
            CheckpointBarrier receivedBarrier, InputChannelInfo channelInfo, boolean isRpcTriggered)
            throws IOException;

    protected void notifyCheckpoint(CheckpointBarrier checkpointBarrier) throws IOException {
        CheckpointMetaData checkpointMetaData =
                new CheckpointMetaData(
                        checkpointBarrier.getId(),
                        checkpointBarrier.getTimestamp(),
                        System.currentTimeMillis());

        CheckpointMetricsBuilder checkpointMetrics;
        if (checkpointBarrier.getId() == startAlignmentCheckpointId) {
            checkpointMetrics =
                new CheckpointMetricsBuilder()
                    .setAlignmentDurationNanos(latestAlignmentDurationNanos)   // 对齐耗时
                    .setBytesProcessedDuringAlignment(latestBytesProcessedDuringAlignment)
                    .setCheckpointStartDelayNanos(latestCheckpointStartDelayNanos); // 传播延迟
        }
        toNotifyOnCheckpoint.triggerCheckpointOnBarrier(
                checkpointMetaData, checkpointBarrier.getCheckpointOptions(), checkpointMetrics);
    }
}

1.5 两种模式对比

2. CheckpointCoordinator — JM 端的 Checkpoint 大脑

2.1 类结构

CheckpointCoordinator（2453 行）运行在 JobMaster 进程中，是 Checkpoint 的全局协调者。

CheckpointCoordinator 的核心状态:

private final long baseInterval;                       // Checkpoint 间隔（ms）
private final long checkpointTimeout;                   // Checkpoint 超时（ms）
private final long minPauseBetweenCheckpoints;          // 两次 Checkpoint 最小间隔
private final Map<Long, PendingCheckpoint> pendingCheckpoints;  // 进行中的 Checkpoint
private final CompletedCheckpointStore completedCheckpointStore; // 已完成的 Checkpoint
private final CheckpointIDCounter checkpointIdCounter;  // CheckpointID 生成器（单调递增）
private final ArrayDeque<Long> recentExpiredCheckpoints; // 最近过期的 Checkpoint ID（去重用）
private ScheduledTrigger currentPeriodicTrigger;        // 当前周期性触发器

2.2 triggerCheckpoint — 触发全流程

// CheckpointCoordinator.java — 核心触发方法 (简化逻辑)
public CompletableFuture<CompletedCheckpoint> triggerCheckpoint(
        CheckpointTriggerRequest request) {

    synchronized (lock) {
        // 1. 检查是否允许触发
        // - 不能有正在进行的 Checkpoint（maxConcurrentCheckpoints=1 时）
        // - 检查 minPauseBetweenCheckpoints
        // - 检查 maxConcurrentCheckpoints

        // 2. 分配 CheckpointID（单调递增，跨 JM 重启保持递增）
        long checkpointID = checkpointIdCounter.getAndIncrement();

        // 3. 获取 Checkpoint 存储位置
        CheckpointStorageLocation checkpointLocation =
            checkpointStorageView.initCheckpoint(checkpointID);

        // 4. 构造 PendingCheckpoint（追踪所有算子的 Ack）
        PendingCheckpoint pendingCheckpoint = new PendingCheckpoint(
            job, checkpointID, checkpointTimestamp,
            checkpointProperties, operatorStates, coordinatorStates,
            checkpointLocation, ...);

        pendingCheckpoints.put(checkpointID, pendingCheckpoint);

        // 5. 触发 Master Hooks（如外部存储的预提交）
        for (MasterTriggerRestoreHook<?> hook : masterHooks.values()) {
            hook.triggerCheckpoint(checkpointID, timestamp, ...);
        }

        // 6. 向所有 Source 算子发送 Checkpoint Trigger（RPC）
        for (ExecutionVertex source : sources) {
            source.triggerCheckpoint(checkpointID, timestamp, checkpointOptions);
        }

        return pendingCheckpoint.getCompletionFuture();
    }
}

触发时机：

1. 周期性触发：Timer 按 baseInterval 定时触发
2. 手动触发：用户通过 REST API 或 flink savepoint 命令触发
3. Checkpoint 完成后自动触发下一个

2.3 PendingCheckpoint — 追踪 Ack

PendingCheckpoint 内部状态:

┌─────────────────────────────────────────┐
│         PendingCheckpoint               │
├─────────────────────────────────────────┤
│ checkpointID: 42                        │
│ numberOfSubtasks: 100                   │
│ numberOfAcknowledgedSubtasks: 0→100       │
│ acknowledgedTasks: Set<ExecutionAttemptID>│
│ stateHandles: List<StateHandle>          │
│ masterStates: Map<String, StateHandle>   │
│ checkpointStorageLocation: .../chk-42    │
│ completionFuture: CompletableFuture<>    │
│                                         │
│ 当 numberOfAcknowledgedSubtasks ==      │
│     numberOfSubtasks 时                 │
│ → Checkpoint COMPLETE!                  │
└─────────────────────────────────────────┘

2.4 receiveAcknowledgeMessage — 接收 Ack

// CheckpointCoordinator.java — 接收算子 Ack
public boolean receiveAcknowledgeMessage(
        AcknowledgeCheckpoint message, String taskManagerLocationInfo) {

    synchronized (lock) {
        long checkpointId = message.getCheckpointId();
        PendingCheckpoint pendingCheckpoint = pendingCheckpoints.get(checkpointId);

        if (pendingCheckpoint == null) {
            // 去重: 检查是否是最近过期的 Checkpoint
            // 如果是则忽略（late ack），否则报错
            if (recentExpiredCheckpoints.contains(checkpointId)) {
                return false;  // 正常的 late ack
            }
            throw new IllegalArgumentException("unknown checkpoint: " + checkpointId);
        }

        // 记录 Ack
        boolean complete = pendingCheckpoint.acknowledgeTask(
            message.getTaskExecutionId(),
            message.getSubtaskState(),    // 算子的 StateHandle
            message.getCheckpointMetrics());

        if (complete) {
            // 所有算子都 Ack 了 → 完成 Checkpoint
            completedCheckpointStore.addCheckpoint(
                pendingCheckpoint.finalizeCheckpoint(), ...);
            // 清理 pending
            pendingCheckpoints.remove(checkpointId);
            recentExpiredCheckpoints.addLast(checkpointId);
            // 通知 Sink Committer 可以提交
            // （Kafka Sink 的两阶段提交在这里最终 commit）
        }

        return complete;
    }
}

Ack 的去重逻辑：使用 recentExpiredCheckpoints (ArrayDeque, 容量 16) 维护最近过期的 Checkpoint ID。当一个 Ack 到达时如果 Checkpoint 已完成/超时，检查是否在过期列表中——在则是正常的 late ack（忽略），不在则是未知的 CheckpointID（报错）。

2.5 Checkpoint 超时处理

// Registered at checkpoint creation
synchronized (lock) {
    ScheduledFuture<?> canceller = timer.schedule(
        () -> {
            synchronized (lock) {
                if (pendingCheckpoints.containsKey(checkpointId)) {
                    PendingCheckpoint timedOut = pendingCheckpoints.remove(checkpointId);
                    timedOut.abortCheckpointOnTimeout();
                    // → notify all operators to cancel this checkpoint
                }
            }
        },
        checkpointTimeout,
        TimeUnit.MILLISECONDS
    );
    pendingCheckpoint.setCancellerHandle(canceller);
}

3. Checkpoint 在算子端的执行

3.1 StreamTask 中的 Checkpoint 处理

// StreamTask.java — 收到 Barrier 后的处理
public void triggerCheckpointOnBarrier(
        CheckpointMetaData checkpointMetaData,
        CheckpointOptions checkpointOptions,
        CheckpointMetricsBuilder checkpointMetrics) throws IOException {

    // 1. 执行算子链中所有算子的 snapshotState()
    operatorChain.snapshotState(
        checkpointMetaData.getCheckpointId(),
        checkpointMetaData.getTimestamp(),
        checkpointOptions,
        checkpointStorageLocation);

    // 2. 完成 Checkpoint（发送 Ack 到 JM）
    operatorChain.finishCheckpoint(checkpointMetaData, checkpointMetrics);
}

3.2 单个算子的 snapshotState

// AbstractStreamOperator.java — 每个算子的快照
public void snapshotState(StateSnapshotContext context) throws Exception {
    // 1. 快照 KeyedState (通过 KeyedStateBackend)
    if (keyedStateBackend != null) {
        keyedStateBackend.snapshot(
            context.getCheckpointId(),
            context.getCheckpointTimestamp(),
            checkpointStorage,
            ...);
    }

    // 2. 快照 OperatorState (如 Kafka Source offset)
    if (operatorStateBackend != null) {
        operatorStateBackend.snapshot(
            context.getCheckpointId(),
            context.getCheckpointTimestamp(),
            checkpointStorage,
            ...);
    }

    // 3. 快照 Timer 状态 (WindowOperator 的 Timer 堆)
    if (internalTimerService != null) {
        internalTimerService.snapshotState(...);
    }
}

3.3 完整 Checkpoint 执行时序

4. 增量 Checkpoint（RocksDB Incremental）

4.1 为什么需要增量 Checkpoint

全量 Checkpoint 每次把 RocksDB 的所有 SST 文件写入外部存储（HDFS/S3），代价极高：

Checkpoint #1: 上传 SST_1, SST_2, SST_3 (3 个文件, 全量)
Checkpoint #2: 上传 SST_1, SST_2, SST_3, SST_4 (4 个文件, 全量) ← SST_1~3 重复上传!
Checkpoint #3: 上传 SST_1,..., SST_5 (5 个文件, 全量)

增量 Checkpoint 只上传与上次 Checkpoint 之间的差异 SST 文件：

Checkpoint #1: 上传 SST_1, SST_2, SST_3 (全量, 没有基线)
Checkpoint #2: 上传 SST_4 (只上传新增的) + SharedStateRegistry 记录 SST_1~3 来自 CHK#1
Checkpoint #3: 上传 SST_5 (只上传新增的) + 记录 SST_3 被 compaction 删除

4.2 SharedStateRegistry — SST 文件跨 Checkpoint 共享

SharedStateRegistry
├── 跟踪每个 SST 文件被哪些 Checkpoint 引用
├── 引用计数 GC: 当所有引用的 Checkpoint 都被丢弃后，SST 文件被删除
└── 文件去重: 多个 Checkpoint 多次引用同一个 SST → 只存一份

文件生命周期:
  SST_1 → 被 Checkpoint#1, #2 引用 (refCount=2)
       → Checkpoint#1 过期 → refCount=1
       → Checkpoint#2 过期 → refCount=0 → 文件被删除

4.3 RocksDB 增量 Snapshot 源码流程

// RocksDBKeyedStateBackend.java — 增量 Checkpoint 的核心
public RunnableFuture<SnapshotResult<KeyedStateHandle>> snapshot(
        long checkpointId, long timestamp,
        CheckpointStreamFactory streamFactory,
        CheckpointOptions checkpointOptions) throws Exception {

    // 1. Flush MemTable → SST (确保内存数据持久化到磁盘)
    for (RocksDBIncrementalSnapshotOperation op : snapshotOperations) {
        op.takeSnapshot();  // RocksDB 的 checkpoint API
    }

    // 2. 获取当前 Checkpoint 的 SST 文件列表
    // 底层调用: RocksDB.getLiveFiles() 或 getColumnFamilyMetaData()
    Set<SstFile> currentSstFiles = getLiveSstFiles();

    // 3. 与上次 Checkpoint 的 SST 文件列表做 diff
    // RocksDBIncrementalCheckpointUtils.chooseSstFilesForUpload()
    Set<SstFile> newSstFiles = diff(currentSstFiles, lastCheckpointSstFiles);

    // 4. 上传新增的 SST 文件到 DFS
    Map<Long, StreamStateHandle> uploadedFiles = rocksDBStateUploader.uploadFilesToCheckpointFs(
        newSstFiles, streamFactory, ...);

    // 5. 构造 IncrementalRemoteKeyedStateHandle
    // - 记录本次 Checkpoint 的新增 SST handles
    // - 通过 SharedStateRegistry 注册对上次 Checkpoint SST 的引用
    IncrementalRemoteKeyedStateHandle handle =
        new IncrementalRemoteKeyedStateHandle(
            backendUID, keyGroupRange, checkpointId,
            sharedStateRegistry,
            uploadedFiles,           // 新上传的
            lastCheckpointSstFiles   // 共享上次的
        );

    // 6. 更新基线
    lastCheckpointSstFiles = currentSstFiles;

    return handle;
}

4.4 增量 Checkpoint vs 全量

5. Savepoint vs Checkpoint — 深入对比

5.1 本质区别

5.2 Savepoint 的自包含格式（Canonical）

Savepoint 使用 Self-Contained 格式：所有状态数据（SST 文件）完整写入 Savepoint 目录，不依赖任何外部引用。

[hdfs://savepoints/savepoint-abc123]
├── _metadata                    # 所有算子的状态元数据
│   ├── OperatorID → KeyGroupRanges → StateHandle 列表
│   ├── MasterState (CheckpointCoordinator 状态)
│   └── JobGraph 快照 (用于恢复验证)
├── 00000000-0000-...            # SST 文件（全量，非增量引用）
├── 00000001-0000-...
└── ...

Savepoint 为什么必须是自包含的？

• Checkpoint 可以增量引用，因为它们是时间序列（下一个总是存在）
• Savepoint 可能被拷贝到其他集群、HDFS → S3 迁移 → 必须自包含

5.3 Savepoint 恢复流程

// ExecutionGraph.java — 从 Savepoint 恢复
public void restoreSavepoint(SavepointRestoreSettings settings) {
    // 1. 从 CheckpointStore 加载 Savepoint
    CompletedCheckpoint savepoint = checkpointStore.getLatestCheckpoint();

    // 2. 解析 StateAssignment
    // - KeyGroupRangeAssignment 重新计算每个 Subtask 的 KeyGroup 范围
    // - 将 StateHandle 重新分配给新的 Subtask
    Map<OperatorID, OperatorState> oldStates = savepoint.getOperatorStates();
    StateAssignmentOperation stateAssignment = new StateAssignmentOperation(
        oldStates, newExecutionVertices, maxParallelism);

    // 3. 为每个 ExecutionVertex 分配对应的 StateHandle
    Map<ExecutionAttemptID, List<StateHandle>> assigned =
        stateAssignment.assignStates();

    // 4. Deploy Task 时带上 StateHandle
    // → Task.doRun() → StreamTask.initializeState(stateHandles)
    // → StateBackend.initializeState(stateHandles)
}

6. Checkpoint 存储后端

6.1 CheckpointStorage 接口

// CheckpointStorage.java — 决定 Checkpoint 写到哪里
public interface CheckpointStorage {
    // 初始化 Checkpoint 存储位置
    CheckpointStorageLocation initCheckpoint(long checkpointId) throws IOException;

    // 解析存储位置（恢复时用）
    CompletedCheckpointStorageLocation resolveCheckpoint(String externalPointer);
}

// CheckpointStorageLocation — 一次 Checkpoint 的存储位置
public interface CheckpointStorageLocation {
    StreamFactory createCheckpointStream();  // 创建写入流
    String getLocationAsPointer();           // 返回外部指针（如 hdfs://path/chk-42）
}

6.2 存储类型

6.3 CompletedCheckpointStore

CompletedCheckpointStore — 存储已完成的 Checkpoint 元数据

实现:
  - StandaloneCompletedCheckpointStore: JM 内存中，最多保留 N 个
  - ZooKeeperCompletedCheckpointStore: ZK 持久化
  - FileSystemCompletedCheckpointStore: HDFS/S3 文件持久化

7. Checkpoint 关键配置深度解析

8. Checkpoint 性能分析

8.1 Checkpoint 耗时组成

TotalCheckpointDuration =
│
├── BarrierInjectionTime           (0-100ms, Source 注入 Barrier 的延迟)
├── BarrierPropagationTime         (Barrier 穿越整个 DAG 的时间)
│   ├── AlignmentTime              (对齐时间，主要耗时)
│   └── NetworkLatency             (Channel 延迟)
├── SyncSnapshotTime               (算子的同步快照时间: flush MemTable 等)
└── AsyncSnapshotTime              (算子的异步快照时间: 上传 SST 到 DFS)
                                    ↑ 通常最长

典型 Checkpoint 耗时分布 (100GB RocksDB 状态，S3 存储):
  BarrierPropagation: 100ms  (对齐)
  SyncSnapshot:       200ms  (flush MemTable)
  AsyncSnapshot:      30s    (上传增量 SST，可以在后台继续处理数据)

8.2 大状态 Checkpoint 优化

1. 增量 Checkpoint：最有效的优化
2. 增加 Checkpoint 间隔：60s → 180s → 600s，减少 Checkpoint 频率
3. 使用更快的存储：S3 → HDFS（本地机房）→ 本地 SSD
4. 减小状态大小：设置 State TTL、使用 RocksDB 压缩、清理无用状态
5. Unaligned Checkpoint：在反压严重时消除对齐时间

9. 源码导航（完整版）

10. 常见问题 / 面试题

Q1: Checkpoint 超时的主要原因和排查方法？

A: 原因：(1) 反压 → Barrier 传播慢 → 查看 backPressuredTime；(2) 状态太大 → snapshot 耗时长 → 查看 checkpointSize 指标；(3) 外部存储慢（S3/EBS）→ 查看 AsyncSnapshotTime；(4) 对齐时间长 → 查看 AlignmentDuration。排查：从 Checkpoint History 中看哪个 Subtask 最后 Ack → 定位到具体的慢算子。

Q2: 非对齐 Checkpoint 为什么能降低延迟？原理是什么？

A: Aligned checkpoint 需要等待所有 Channel 的 Barrier 对齐，快 Channel 在这期间阻塞。Unaligned 不阻塞——第一个 Channel 的 Barrier 到达后立即开始 snapshot，将 Channel 中已读但未处理的 in-flight data（InputChannel buffer 内容）也持久化。代价是 Checkpoint 体积更大（包含 in-flight buffer），且恢复时需要先恢复 in-flight data 再重放。

Q3: 为什么不把 Checkpoint 间隔设得非常短（如每 5 秒）？

A: (1) Checkpoint 本身有开销（CPU + I/O + 网络）; (2) 如果一次 Checkpoint 耗时 > 5 秒，下一次 Checkpoint 被 min-pause 阻塞，系统陷入持续的 Checkpoint 循环; (3) 小间隔意味着很多 Checkpoint 都是几乎相同的状态（增量很小），浪费资源。推荐最小值 60 秒。

Q4: 增量 Checkpoint 恢复时会不会因为 SST 文件跨 Checkpoint 共享而变慢？

A: 恢复时仍需下载所有 SST 文件（新增的 + 引用的），所以恢复时间由总状态大小决定，与增量无关。但：下载可以并发（多个文件从 S3 并行下载），且 RocksDB 恢复后只加载到 Block Cache 而不全量加载到内存。

Q5: Checkpoint 失败后 Flink 如何处理？

A: (1) 失败的 Checkpoint 被丢弃（pendingCheckpoints.remove()）; (2) 向所有算子发送 CancelCheckpointMarker，让算子清理该 Checkpoint 的中间状态; (3) 下一个 Checkpoint 使用新的 CheckpointID; (4) 如果连续失败次数超过 tolerableCheckpointFailureNumber（默认 0 = 无容忍），Job 会 Fail。

Q6: Savepoint 恢复时并行度可以改吗？

A: 可以。Savepoint 按 KeyGroup 存储状态，KeyGroup 再映射到 Subtask。修改并行度 = 改变 KeyGroup→Subtask 的映射，不改变 Key→KeyGroup 的映射。因此状态可以重分配。但 OperatorState（如 Kafka Source offset）的恢复需要特殊处理——FLIP-27 Source 的 SplitEnumerator 会重新分配 Partition→Subtask。

下一步

• StateBackend 深度: doc-04-state-and-statebackend.md — RocksDB/HashMap/Changelog Backend 内部原理
• 状态恢复与 Rescale: doc-13-state-recovery-and-rescale.md — Savepoint 恢复 + 弹性扩缩容