📚 课程目录第2章 State 的概念与分类

State 的概念与分类：Keyed State 与 Operator State 的区别，Managed State 与 Raw State 的区别

好，咱们正式开始聊 State。很多初学者一上来就被各种 State 名词搞晕了——Keyed State、Operator State、Managed State、Raw State……说实话，我当年刚接触 Flink 时也懵过一阵子。但搞清楚这些分类，是选对 State Backend 的前提。

你想想看，State 说白了就是 Flink 算子需要记住的「中间结果」或「历史数据」。没有 State，流处理就只能做无状态计算，比如简单的 map、filter。但现实业务中，你需要做聚合、做窗口、做去重、做关联——这些统统离不开 State。

核心观点：State 的分类维度有两个——按「作用域」分 Keyed State 和 Operator State；按「管理方式」分 Managed State 和 Raw State。这两组分类是正交的，可以组合使用。

Keyed State 与 Operator State 的区别

先聊 Keyed State。这是 Flink 里最常用的 State 类型。它的核心逻辑是：每个 key 维护一份独立的 State。你只要在 DataStream 上做了 keyBy()，后续算子就能用 Keyed State。

举个例子，统计每个用户的点击次数。用户 ID 就是 key，每个用户对应一个计数器。Flink 会自动帮你把相同 key 的数据路由到同一个 subtask 上，State 也按 key 隔离。

我的习惯：只要业务逻辑里出现了「按某字段分组统计」的场景，我第一反应就是用 Keyed State。它天然支持分区、支持快照、支持增量 checkpoint，是 Flink 最成熟的状态模型。

Operator State 就不一样了。它的作用域是「每个并行子任务实例」。同一个 subtask 处理的所有数据共享一份 State。典型的场景是 Kafka 连接器的 offset 记录——每个 subtask 只需要记住自己消费的分区 offset 就行，不需要按 key 拆分。

我曾在项目中用 Operator State 实现过一个自定义的「数据采样器」。每个 subtask 维护一个固定大小的缓冲区，缓存最近 1000 条数据，用于实时调试。这种场景用 Keyed State 反而别扭，因为采样逻辑跟 key 无关。

对比维度	Keyed State	Operator State
作用域	每个 key 一份	每个算子实例一份
前提条件	必须先做 keyBy()	无需 keyBy，任何算子可用
支持的数据结构	ValueState、ListState、MapState、ReducingState、AggregatingState	ListState、UnionListState、BroadcastState
重分配策略	按 key 哈希，自动均匀分配	支持均匀分配（Even-split）或联合分配（Union）
典型场景	按用户/设备/订单聚合统计	Kafka offset、自定义 Source 状态

避坑指南：我曾经在项目里犯过一个低级错误——在非 keyed stream 上直接调用了 ValueState。编译能过，运行直接抛异常。记住：Keyed State 只能在 KeyedStream 上使用，也就是必须经过 keyBy() 之后才行。

Managed State 与 Raw State 的区别

这个分类维度关注的是「谁负责管理 State 的生命周期」。

Managed State 是 Flink 帮你管理的。你只需要声明 State 描述符（StateDescriptor），Flink 自动负责序列化、反序列化、checkpoint 快照、恢复。我平时 99% 的场景都用 Managed State——省心、安全、不容易出错。

Raw State 就不一样了。你得自己管理字节数组，自己处理序列化逻辑，自己实现 checkpoint 的快照和恢复接口。说白了，Flink 只给你一个 ListState<byte[]> 的容器，里面的数据怎么组织、怎么读写，全看你自己的本事。

嗯，这里要注意：Raw State 在 Flink 1.x 时代还有一些使用场景，比如自定义算子需要极致性能优化时。但到了 Flink 2.x，Managed State 的性能已经非常接近 Raw State 了，而且 TypeInformation 的序列化效率很高。我个人建议：除非你有非常特殊的序列化需求，否则一律用 Managed State。

对比维度	Managed State	Raw State
管理方式	Flink 自动管理	用户自行管理
数据结构	ValueState、ListState、MapState 等	仅 byte[]
序列化	Flink 自动使用 TypeInformation	用户自定义序列化
Checkpoint	自动快照和恢复	需实现 CheckpointedFunction
使用难度	低，声明式 API	高，需处理底层细节
推荐度	⭐⭐⭐⭐⭐ 首选	⭐⭐ 仅特殊场景

关键结论：Managed State + Keyed State 是 Flink 开发中最主流的组合。Operator State 主要用于 Source/Sink 连接器。Raw State 除非你很清楚自己在做什么，否则别碰。

代码示例：Keyed Managed State 的典型用法

光说不练假把式。我写个最简单的 ValueState 示例，统计每个用户的点击次数：

public class ClickCountFunction extends RichFlatMapFunction<ClickEvent, String> {
    
    // 声明 Managed Keyed State
    private transient ValueState<Long> countState;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 创建 State 描述符
        ValueStateDescriptor<Long> descriptor = 
            new ValueStateDescriptor<>("click-count", Types.LONG);
        
        // 获取 State
        countState = getRuntimeContext().getState(descriptor);
    }
    
    @Override
    public void flatMap(ClickEvent event, Collector<String> out) throws Exception {
        // 读取当前值
        Long currentCount = countState.value();
        if (currentCount == null) {
            currentCount = 0L;
        }
        
        // 更新
        currentCount += 1;
        countState.update(currentCount);
        
        out.collect("User " + event.userId + " clicked " + currentCount + " times");
    }
}

你看，代码非常简洁。Flink 帮你处理了 State 的序列化、checkpoint、故障恢复。你只需要关注业务逻辑。

我的经验：State 描述符的 name 字段最好用有业务含义的字符串，比如 "user-click-count"。别用 "state-1" 这种名字。为什么？因为 Flink 的 savepoint 恢复时是按 name 匹配的。名字起得好，恢复没烦恼。我曾经接手过一个项目，里面全是 "state-0" 到 "state-9"，改个需求差点没把我逼疯。

Operator State 代码示例：实现一个简单的 offset 记录器

public class OffsetOperator extends RichParallelSourceFunction<String> 
        implements CheckpointedFunction {
    
    // Operator State：每个 subtask 维护自己的 offset 列表
    private transient ListState<Long> offsetState;
    private Long currentOffset = 0L;
    
    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        ListStateDescriptor<Long> descriptor = 
            new ListStateDescriptor<>("offset-state", Types.LONG);
        
        offsetState = context.getOperatorStateStore().getListState(descriptor);
        
        // 从 checkpoint 恢复
        if (context.isRestored()) {
            for (Long offset : offsetState.get()) {
                currentOffset = offset;
            }
        }
    }
    
    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        // 快照时清空并写入当前 offset
        offsetState.clear();
        offsetState.add(currentOffset);
    }
    
    // ... run() 和 cancel() 省略
}

注意到区别了吗？Operator State 是通过 getOperatorStateStore() 获取的，而 Keyed State 是通过 getRuntimeContext().getState()。这个 API 差异背后是两种完全不同的状态模型。

曾经踩过的坑：有一次我在 Operator State 里用了 getUnionListState()，结果并行度从 4 改成 8 后，状态恢复时每个 subtask 拿到了所有分区的 offset，导致数据重复消费。后来才搞清楚：UnionListState 会把所有 subtask 的状态广播给每个新 subtask，适合需要「全量恢复」的场景；而普通 ListState 采用 even-split 策略，每个 subtask 只拿一部分。选型时一定要想清楚你的恢复语义。

总结一下

State 的分类其实不复杂。记住两条线：

Keyed State vs Operator State：看你的状态是按 key 隔离，还是按算子实例共享。
Managed State vs Raw State：看你想让 Flink 帮你管，还是自己撸起袖子干。

实际项目中，Managed Keyed State 是的主力。Operator State 主要用在 Source/Sink 连接器里。Raw State 嘛……我上次用 Raw State 还是三年前做自定义序列化优化的时候，后来发现 Managed State 的性能已经够用了，就再也没碰过。