Flink RocksDB状态后端能否清理Orders表中2小时以上的旧数据？

当然可以实现这个需求。Flink自带的**状态TTL（Time-To-Live）**机制，能自动清理状态里超过指定存活时间的记录，不管你用的是EmbeddedRocksDBStateBackend还是其他状态后端，都能生效。对于用来做数据对比的Orders状态表，只要在表定义里配置好状态TTL相关参数，就能让状态后端自动删掉存活超过2小时的旧数据，只保留最近2小时的记录。

对应的CREATE TABLE配置示例 #

假设你的Orders表是用于关联、聚合这类需要维护状态的场景，修改后的CREATE TABLE语句如下：

CREATE TABLE Orders (
    order_id STRING,
    user_id STRING,
    amount DOUBLE,
    create_time TIMESTAMP(3),
    -- 其他业务字段根据实际情况补充
    PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'kafka', -- 替换成你的实际连接器类型，比如upsert-kafka、hbase等
    'format' = 'json', -- 替换成你的实际数据格式，比如avro、csv等
    -- 状态TTL核心配置
    'state.ttl' = '2 h',
    'state.ttl.cleanup.strategy' = 'periodic, incremental',
    'state.ttl.cleanup.interval' = '10 min',
    -- 以下是连接器的其他必填配置，示例以Kafka为例
    'topic' = 'orders_topic',
    'properties.bootstrap.servers' = 'localhost:9092',
    'scan.startup.mode' = 'latest-offset'
);

参数说明 #

• 'state.ttl' = '2 h'：指定状态记录的存活时长为2小时，支持的时间单位包括s（秒）、min（分钟）、h（小时）、d（天）。
• 'state.ttl.cleanup.strategy'：过期状态的清理策略，可同时配置两种：
  • periodic：定期触发全量清理任务，适合批量清理过期数据。
  • incremental：在状态访问或数据处理过程中增量清理过期数据，降低批量清理带来的性能波动。
• 'state.ttl.cleanup.interval' = '10 min'：定期清理任务的执行间隔，默认是1分钟，可根据业务负载调整。

额外注意点 #

• 如果是基于Key的状态存储场景（比如按order_id维护状态），必须在表定义里指定PRIMARY KEY，状态TTL是基于Key来管理过期数据的。
• 默认状态TTL按**处理时间（Processing Time）**计算，如果要基于事件时间（比如订单的create_time），需要额外添加配置：'state.ttl.time-characteristic' = 'event_time'，同时确保表中配置了事件时间字段和对应的水位线（Watermark）。

内容的提问来源于stack exchange，提问作者overexchange