2. Flink 状态类型详解：Keyed State 与 Operator State 的区别与适用场景

状态管理是实时数仓的基石。说白了，没有状态，Flink 就是一个高级的 ETL 管道，做不了聚合、做不了去重、做不了窗口计算。

Flink 把状态分成两大类：Keyed State 和 Operator State。很多初学者容易混淆，觉得「不都是存数据吗？有啥区别？」。嗯，区别大了去了。我刚开始用 Flink 时也踩过坑，把 Operator State 当 Keyed State 用，结果数据全乱了。

今天咱们就把这两类状态彻底讲透。

2.1 核心区别：谁在管理状态？

先看一张图，直观感受一下两者的定位。

这张图把 Flink 的状态体系讲清楚了。左边是 Keyed State，右边是 Operator State。你想想看，它们最大的不同是什么？

Keyed State 是「按 Key 隔离」的。每个 Key 都有自己的状态副本。比如统计每个用户的 PV，用户 A 的状态不会影响用户 B。这就像每个人都有自己的储物柜，钥匙在自己手里。

Operator State 是「按并行实例隔离」的。每个并行子任务（Subtask）维护一份状态。比如我用 Kafka Source 读取数据，每个并行度负责消费几个分区，Source 的偏移量就是 Operator State。

2.2 Keyed State 详解

Keyed State 是实时数仓里用得最多的状态类型。我做过的大多数项目，90% 以上的状态都是 Keyed State。

2.2.1 前提条件

使用 Keyed State 之前，数据流必须先做 keyBy() 操作。这是硬性要求。为什么？因为 Flink 需要根据 Key 的哈希值把数据路由到对应的分区，每个分区只处理一部分 Key 的状态。

// 必须先 keyBy，才能使用 Keyed State
DataStream<UserEvent> keyedStream = inputStream
    .keyBy(UserEvent::getUserId);

// 然后在 ProcessFunction 中声明状态
keyedStream.process(new KeyedProcessFunction<String, UserEvent, Void>() {
    private ValueState<Long> lastTimestamp;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<Long> descriptor = 
            new ValueStateDescriptor<>("last-timestamp", Types.LONG);
        lastTimestamp = getRuntimeContext().getState(descriptor);
    }
});

2.2.2 五种状态原语

Flink 提供了五种 Keyed State 原语，我按使用频率排个序：

状态类型	数据结构	典型场景
ValueState	单值	记录用户最近一次登录时间
MapState	KV 映射	存储用户购物车商品明细
ListState	列表	缓存用户最近 100 条行为记录
ReducingState	聚合值	实时累加用户消费金额
AggregatingState	自定义聚合	计算用户平均消费金额

我个人习惯用 ValueState 和 MapState 最多。ValueState 简单直接，适合存单值。MapState 适合存「Key 多、更新频繁」的场景，比如用户画像标签。

2.2.3 适用场景

• 实时聚合：按用户、商品、店铺等维度做累加、计数、求平均
• 去重：记录已处理过的订单 ID，防止重复支付回调
• 窗口计算：窗口内部的状态管理，比如 Session Window 的合并
• 状态机：根据事件序列驱动状态流转，比如风控规则引擎

2.3 Operator State 详解

Operator State 用得少，但用对地方能解决大问题。我记得有一次做 Kafka 精确一次语义，就是靠 Operator State 保存分区偏移量。

2.3.1 三种形式

形式	说明	重分配策略
ListState	每个算子实例维护一个列表	均匀分配（Round-Robin）
UnionListState	每个算子实例维护一个列表	广播到所有实例（每个实例拿到全量）
BroadcastState	所有算子实例共享同一份状态	广播到所有实例

这里有个坑，我踩过。用 UnionListState 时，如果状态数据量很大，重分配时每个算子实例都会拿到全量数据，容易 OOM。我曾经在 64 个并行度的任务上用 UnionListState 存了 10GB 的偏移量，结果恢复时直接内存溢出。后来改成 ListState 才解决。

2.3.2 适用场景

• Kafka Source 偏移量管理：每个并行度记录自己消费的分区偏移量
• 自定义 Source/Sink：比如读取文件时记录已处理的行号
• 广播配置：用 BroadcastState 把规则配置广播到所有算子实例
• 全局计数器：统计整个作业处理了多少条数据

2.4 关键差异对比

我把两者的核心差异整理成一张表，方便你对照：

对比维度	Keyed State	Operator State
作用域	每个 Key 独立	每个算子实例独立
前置条件	必须 keyBy()	不需要 keyBy()
状态原语	5 种（Value/List/Map/Reducing/Aggregating）	3 种（List/UnionList/Broadcast）
重分配策略	按 Key 哈希自动重分配	手动控制（均匀/广播/自定义）
访问方式	通过 RuntimeContext 获取	通过 CheckpointedFunction 接口
典型场景	实时聚合、去重、状态机	Source 偏移量、广播配置

2.5 如何选择？

这个问题没有标准答案，但我可以给你一个决策思路：

1. 先看数据是否按 Key 分区。如果业务逻辑天然按用户、商品等维度分组，用 Keyed State。
2. 再看状态是否需要跨 Key 共享。如果所有 Key 共享同一份配置或规则，用 BroadcastState。
3. 最后看重分配策略。如果状态需要均匀分配到新并行度，用 ListState；如果需要全量复制，用 UnionListState。

2.6 实战：一个实时去重的例子

最后给个实战例子。假设我们要做订单去重，防止重复支付回调：

// 使用 Keyed State 做去重
DataStream<PaymentEvent> paymentStream = env
    .addSource(kafkaConsumer)
    .keyBy(PaymentEvent::getOrderId)
    .process(new KeyedProcessFunction<String, PaymentEvent, Void>() {
        
        private ValueState<Boolean> processed;

        @Override
        public void open(Configuration parameters) {
            processed = getRuntimeContext()
                .getState(new ValueStateDescriptor<>("processed", Types.BOOLEAN));
        }

        @Override
        public void processElement(PaymentEvent event, Context ctx, Collector<Void> out) {
            // 如果已经处理过，直接丢弃
            if (processed.value() != null && processed.value()) {
                return;
            }
            // 标记为已处理
            processed.update(true);
            // 处理业务逻辑
            handlePayment(event);
        }
    });

这个例子很简单，但很实用。用 ValueState 记录每个订单是否已处理，状态按 orderId 隔离，互不影响。

好了，Keyed State 和 Operator State 的区别就讲到这里。记住一句话：Keyed State 管数据，Operator State 管算子。搞清楚了这一点，后面学状态后端、状态一致性就轻松多了。