java

11.3 状态 TTL 与过期清理

Flink 状态过期与清理策略：StateTtlConfig 的三种 UpdateType 和两种 StateVisibility、过期数据的清理时机（增量/全量/RocksDB 压缩）、Idle State Retention 的窗口级清理，以及 eventd v3 各状态存储的 TTL 配置方案。

Leeting Yan 4 分钟阅读 1953 字

状态为什么需要过期

流处理作业长期运行时，状态持续增长：

去重状态：eventId → true
    → 无限增长 → RocksDB SST 文件膨胀 → GC 延迟

窗口状态：key → List<Event>
    → 窗口关闭后未清理 → 内存泄漏

用户资料：userId → UserProfile
    → 离职用户状态永不删除 → 状态过大

TTL（Time To Live）自动清理过期状态，防止状态无限膨胀。

StateTtlConfig 基础

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Duration.ofHours(24))  // 24 小时过期（1.20 推荐 Duration 重载）
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupFullSnapshot()  // Checkpoint 时全量清理
    .build();
参数选项说明
UpdateTypeDisabled不刷新 TTL，写入时的过期时间就是创建时间 + TTL
OnCreateAndWrite创建/写入时刷新 TTL
OnReadAndWrite读/写都刷新 TTL
StateVisibilityNeverReturnExpired不返回过期值（读到的可能是 null）
ReturnExpiredIfNotCleanedUp返回未清理的过期值

UpdateType 对比

OnCreateAndWrite（默认）
    写操作: put("alice", count) → TTL 刷新
    读操作: get("alice") → TTL 不变
    → 适合: 只关心最近活跃的状态

OnReadAndWrite
    写操作: put → TTL 刷新
    读操作: get → TTL 也刷新
    → 适合: 热点数据需要「持续存活」

eventd v3 的去重状态用 OnCreateAndWrite：eventId 写一次就不会再访问。

eventd v3 的用户资料用 OnReadAndWrite：活跃用户频繁读取 → 延长存活。

TTL 的计时基准只有处理时间（1.20 的 TtlTimeCharacteristic 只剩 ProcessingTime 一个值），与作业是否使用事件时间无关。所以重放历史数据时，状态可能因为「处理时钟走到了」而提前过期——TTL 解决的是状态膨胀，不是「按事件时间保留 N 天」。

清理策略

全量清理（cleanupFullSnapshot）

在每次 Checkpoint 时，对整个状态做一次全量扫描，删除过期数据。

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Duration.ofHours(24))
    .cleanupFullSnapshot()
    .build();

优点：彻底。
缺点：Checkpoint 时间增加（需遍历所有 key），且只处理「快照时已过期」的条目，两次 Checkpoint 之间过期的不清理。

增量清理（cleanupIncrementally）

状态被访问时，顺带遍历一部分条目并删除已过期的：

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Duration.ofHours(24))
    // 每次触发遍历 10 个条目；第二个参数表示每条记录都触发一次清理
    .cleanupIncrementally(10, true)
    .build();

参数：

  • cleanupSize：每次触发时最多遍历多少个状态条目（不是「删除 10 条」，遍历到的条目里只有过期的才删）
  • runCleanupForEveryRecord：true = 每处理一条记录就触发一次；false = 仅在状态被访问时触发

优点：Checkpoint 不受影响。
缺点：清理不完全——从未被再次访问的 key 不会被遍历到，仍然占用空间。

RocksDB 压缩清理（cleanupInRocksdbCompactFilter）

利用 RocksDB 的后台 Compaction 过程清理过期数据：

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Duration.ofHours(24))
    // 压缩过滤器中累计处理 1000 个条目后刷新一次「当前时间」用于判断过期
    .cleanupInRocksdbCompactFilter(1000)
    .build();

优点：不占用 Stream Task 线程，清理随 Compaction 在后台完成。
缺点：依赖 RocksDB Compaction 频率，冷数据可能很久才被压缩到，过期条目会在磁盘上多留一段时间。

这个选项只对 RocksDB 状态后端生效；换成 HashMapStateBackend 时该配置被忽略（启动日志里会有提示），不会报错但也不会清理，容易误以为已经生效。eventd v3 的大状态作业固定使用 RocksDB，才采用此策略。

eventd v3 的 TTL 配置

public class AnalyticsJob {

    private static StateTtlConfig dedupTtl() {
        return StateTtlConfig
            .newBuilder(Duration.ofHours(168))  // 7 天（覆盖业务重复窗口）
            .setUpdateType(UpdateType.OnCreateAndWrite)
            .setStateVisibility(StateVisibility.NeverReturnExpired)
            .cleanupInRocksdbCompactFilter(1000)
            .build();
    }

    private static StateTtlConfig sessionTtl() {
        return StateTtlConfig
            .newBuilder(Duration.ofHours(4))  // 4 小时会话超时
            .setUpdateType(UpdateType.OnReadAndWrite)
            .setStateVisibility(StateVisibility.NeverReturnExpired)
            .cleanupFullSnapshot()
            .build();
    }

    private static StateTtlConfig profileTtl() {
        return StateTtlConfig
            .newBuilder(Duration.ofDays(30))  // 30 天用户资料
            .setUpdateType(UpdateType.OnReadAndWrite)
            .setStateVisibility(StateVisibility.NeverReturnExpired)
            .cleanupIncrementally(100, false)
            .build();
    }
}
状态类型TTL策略原因
去重（eventId）7 天OnCreateAndWrite + RocksDB Compact重复只发生在短期内
会话窗口4 小时OnReadAndWrite + Full Snapshot会话最多 4 小时
用户资料30 天OnReadAndWrite + Incremental用户活跃时保留
CEP 模式5 分钟CEP 内部管理within() 控制

窗口状态的过期

窗口内状态在窗口关闭后自动清理：

stream.keyBy(Event::userId)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .allowedLateness(Duration.ofSeconds(10))
    .sideOutputLateData(lateTag)
    .aggregate(new CountAggregate());

窗口状态生命周期：

窗口创建: 第一条元素被分配到该窗口时（不是 Watermark 到达时）
    ↓
窗口触发: Watermark 越过 window.maxTimestamp()
    ↓
allowedLateness: 在 windowEnd + allowedLateness 之前，迟到数据仍会触发更新
    ↓
窗口清理: 清理时间 = maxTimestamp + allowedLateness，之后状态被删除

allowedLateness 不阻止窗口清理，只是把清理点从 windowEnd 推迟到 maxTimestamp + allowedLateness。设置得过大会同时拉长状态存活时间和重复输出窗口。

Checkpoint 大小与 TTL

TTL 直接影响 Checkpoint 大小：

无 TTL: 状态持续增长 → Checkpoint 越来越大 → 恢复变慢
24h TTL: 状态稳定在「24h 活跃数据」范围内 → Checkpoint 大小稳定

eventd v3 的 Checkpoint 大小监控：

env.getCheckpointConfig().enableUnalignedCheckpoints();
// Checkpoint 大小不需要自己埋点：JobManager 已导出
// lastCheckpointSize / lastCheckpointFullSize（字节），
// 在 Prometheus 里按作业维度取，超过基线 20% 时告警。

手动清理状态

某些场景需要手动清理（如用户注销）：

// 前提：这个算子已经 keyBy(Event::userId)，
// 所以 clear() 只作用于当前这条事件所属的 key
private transient ValueState<Long> countState;
private transient ListState<Order> orderListState;

public void processElement(Event event, Context ctx, Collector<Result> out)
        throws Exception {
    if ("DELETE_ACCOUNT".equals(event.type())) {
        countState.clear();
        orderListState.clear();
    }
}

clear() 清的是当前 key 的状态，不是全作业的状态。如果删号事件和订单事件不在同一个 key（比如删号流按 eventId 分区），这里什么也清不掉，只能再把删号事件广播出去，让所有并行实例各自处理，或在离线侧按用户维度清理。

小结

清理策略时机开销适用
Full SnapshotCheckpoint 时高小状态
Incremental每次访问低中等状态
RocksDB CompactCompaction 时低（后台线程）大状态（推荐）
UpdateType刷新时机适用
OnCreateAndWrite创建/写入一次性状态（去重）
OnReadAndWrite读/写持续访问状态（用户资料）

第 11 章结束。下一节进入全链路集成：12.1 eventd v3 全链路集成 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南