【RocksDB 内核机制】生产嵌入对照：Flink · TiKV · Kafka Streams

stream-processing 第 12 篇 已从 Flink 作业侧 说明：EmbeddedRocksDBStateBackend 为每个 subtask 打开独立 RocksDB、每种 registered state 对应一个 Column Family（CF）、KeyGroup 前缀决定 key 排序区间，增量 checkpoint 上传自上次 completed checkpoint 以来 新增的 SST。第 13 篇从 state 设计与调参 讨论写放大、hot key 与 checkpoint 争抢磁盘。

本文是系列 第 16 篇，站在 RocksDB 内核 回答三个互补问题——不迁移 stream/12 的 barrier 对齐与 Web UI 解读全文：

1. Flink 的 KeyGroup | Namespace | UserKey 在 RocksDB InternalKey 排序 里意味着什么？为何 Rescale 必须保留 KeyGroup 前缀？
2. 增量 checkpoint 依赖 RocksDB 哪些 不可变文件语义（SST、MANIFEST、Checkpoint API）？Compaction 如何改变 Flink 的 shared state 引用链？
3. TiKV、Kafka Streams 与 Flink 在 单 DB 实例边界、CF 划分、容错日志 上有何差异？

本文是「RocksDB 内核机制」系列第 16 篇（共 18 篇）。→ 系列目录

篇目	核心内容
第 15 篇 · Checkpoint / Ingest	硬链接快照、SstFileWriter
第 16 篇 · 生产嵌入对照	KeyGroup 内核侧、增量 SST、三系统对照表
第 17 篇 · 经典故障与排查	L0 堆积、LOG / GetProperty

版本锚定：RocksDB 9.x；Flink 1.20+ / 2.x（EmbeddedRocksDBStateBackend）；TiKV 7.x+ Region 本地引擎（文档 Storage）；Kafka Streams 3.x+（RocksDBStore）。
环境说明：本机 WSL2（内核 6.6.x），未部署 Flink 集群。机制来自 Flink / TiKV 官方文档、flink-statebackend-rocksdb 源码与 RocksDB Wiki；不粘贴未执行的 Web UI 截图，也不伪造 checkpoint 耗时或 checkpointedSize 数字。文末给出 Docker 复现步骤，供读者在自有环境验证。

一、与 stream/12、stream/13 的分工

stream/13 强调 改 state 设计优先于拧参数；本篇说明：即便参数正确，L0 堆积或 pending compaction bytes 触发的 write stall 仍会在 checkpoint 前的 强制 Flush 阶段暴露为 syncDuration 上升——这是 LSM 层信号，不是 Flink 调度器单独能消掉的。

二、Flink：KeyGroup 前缀的内核含义

2.1 CF 内 key 布局与排序

Flink 在 CompositeKeySerializationUtils 中构造 RocksDB key（Apache Flink 源码，A 级）：

\[ \text{key}_{\text{rocksdb}} = \underbrace{\text{KeyGroupPrefix}}_{1\text{ 或 }2\text{ 字节}} \,\|\, \text{Namespace} \,\|\, \text{UserKey} \]

• KeyGroup 总数 \(\leq 256\) 时前缀 1 字节，否则 2 字节（大 maxParallelism 作业）。
• Namespace：窗口 state 为 window 标识序列化；非窗口 state 可为空。
• UserKey：业务 keyBy key 经 TypeSerializer 序列化。

RocksDB 在 CF 内按 完整 byte[] key 字典序排序（第 13 篇 Column Family）。因此：

• 同一 subtask 负责的 连续 KeyGroup 区间 在 key 空间上形成 连续前缀段——Rescale 时必须把属于新 subtask 的 KeyGroup 条目迁走，否则 MANIFEST 无法按算子语义切分状态。
• Flink 不在 value 层做 MVCC sequence；窗口「多版本」靠 不同 Namespace 区分，而不是 InternalKey 的 seq（对照 lsm-tree InternalKey）。

flowchart LR
  subgraph CF["Column Family: count"]
    K0["KeyGroup=3 | NS | userA"]
    K1["KeyGroup=3 | NS | userB"]
    K2["KeyGroup=4 | NS | userC"]
  end
  K0 --> K1 --> K2

2.2 与读路径、Block Cache 的交叉

点查 ValueState.value() 走 第 7–9 篇 路径：MemTable → L0（可能多文件）→ L1…Ln。KeyGroup 前缀 不 改变 LSM 层级查找算法，但影响：

• 范围 scan（MapState.entries()、窗口清理 iterator）：扫描范围若跨多个 KeyGroup，仍在一个 CF 内顺序读，成本与 范围内 SST 层数 成正比（stream/12 第四节）。
• Block Cache 局部性：热点 UserKey 若集中在少数 KeyGroup，cache 友好；均匀 hash 到全 KeyGroup 时，读放大接近 leveled 最坏情况（第 9 篇）。

stream/13 的 hot key 诊断应同时看 Flink subtask 指标 与 RocksDB rocksdb.num-files-at-level0（第 17 篇）。

2.3 多 CF 与共享 WAL

每种 registered state 一个 CF（Flink 文档 State Backends，A 级）。内核侧后果（第 13 篇）：

• 共享一条 WAL：一次 WriteBatch 可跨 CF 原子提交；checkpoint 复制 WAL 时需覆盖 多 CF 一致时间点（第 15 篇 Checkpoint）。
• 共享 Block Cache / Write Buffer Manager（managed memory 开启时）：一个 CF 的 compaction 风暴会挤占同 slot 其他算子 CF 的 cache（stream/12 第六节）。

三、增量 checkpoint：SST 上传的内核链

3.1 为何必须先 Flush

增量路径在 barrier 对齐后的 snapshot 阶段 同步调用 Flush（Flink 官方博客 Managing Large State in Apache Flink: An Intro to Incremental Checkpointing，B 级；与 RocksIncrementalSnapshotStrategy 源码一致，A 级）。内核原因：

1. MemTable 中的数据 尚未成为不可变 SST，无法作为稳定文件清单 diff。
2. Flush 产出 L0 SST 后，MANIFEST 记录新 file_number；此后该文件 内容不变，直到 Compaction 淘汰。

这与 第 5 篇 MemTable Flush 的 FlushJob 路径一致：Immutable MemTable → 顺序写 SST → LogAndApply 安装新 Version。

3.2 Checkpoint API 与硬链接

Flink 调用 RocksDB Checkpoint（utilities/checkpoint/checkpoint_impl.*，A 级）在本地临时目录创建 时间点一致 视图（第 15 篇）：

• 与源 DB 同文件系统 时，SST 硬链接（inode 共享，空间不翻倍）。
• MANIFEST、CURRENT、OPTIONS 等 复制。
• 多 CF 时复制 checkpoint 窗口内 WAL，保证跨 CF 一致。

Flink 对比 上次 completed checkpoint 的文件清单：

• 新 SST / 新 MANIFEST 片段 → 上传至 CheckpointStorage；
• 已上传且仍被 live DB 引用 的 SST → 仅在 state handle 中 增加引用，不上传内容。

sequenceDiagram
  participant FL as Flink Snapshot
  participant RB as RocksDB DBImpl
  participant CP as Checkpoint dir
  participant FS as Remote CheckpointStorage
  FL->>RB: Flush()
  RB->>RB: L0 SST + VersionEdit
  FL->>RB: CreateCheckpoint(CP)
  RB->>CP: hardlink SST, copy MANIFEST
  FL->>FL: diff vs last completed CP
  FL->>FS: upload new SST only
  FL->>FS: add ref to shared SST

3.3 Compaction 与 shared state 引用

Compaction 合并 SST 后，旧 file_number 从当前 Version 移除，但 仍被未完成 checkpoint 引用 时不能物理删除——Flink notifyCheckpointComplete 后递减引用，与 RocksDB SuperVersion 引用计数 同构不同层（stream/12 第 5.2 节）。

内核侧推论（机制推导，非 fabricated benchmark）：

• Compaction 频繁时，远程 shared SST 集合 会随合并而 subsumed 旧 checkpoint；Flink 文档说明增量链不会无限增长。
• Compaction 与 checkpoint 争抢 TM 磁盘 I/O；syncDuration 变长时，应同时查 rocksdb.compaction.pending 与 num-running-compactions（第 17 篇），而不只调 execution.checkpointing.timeout。

3.4 与全量 Savepoint 的边界

Savepoint（canonical）走 全量 路径，不依赖 SST 引用链（stream/12 第 5.1 节）。内核上等价于迭代 kv 或导出完整 checkpoint 目录再整体上传——CPU 与网络体积 更高，但 恢复不依赖历史增量链。生产上增量 checkpoint 与 Savepoint 分工见 stream/11–12；本篇只强调：增量语义完全建立在 RocksDB 不可变 SST + MANIFEST 之上。

四、TiKV · Kafka Streams · Flink 对照表

下表比较 单个 RocksDB 实例 在三种系统中的角色。TiKV Raft / Percolator 不在此展开（续作见 db-frontier/12 HTAP）。

TiKV（TiKV Documentation Storage，B 级）：RocksDB 是 Region 本地有序 KV；分布式事务时间戳在 上层编码，不是 RocksDB Snapshot 的 sequence 语义。读 TiKV 性能问题需 分层：Coprocessor / Raft / 本地 LSM。

Kafka Streams（Confluent / Apache Kafka 文档 RocksDB State Store，B 级）：状态变更 双写 本地 RocksDB 与 changelog topic；恢复时 replay changelog 重建 store。没有 Flink 式 增量 SST 上传，但 Compaction 与磁盘满 仍走同一套 write stall（storage/76 ENOSPC）。

五、ClickHouse Embedded 与边界

第 1 篇 LSM 生态 已说明：ClickHouse 主存储是 MergeTree（列存角），RocksDB 出现在 EmbeddedRocksDB 等 辅助 KV。选型上属于 列存为主、点查 KV 为辅——与纯 Flink/TiKV 负载不同，详见 第 18 篇。

本篇不展开：ForSt 远程 SST 实验后端、Changelog state backend（FLIP-158）与 RocksDB 叠加、RocksDBOptionsFactory 全参数表（见 stream/13 与 storage/32）。

六、Docker 复现步骤（不伪造指标）

下列步骤在 读者自有环境 验证 Flink + RocksDB 目录布局与 checkpoint 定性 行为；本文未在本地执行，故不写入 checkpoint 毫秒数或 GB 数。

6.1 环境

• Docker Engine 24+，docker compose v2
• 主机可用内存建议 \(\geq 8\,\text{GiB}\)（JobManager + TaskManager）
• 挂载目录：./flink-data（checkpoint）、./rocksdb-local（TM 本地 state）

6.2 docker-compose.yaml 示例

services:
  jobmanager:
    image: flink:1.20-scala_2.12-java11
    ports:
      - "8081:8081"
    command: jobmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        state.backend.type: rocksdb
        execution.checkpointing.incremental: true
        execution.checkpointing.interval: 60s
        execution.checkpointing.dir: file:///checkpoint-data
        state.backend.rocksdb.localdir: /rocksdb-local
        state.backend.rocksdb.memory.managed: true
    volumes:
      - ./flink-data/checkpoints:/checkpoint-data
      - ./rocksdb-local:/rocksdb-local

  taskmanager:
    image: flink:1.20-scala_2.12-java11
    depends_on:
      - jobmanager
    command: taskmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        taskmanager.numberOfTaskSlots: 4
        state.backend.type: rocksdb
        execution.checkpointing.incremental: true
        execution.checkpointing.dir: file:///checkpoint-data
        state.backend.rocksdb.localdir: /rocksdb-local
    volumes:
      - ./flink-data/checkpoints:/checkpoint-data
      - ./rocksdb-local:/rocksdb-local
    deploy:
      resources:
        limits:
          memory: 4g

6.3 提交带 Keyed State 的作业

使用 Flink 发行版自带的 State Machine Example 或自行打包的 keyed counter JAR（需 EmbeddedRocksDBStateBackend(true) 与 enableCheckpointing）。提交后：

docker compose up -d
# 将作业 JAR 复制进 JobManager 容器后：
docker compose exec jobmanager flink run \
  -d /opt/flink/examples/streaming/StateMachineExample.jar

6.4 观测清单（定性）

1. RocksDB 目录：在 TaskManager 容器内查看 /rocksdb-local 下 {jobId}/{operatorId}/{subtaskIndex}/ 是否出现 CURRENT、MANIFEST-*、*.sst、*.log（与 第 6 篇 布局一致）。
2. Checkpoint 目录：./flink-data/checkpoints 下是否按 checkpoint ID 增长；增量模式下多次 checkpoint 后，单次 checkpointedSize（REST API checkpoints 或 Web UI）应 小于 du -sh 的 RocksDB 本地占用——stream/12 第 5.3 节机制推论。
3. RocksDB LOG（可选）：在 flink-conf 增加 state.backend.rocksdb.log.dir 指向挂载卷，grep Stalling writes / Stopping writes（格式见 第 17 篇 与 db/column_family.cc 源码）。
4. 对比实验：同一作业改用 HashMapStateBackend 运行，比较 checkpoint 上传 内容形态（Java 序列化 blob vs SST 文件集）——记录相对趋势即可，无需在本文中固定数值。

6.5 与 stream/13 联合调优

若 Docker 实验中 syncDuration 持续偏高且 LOG 出现 L0 stall，先按 stream/13 检查 窗口 state 膨胀与 TTL，再按本系列 第 11–12 篇 理解 level0_slowdown_writes_trigger 与 max_background_jobs 的作用边界——改设计优先，调参兜底。

七、小结

• Flink KeyGroup 前缀决定 CF 内 字典序分区；Rescale 与增量恢复依赖此前缀，与 RocksDB InternalKey 的 sequence MVCC 正交。
• 增量 checkpoint = Flush → Checkpoint 硬链接 → SST 文件 diff 上传 + shared 引用；Compaction 改变 SST 集合，由 Flink 引用计数与 RocksDB Version 分层回收。
• TiKV、Kafka Streams、Flink 共享 嵌入 LSM 模式，但 实例边界、CF、容错日志 不同；排查需分层，不把 Raft lag 误当作单一 write_buffer_size 问题。
• stream/12–13 管 作业与 state 设计；本系列第 16–17 篇管 LSM 文件语义与 stall 信号。

下一篇 经典故障与排查 用 GetProperty 与 LOG 把 L0 堆积、compaction 落后、Block Cache 未命中、ENOSPC stall 钉到可核对信号。

上一篇：Checkpoint、Backup 与 External Ingest

下一篇：经典故障与排查

返回 系列目录

参考资料

1. Apache Flink Documentation, State Backends / Checkpointing / Memory Management（EmbeddedRocksDBStateBackend、incremental checkpoint、managed memory；A 级）。
2. Apache Flink 源码：flink-statebackend-rocksdb — CompositeKeySerializationUtils、RocksIncrementalSnapshotStrategy、RocksDBKeyedStateBackend（release-1.20+；A 级）。
3. Apache Flink 官方博客, Managing Large State in Apache Flink: An Intro to Incremental Checkpointing（2018-01-30；B 级）。
4. TiKV Documentation, Storage（Region 本地引擎；B 级）。
5. Apache Kafka Documentation, RocksDB State Store（Kafka Streams；B 级）。
6. RocksDB Wiki, Checkpoints；RocksDB 9.x 源码 utilities/checkpoint/checkpoint_impl.*（A 级）。
7. stream-processing 第 12–13 篇（Flink 作业侧；本篇互补不重复）。
8. 本系列 第 13 篇 Column Family、第 15 篇 Checkpoint。