Flink 支持通过 table.exec.state.ttl 全局配置为 RocksDB 状态设置生存时间（TTL），自动清理超时数据；但该配置作用于整个作业，不支持按表粒度单独设定。
flink 支持通过 `table.exec.state.ttl` 全局配置为 rocksdb 状态设置生存时间（ttl），自动清理超时数据；但该配置作用于整个作业，不支持按表粒度单独设定。
在 Flink 流处理中，RocksDB StateBackend 默认不会主动删除旧状态数据——它仅负责持久化和高效访问，不提供基于时间的自动清理能力。若需实现“仅保留最近 2 小时订单记录”这类业务需求（例如用于实时比对、滑动参考等场景），不能依赖 RocksDB 自身机制，而必须借助 Flink 的 State TTL（Time-To-Live）功能。
✅ 正确做法：启用全局 State TTL
State TTL 是 Flink 内置的状态生命周期管理机制，适用于所有状态后端（包括 RocksDB）。它会在状态访问或定期后台检查时，自动剔除已过期的条目。要为本例中的 Orders 表启用 2 小时 TTL，需在创建 StreamTableEnvironment 后、执行 SQL 前，通过 TableConfig 设置全局 TTL：
val tableEnvironment = StreamTableEnvironment.create(environment)
// ✅ 启用全局状态 TTL：2 小时（单位：毫秒）
tableEnvironment.config.set(
"table.exec.state.ttl", "7200000" // 2 * 60 * 60 * 1000
)⚠️ 注意：
table.exec.state.ttl是作业级配置，对当前TableEnvironment中所有注册表（如Orders）的状态均生效，无法在CREATE TABLE语句中按表指定（Flink 当前版本不支持 per-table TTL）。
? 补充建议：优先使用语义驱动的自动清理
更推荐的做法是避免手动管理状态生命周期，而是将时间约束融入查询逻辑本身。Flink SQL 在识别到明确的时间语义（如窗口、事件时间 JOIN、MATCH_RECOGNIZE 等）时，会自动推导并清理无用状态，无需显式配置 TTL。
例如，若你的比对逻辑本质是“查找过去 2 小时内同用户的订单”，可改写为基于事件时间的 Interval Join：
CREATE TABLE Orders ( user BIGINT, product STRING, amount INT, ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL '5' SECOND ) WITH ( /* your connector config */ ); -- 自动维护 2 小时窗口状态，超时后 Flink 自行清理 SELECT o1.user, o1.product, o2.product AS prev_product FROM Orders AS o1 JOIN Orders AS o2 ON o1.user = o2.user AND o2.ts BETWEEN o1.ts - INTERVAL '2' HOUR AND o1.ts - INTERVAL '1' SECOND;
此类查询由 Flink 优化器自动管理状态 TTL，兼具语义清晰性与资源效率。
? 总结
- ✅ 可通过
table.exec.state.ttl全局启用 RocksDB 状态自动过期（如"7200000"毫秒）； - ❌ 不支持在
CREATE TABLE ... WITH (...)中为单表声明 TTL； - ? 优先采用带时间语义的 SQL（窗口、interval join、over 窗口聚合等），让 Flink 自动推导并清理状态；
- ? 显式 TTL 配置适用于无法重构为语义化查询的遗留逻辑，但需注意其作用域为整个作业，可能影响其他表状态。
