KeyedState 简介
KeyedState 是 Flink 中与 key 绑定的状态。每个 key 有自己的独立状态实例，类似于按 key 分区的 HashMap。
KeyedStream keyed by userId:
user=alice → ValueState<OrderCount>（alice 的计数）
user=bob → ValueState<OrderCount>（bob 的计数）
user=charlie → ValueState<OrderCount>（charlie 的计数）
KeyedState 只在 keyBy 之后的算子里可用：KeyedProcessFunction、KeyedStream 上的 RichFlatMapFunction、ProcessWindowFunction 等。判据是「当前记录有没有 key」——挂在 RichFunction 上并不够，非 keyed 流的算子拿不到 key，getState() 会直接抛异常。
ValueState
最简单的状态：每个 key 对应一个值。
public class OrderCounter extends KeyedProcessFunction<String, Event, Result> {
private ValueState<Long> countState;
@Override
public void open(OpenContext openContext) {
StateTtlConfig ttl = StateTtlConfig
.newBuilder(Duration.ofHours(24))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.build();
ValueStateDescriptor<Long> descriptor = new ValueStateDescriptor<>(
"order-count", Types.LONG);
descriptor.enableTimeToLive(ttl);
countState = getRuntimeContext().getState(descriptor);
}
@Override
public void processElement(Event event, Context ctx, Collector<Result> out)
throws Exception {
Long current = countState.value();
if (current == null) current = 0L;
countState.update(current + 1);
out.collect(new Result(ctx.getCurrentKey(), current + 1));
}
}
使用：
stream.keyBy(Event::getUserId)
.process(new OrderCounter());
ListState
每个 key 对应一个列表。
// 用独立的 OrderSummary 表示「某个用户在窗口内的订单金额」，
// 避免和第 2 章 KStream 示例里的 Order 模型混淆；
// 第 4 章的 Event 是 POJO，字段要通过 getUserId()/getAmount() 读取。
// record 没有无参构造器，Flink 不会把它识别成 POJO，会回退到 Kryo：
// 功能上可用，但序列化更慢、状态更大。状态里频繁读写的类型建议写成 POJO
record OrderSummary(String userId, double amount) {}
public class OrderCollector extends KeyedProcessFunction<String, Event, List<OrderSummary>> {
private ListState<OrderSummary> orderListState;
@Override
public void open(OpenContext openContext) {
ListStateDescriptor<OrderSummary> descriptor =
new ListStateDescriptor<>("orders", OrderSummary.class);
orderListState = getRuntimeContext().getListState(descriptor);
}
@Override
public void processElement(Event event, Context ctx, Collector<List<OrderSummary>> out)
throws Exception {
orderListState.add(new OrderSummary(event.getUserId(), event.getAmount()));
// 每收集 10 条输出一次
List<OrderSummary> orders = new ArrayList<>();
orderListState.get().forEach(orders::add);
if (orders.size() >= 10) {
out.collect(orders);
orderListState.clear();
}
}
}
MapState
每个 key 对应一个 Map。
private MapState<String, Double> symbolAmountState;
// 添加
symbolAmountState.put("BTC", 10000.0);
// 读取
Double amount = symbolAmountState.get("BTC");
// 遍历
for (Map.Entry<String, Double> entry : symbolAmountState.entries()) {
System.out.println(entry.getKey() + ": " + entry.getValue());
}
ReducingState
自动使用 reduce 函数聚合的元素列表，只保留聚合结果。
ReducingStateDescriptor<OrderSummary> descriptor = new ReducingStateDescriptor<>(
"total-amount",
(o1, o2) -> new OrderSummary(o1.userId(), o1.amount() + o2.amount()),
OrderSummary.class);
ReducingState<OrderSummary> reducingState = getRuntimeContext().getReducingState(descriptor);
AggregatingState
类似 ReducingState，但输入类型和输出类型可以不同。
AggregatingStateDescriptor<Event, AverageAccumulator, Double> descriptor =
new AggregatingStateDescriptor<>(
"avg-amount",
new AggregateFunction<Event, AverageAccumulator, Double>() {
public AverageAccumulator createAccumulator() { return new AverageAccumulator(0, 0); }
public AverageAccumulator add(Event value, AverageAccumulator acc) {
return new AverageAccumulator(acc.sum + value.getAmount(), acc.count + 1);
}
public Double getResult(AverageAccumulator acc) {
return acc.count == 0 ? 0.0 : acc.sum / acc.count;
}
public AverageAccumulator merge(AverageAccumulator a, AverageAccumulator b) {
return new AverageAccumulator(a.sum + b.sum, a.count + b.count);
}
},
AverageAccumulator.class);
StateTtlConfig
状态不能无限增长，配置 TTL：
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Duration.ofHours(24))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.cleanupIncrementally(10, true) // 增量清理
.build();
| 参数 | 选项 | 说明 |
|---|---|---|
UpdateType | OnCreateAndWrite | 每次写入/更新时刷新 TTL |
OnReadAndWrite | 读写都刷新 | |
StateVisibility | NeverReturnExpired | 不返回过期值（可能空） |
ReturnExpiredIfNotCleanedUp | 返回未清理的过期值 | |
cleanup | cleanupFullSnapshot | Checkpoint 时全量清理 |
cleanupIncrementally | 增量清理 |
RocksDB 状态后端
大状态场景使用 RocksDB：
env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoints");
RocksDB 配置：
// 直接声明成具体类型：EmbeddedRocksDBStateBackend 实现的是
// ConfigurableStateBackend，不存在 DefaultConfigurableStateBackend 这个类型
EmbeddedRocksDBStateBackend stateBackend = new EmbeddedRocksDBStateBackend();
stateBackend.setPredefinedOptions(PredefinedOptions.FLASH_SSD_OPTIMIZED);
env.setStateBackend(stateBackend);
RocksDB 的内存不是「随便给多少」：开启 state.backend.rocksdb.memory.managed（默认 true）后，block cache、write buffer 等都从 TaskManager 的 managed memory 里按比例分配，调大 RocksDB 只能先调大 managed memory。
| 指标 | 怎么读 |
|---|---|
rocksdb.block-cache-hit / block-cache-miss | 两个都是累计计数，命中率要自己算 hit / (hit + miss)；长期偏低说明热点数据大于 block cache |
rocksdb.compaction-pending | 持续为 1 → compaction 跟不上写入，读放大升高 |
rocksdb.num-running-compactions | 长期打满 compaction 线程数 → 磁盘带宽或 CPU 是瓶颈 |
rocksdb.estimate-live-data-size | 持续上涨且不回落 → 状态在膨胀，先查 TTL 和 key 设计 |
小结
| State 类型 | 结构 | 适用 |
|---|---|---|
| ValueState | key → value | 最新值、计数器 |
| ListState | key → List | 收集元素、列表 |
| MapState | key → Map | key-value 映射 |
| ReducingState | key → reduce 结果 | 持续聚合 |
| AggregatingState | key → aggregate 结果 | 输入输出类型不同的聚合 |
下一节 5.2 Flink 状态管理：OperatorState 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。