【流式数据处理】状态放大、Compaction 与调优

第 12 篇 把 EmbeddedRocksDBStateBackend 的 ColumnFamily、KeyGroup 前缀、增量 checkpoint 与 lsm-tree 系列 的 WAL / MemTable / SSTable 路径对齐了。生产里更常见的故障却是：磁盘占用持续爬升、checkpoint 同步阶段越来越长、某个 subtask 的 RocksDB 指标独高、全作业背压但 CPU 并不满——这些往往不是「再调一个 JVM 参数」能解决的，而是 state 设计、LSM 写放大、KeyGroup 倾斜 叠在一起的结果。

本文回答四个问题：

1. 窗口 state、长 TTL、MapState 无限增长各自如何把 逻辑状态 放大成 磁盘 bytes？
2. RocksDB Compaction 与 Flink 增量 checkpoint 如何在 TaskManager 磁盘上抢 I/O？写放大公式 \(WA \approx T \times L\) 在流作业里意味着什么？
3. Flink 场景下 write_buffer_size、max_background_jobs、managed memory 比例等旋钮 有效边界 在哪？
4. Hot key 倾斜 时，为什么加并行度无效，何时该改 key 设计或预聚合而不是继续调 RocksDB？

本文是「流式数据处理」系列第 13 篇（共 18 篇）。→ 系列目录

篇目	核心内容
第 12 篇 · RocksDB 内核路径	ColumnFamily、读写路径、增量 checkpoint
第 13 篇 · 状态放大与调优	窗口膨胀、写放大、hot key、设计 vs 调参
第 14 篇 · 交付语义	Source / 引擎 / Sink 三层语义

版本锚定：Flink 1.20+ / 2.x；RocksDB 随 flink-statebackend-rocksdb 捆绑。调参 API 以 EmbeddedRocksDBStateBackend、RocksDBOptionsFactory 为准。

环境说明：本机为 Arch Linux on WSL2（内核 6.6.87.2），未安装 JVM / Flink 集群。机制来自 Flink 官方文档、RocksDB Wiki Tuning Guide 与 flink-statebackend-rocksdb 源码；不粘贴未执行的 Web UI 截图，也不伪造 checkpoint 耗时或 live-sst-files 曲线数字。文末给出可复现实验步骤。

一、状态放大的三个来源

1.1 逻辑条目数 ≠ 磁盘占用

第 9 篇 给出窗口 state 条目估算：

\[ N_{\text{state}} \approx N_{\text{key}} \times N_{\text{window/key}} \]

对 RocksDB 后端，每条 entry 还要乘以 序列化开销 与 LSM 多版本 / 未 compaction 的 tombstone：

\[ \text{磁盘占用} \gtrsim N_{\text{state}} \times (\|key\| + \|value\| + \text{overhead}) \times (1 + WA_{\text{factor}}) \]

其中 \(WA_{\text{factor}}\) 来自 lsm-tree 第 4 篇 Compaction 的 Leveled 写放大：逻辑写入 1 GB，Compaction 可能额外写数十 GB。Web UI 里的「状态大小」与 TM 上 du -sh 的 RocksDB 目录 often 差一个数量级——这不是 bug，是 LSM 账单。

1.2 窗口 state 膨胀

典型模式：keyBy(userId) + 滑动窗口（例如 1 小时窗口、5 分钟 slide）+ ProcessWindowFunction 保留明细 ListState。

第 3 篇 的 Session 窗口在 gap 内合并，活跃 session 数随 key 波动；高基数 key（百万级 userId）× 长 session gap → RocksDB iterator 扫描成本上升（第 12 篇 读路径）。

设计层修复（优先于调参）：

• 用 AggregatingState / ReducingState 代替 ListState 存全量事件；
• 缩短 slide 或改 滚动窗口 + 下游批补；
• 在 WindowFunction 触发后显式 clear；配合 State TTL（第 9 篇）兜底。

1.3 State TTL 与 Compaction 清理不同步

启用 StateTtlConfig 且 cleanupInRocksdbCompactFilter=true 时，过期 entry 在 Compaction 中过滤（Flink 文档 State TTL）。这意味着：

• TTL 到期的 key 不会立刻 从磁盘消失；
• Compaction 跟不上写入时，live-sst-files 与 tombstone 堆积，读放大上升；
• checkpoint 前 强制 Flush（增量 checkpoint 路径）把 MemTable 里的「已逻辑过期但未 compact」条目也落盘——同步阶段变长。

TTL 是 最终清理，不是 实时 cap。若业务要求「状态不能超过 X GB」，必须 限制 key 基数或窗口覆盖范围，不能单靠 TTL。

1.4 MapState / ListState 无界增长

反模式：MapState 存「每个 user 见过的所有 productId」且无 TTL、无窗口边界。Keyed state 按 KeyGroup 分片，单 key 的 Map 再大也落在某一个 subtask 的一个 RocksDB 实例里——这是 hot key 问题的温床（第四节）。

二、写放大、Compaction 与 Checkpoint 的 I/O 争抢

2.1 写路径回顾与放大公式

第 12 篇 与 lsm-tree 全景 一致：前台 Put → WAL + MemTable；后台 Flush → L0 SST；Leveled Compaction → L1…Ln。

Leveled 策略的写放大近似：

\[ WA \approx T \times L \]

\(T\) 为层间容量倍数（RocksDB 默认 10），\(L\) 为层数。Flink 作业 持续更新 同一批 hot key 时，Compaction 反复归并相同 key 范围，写放大感知更明显——磁盘带宽被 Compaction 吃满后，前台 Put stall（RocksDB stall-micros 上升），反压经算子链上传（第 18 篇）。

2.2 增量 Checkpoint 叠加 Flush

增量 checkpoint 在 snapshot 前 同步 Flush MemTable（Flink 官方博客 Incremental Checkpointing；第 12 篇）。时间线上：

sequenceDiagram
  participant OP as 算子写入
  participant R as RocksDB
  participant CP as Checkpoint
  OP->>R: 持续 Put
  R->>R: 后台 Compaction
  CP->>R: 同步 Flush（barrier 对齐后）
  R->>R: 新 L0 SST
  CP->>CP: diff SST 上传
  Note over R,CP: Flush + Compaction + 上传<br/>共享 TM 磁盘与网络

症状对照：

调短 checkpoint interval 不会 降低 Compaction 压力，反而 增加 Flush 与上传频率。第 10 篇 的 min-pause-between-checkpoints 在此有实际意义：给 Compaction 留呼吸窗口。

2.3 与 HashMap 后端的对比边界

状态小于 JVM 堆安全阈值、更新模式偏 读多写少 时，HashMap 后端避免 Compaction 税（第 9 篇）。一旦：

• 状态超过 managed heap 或频繁 Full GC；
• checkpoint 全量序列化体积超过网络容忍；

就应接受 RocksDB 的 LSM 账单，并在 state 设计 + Compaction 预算 上投入，而不是指望「调大堆」一劳永逸。

三、RocksDB 与 Flink Managed Memory 调参

3.1 Managed Memory 划分

Flink 默认 state.backend.rocksdb.memory.managed=true（文档 Memory Management）：

同一 slot 内 多个算子、多个 RocksDB 实例共享 该预算（第 12 篇）。一个算子 Compaction 风暴会挤占同 slot 其他算子的 block cache——大状态混部时用 SlotSharingGroup 隔离（第 7 篇）。

3.2 RocksDBOptionsFactory 常用项

通过 state.backend.rocksdb.options-factory 或代码注入 RocksDBOptionsFactory（Flink 文档 Advanced RocksDB State Backend Options）：

public class FlinkRocksDBOptionsFactory implements ConfigurableRocksDBOptionsFactory {
    @Override
    public DBOptions createDBOptions(DBOptions currentOptions, ColumnFamilyOptions columnOptions) {
        return currentOptions
            .setMaxBackgroundJobs(4)
            .setMaxSubcompactions(2);
    }

    @Override
    public ColumnFamilyOptions createColumnOptions(
            ColumnFamilyOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
        return currentOptions
            .setWriteBufferSize(64 * 1024 * 1024L)
            .setMaxWriteBufferNumber(3)
            .setMinWriteBufferNumberToMerge(2);
    }
}

原则：在 managed memory 总预算固定时，write buffer 与 block cache 是零和。读热点明显（维表 join、频繁 value()）→ 略增 cache 比例；写密集窗口聚合 → 保证足够 write buffer 并控制 max_background_jobs 不超过磁盘 IOPS 上限。

3.3 Native Metrics 诊断

启用 state.backend.rocksdb.metrics.enable=true 后，选择性打开（文档列出的 metric 名）：

• rocksdb.num-immutable-mem-table — MemTable 排队；
• rocksdb.stall-micros — 写 stall 累计；
• rocksdb.live-sst-files-size — live SST 体积；
• rocksdb.compaction-pending — Compaction 积压；
• rocksdb.block-cache-usage — cache 压力。

Flink Web UI 单个 subtask 的 metrics 若显著高于同算子其他 subtask，优先怀疑 KeyGroup 倾斜（第四节），而非全局 RocksDB 参数。

3.4 Changelog State Backend（边界）

Flink Changelog state backend（FLIP-158）在 RocksDB 之上追加 changelog 以缩短 checkpoint 同步时间——改变的是 快照频率与上传路径，不消除 Compaction。生产默认仍以 EmbeddedRocksDB 为主；若 syncDuration 长期占 checkpoint 总时长大部分，可评估 changelog，但 state 条目数膨胀 仍需从设计层解决。

3.5 Timer Service 与 RocksDB 的叠加

Keyed 算子的 processing-time / event-time timer 默认也写入 RocksDB（state.backend.rocksdb.timer-service.factory=rocksdb）。Timer 条目与 keyed state 共用 同一 RocksDB 实例的 CF 与 I/O 预算，但 key 布局不同（timer 按 timestamp 排序）。

排查 checkpoint 体积异常时，除 window state 外应查看 registered timer 数量（Flink metrics numRegisteredEventTimeTimers / numRegisteredProcessingTimeTimers per subtask）。

3.6 flink-conf.yaml 调优片段（RocksDB 相关）

下列配置与代码中的 EmbeddedRocksDBStateBackend(true) 配合使用；键名以 Flink 1.20 文档为准，2.x 若有迁移以发行说明为准。

state.backend.type: rocksdb
execution.checkpointing.incremental: true
state.backend.rocksdb.localdir: /data/flink/rocksdb

# Managed memory（与 TM process size 联动）
state.backend.rocksdb.memory.managed: true
state.backend.rocksdb.memory.write-buffer-ratio: 0.5
state.backend.rocksdb.memory.high-prio-pool-ratio: 0.1

# 诊断
state.backend.rocksdb.metrics.enable: true
state.backend.rocksdb.metrics.block-cache-usage: true
state.backend.rocksdb.metrics.num-immutable-mem-table: true
state.backend.rocksdb.metrics.stall-micros: true
state.backend.rocksdb.metrics.compaction-pending: true

# Checkpoint 与 Compaction 错峰
execution.checkpointing.interval: 120000
execution.checkpointing.min-pause-between-checkpoints: 60000
execution.checkpointing.timeout: 600000

注意：localdir 必须落在 本地 SSD；与 checkpoint 存储（S3/HDFS）分离。把 RocksDB 目录放在网络盘会放大 第 12 篇 读路径延迟，且 Compaction 与远程 checkpoint 上传 争抢网络。

3.7 远程 checkpoint 存储上的「状态体积」误读

增量 checkpoint 下，Flink Web UI 的 Checkpointed Data Size 是 当次 delta（第 12 篇）。运维容量规划还需：

\[ \text{远程占用} \approx \text{最新 completed checkpoint 引用的 SST 闭包} \]

Compaction 合并 SST 后，旧文件引用被 subsumed，远程体积可能 下降；若 state 逻辑只增不减，引用闭包仍单调涨。不能 用单次 delta 推断总占用，也不能用 TM 本地 du 直接等同远程账单——本地还有 WAL、未上传 SST、Compaction 临时文件。

四、Hot Key 倾斜与 Subtask 负载不均

4.1 KeyGroup 不等于均匀

keyBy 把 key 哈希到 maxParallelism 个 KeyGroup（第 8 篇）。均匀哈希保证 key 空间期望均匀，不保证业务热点均匀：

• 电商大促：少数 merchantId 占 80% 流量；
• CDC 分库分表：未 加盐 的 userId 在单分片堆积；
• 默认 keyBy 字段选错（低基数 countryCode）。

每个 subtask 负责 连续 KeyGroup 区间。倾斜时，一个 subtask 的 RocksDB 实例承担过多更新与 Compaction，表现为：

• 该 subtask busy、同算子其他 subtask idle；
• 该 TM 磁盘与 CPU 飙高；
• checkpoint 最慢 subtask 决定全局 ack 时间（第 10 篇）。

flowchart LR
  subgraph Keys["业务 key 流量"]
    H1["hot key 40%"]
    H2["hot key 35%"]
    T["长尾 25%"]
  end
  subgraph KG["KeyGroup 映射"]
    ST2["subtask 2<br/>RocksDB 过热"]
    ST0["subtask 0"]
    ST1["subtask 1"]
  end
  H1 --> ST2
  H2 --> ST2
  T --> ST0
  T --> ST1

4.2 加并行度为什么常常无效

把 parallelism 从 8 提到 32 不会 拆分单个 hot key 的 state——同一 key 永远落在 唯一 KeyGroup。并行度只切 KeyGroup 区间，不切单 key。有效手段：

第 18 篇 的倾斜诊断清单与此一致：先看 subtask 级 recordsIn 与 RocksDB metrics，再决定是否 rescale。

4.3 rescale 与 savepoint

从 savepoint 恢复并 提高 maxParallelism 可重划 KeyGroup（第 11 篇），但若热点 key 不变，新 subtask 仍会接住该 key。Rescale 解决的是 算力不够，不是 单 key 状态过大。

4.4 诊断步骤（REST / Web UI）

1. Flink Web UI → Metrics → 算子 → subtask：比较 numRecordsInPerSecond、busyTimeMsPerSecond；
2. RocksDB native metrics（第三节）：对比同算子各 subtask 的 live-sst-files-size、stall-micros；
3. Checkpoint → History：最慢 subtask 是否长期为同一 index（sync_duration 瓶颈）；
4. 源端：Kafka partition lag 是否仅对应 busy subtask 消费的 partition 子集——若 source 并行度与 keyBy 不一致，需先排除 source 倾斜 再判 keyed 倾斜。

若仅 一个 subtask 三项均异常，几乎可断定 hot key 或 KeyGroup 分配不均；若 全部 subtask RocksDB 指标同步恶化，回到 全局 state 膨胀或 Compaction 预算（第一、二节）。

4.5 两阶段预聚合示例（思路）

对 keyBy(userId) 全局 UV 统计，热点 userId 可把单 subtask 压垮。可拆为：

flowchart LR
  IN[输入] --> K1["keyBy(hash(userId)%S)<br/>局部 AggregatingState"]
  K1 --> SH[shuffle]
  SH --> K2["keyBy(userId)<br/>全局合并"]
  K2 --> OUT[输出]

第一阶段 \(S\) 个 salt bucket 把同一 userId 的更新分散到 \(S\) 个 subtask；第二阶段 state 仅存 每 user 一条聚合值，不再存明细 ListState。代价是 多一次 shuffle 与 略高 end-to-end 延迟——这是用网络换磁盘与 Compaction 的典型交易。

五、何时改 State 设计，何时拧 RocksDB 参数

5.1 决策表

经验边界：若 state 条目数估算错误一个数量级，任何 RocksDB 调参都只是延缓 OOM。先画 \(N_{\text{key}} \times N_{\text{window/key}}\)（第 9 篇），再动 write_buffer_size。

5.2 与 lsm-tree 系列的对照

深入 Compaction 算法与 Version 见 lsm-tree Compaction 篇；SSTable 布局见 SSTable + Bloom Filter 篇。

六、可复现实验（不提供伪造数字）

环境建议（与 系列 PLAN 实验台账一致）：

• Flink 1.20+，单 JM + 2 TM，local SSD；
• 作业：keyBy + 1 小时滑动窗口（5 分钟 slide）+ ListState 存 String payload（刻意制造膨胀）；
• 对比组 A：改为 AggregatingState 计数；
• 对比组 B：构造 99% 流量进 1% key 的 skew 源；
• checkpoint 60s，RocksDB 增量开启，state.backend.rocksdb.metrics.enable=true。

观测项（≥3 次 checkpoint 取中位数趋势，本文不填具体数值）：

1. TM 上 state.backend.rocksdb.localdir 的 du -sh；
2. REST API：/jobs/:id/checkpoints 的 sync_duration、checkpointed_size；
3. 单 subtask rocksdb.live-sst-files-size、stall-micros；
4. 组 B 中 busy subtask 的 numRecordsInPerSecond vs 其他 subtask。

预期定性结论（机制推导）：

• 组 A 相对 ListState 版，磁盘与 checkpoint 体积应 显著下降；
• 组 B 中 一个 subtask 的 RocksDB metrics 独高，全局 checkpoint 时长跟随最慢 subtask；
• 仅调 max_background_jobs 不能消除组 B 的 skew，预聚合或改 key 才触及根因。

6.1 生产告警阈值（建议口径，非硬编码）

下列为 运维启发式，须按集群基线校准；本文不给未实测的绝对毫秒数。

七、边界与后文

本文 不 重复 第 12 篇 的 ColumnFamily 布局与增量 snapshot 算法，也 不 展开 第 14 篇 的端到端 exactly-once。

未覆盖：

• ForSt 远程 SST 后端的容量规划；
• RocksDBOptionsFactory 全参数枚举（RocksDB Wiki 为准）；
• 背压传播链的完整诊断（第 18 篇）。

下一篇 第 14 篇 把视角从「状态 bytes」转到「记录是否重复、是否丢失」：Source、引擎、Sink 三层交付语义如何组合，以及 at-least-once 与 exactly-once 的边界。

参考资料

1. Apache Flink Documentation, State Backends、RocksDB State Backend、Memory Management、State TTL、Monitoring（RocksDB native metrics）。A 级。
2. Apache Flink 源码：flink-statebackend-rocksdb — EmbeddedRocksDBStateBackend、RocksDBOptionsFactory（release-1.20 / release-2.x）。A 级。
3. RocksDB Wiki, Tuning Guide、Compaction、Write Stalls。A 级。
4. Apache Flink 官方博客，Managing Large State in Apache Flink: An Intro to Incremental Checkpointing（2018）。B 级。
5. 本仓库 lsm-tree 系列：全景、Compaction。机制对照。
6. 本系列：第 9 篇 键控状态、第 10 篇 Checkpoint、第 12 篇 RocksDB 路径。

返回 系列目录 · 上一篇 RocksDB State Backend 内核路径 · 下一篇 交付语义：从 at-most-once 到 exactly-once