一、前言：为什么需要 State？

1.1 无状态 vs 有状态

1.2 哪些场景需要 State？

1.3 Flink State 的核心价值

二、State 类型详解

2.1 State 分类

2.2 Keyed State 详解

Keyed State 类型：

代码示例：

public class StatefulFunction extends KeyedProcessFunction<String, Event, Result> {
    
    // 1. ValueState - 单个值
    private ValueState<Long> countState;
    
    // 2. ListState - 列表
    private ListState<Event> historyState;
    
    // 3. MapState - Map
    private MapState<String, Long> categoryCountState;
    
    // 4. ReducingState - 归约
    private ReducingState<Long> sumState;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化 ValueState
        ValueStateDescriptor<Long> countDesc = new ValueStateDescriptor<>("count", Long.class);
        countState = getRuntimeContext().getState(countDesc);
        
        // 初始化 ListState
        ListStateDescriptor<Event> historyDesc = new ListStateDescriptor<>("history", Event.class);
        historyState = getRuntimeContext().getListState(historyDesc);
        
        // 初始化 MapState
        MapStateDescriptor<String, Long> categoryDesc = new MapStateDescriptor<>("category", String.class, Long.class);
        categoryCountState = getRuntimeContext().getMapState(categoryDesc);
        
        // 初始化 ReducingState
        ReducingStateDescriptor<Long> sumDesc = new ReducingStateDescriptor<>("sum", Long::sum, Long.class);
        sumState = getRuntimeContext().getReducingState(sumDesc);
    }
    
    @Override
    public void processElement(Event event, Context ctx, Collector<r> out) throws Exception {
        // 使用 ValueState
        Long count = countState.value();
        if (count == null) count = 0L;
        countState.update(count + 1);
        
        // 使用 ListState
        historyState.add(event);
        
        // 使用 MapState
        String category = event.getCategory();
        Long categoryCount = categoryCountState.get(category);
        categoryCountState.put(category, categoryCount == null ? 1 : categoryCount + 1);
        
        // 使用 ReducingState
        sumState.add(event.getAmount());
        
        // 输出结果
        out.collect(new Result(countState.value(), sumState.get()));
    }
}

2.3 Operator State 详解

代码示例：

public class MySourceFunction implements SourceFunction<String>, CheckpointedFunction {
    
    private volatile boolean isRunning = true;
    private long offset;
    
    // Operator State
    private ListState<Long> offsetState;
    
    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        // 初始化 Operator State
        ListStateDescriptor<Long> desc = new ListStateDescriptor<>("offset", Long.class);
        offsetState = context.getOperatorStateStore().getListState(desc);
        
        // 从 State 恢复
        if (context.isRestored()) {
            for (Long o : offsetState.get()) {
                offset = o;
            }
        }
    }
    
    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        // Checkpoint 时保存状态
        offsetState.clear();
        offsetState.add(offset);
    }
    
    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (isRunning) {
            synchronized (ctx.getCheckpointLock()) {
                // 读取数据并更新 offset
                String data = readFromSource(offset);
                ctx.collect(data);
                offset++;
            }
        }
    }
    
    @Override
    public void cancel() {
        isRunning = false;
    }
}

2.4 Broadcast State

代码示例：

// 规则流
DataStream<Rule> ruleStream = env.addSource(new RuleSource());

// 数据流
DataStream<Event> dataStream = env.addSource(new EventSource());

// 定义 Broadcast State
MapStateDescriptor<String, Rule> ruleStateDesc = new MapStateDescriptor<>(
    "rules",
    String.class,
    Rule.class
);

// 广播规则流
BroadcastStream<Rule> broadcastRules = ruleStream.broadcast(ruleStateDesc);

// 连接数据流和广播流
dataStream
    .keyBy(Event::getKey)
    .connect(broadcastRules)
    .process(new BroadcastProcessFunction<Event, Rule, Result>() {
        
        @Override
        public void processElement(Event event, ReadOnlyContext ctx, Collector<r> out) {
            // 读取规则（只读）
            ReadOnlyBroadcastState<String, Rule> rules = ctx.getBroadcastState(ruleStateDesc);
            Rule rule = rules.get(event.getRuleId());
            
            if (rule != null && rule.match(event)) {
                out.collect(new Result(event, rule));
            }
        }
        
        @Override
        public void processBroadcastElement(Rule rule, Context ctx, Collector<r> out) {
            // 更新规则
            BroadcastState<String, Rule> rules = ctx.getBroadcastState(ruleStateDesc);
            rules.put(rule.getId(), rule);
        }
    });

三、State Backend

3.1 State Backend 类型

3.2 对比与选型

3.3 配置方式

// 方式1：代码配置
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// HashMapStateBackend
env.setStateBackend(new HashMapStateBackend());

// RocksDBStateBackend
env.setStateBackend(new EmbeddedRocksDBStateBackend());

// 方式2：flink-conf.yaml
// state.backend: rocksdb
// state.checkpoints.dir: hdfs:///flink/checkpoints
// state.backend.rocksdb.localdir: /data/rocksdb

3.4 RocksDB 调优

# flink-conf.yaml RocksDB 配置

# 本地目录（建议 SSD）
state.backend.rocksdb.localdir: /ssd/rocksdb

# 内存管理（托管内存）
state.backend.rocksdb.memory.managed: true

# Block Cache 大小
state.backend.rocksdb.block.cache-size: 256mb

# Write Buffer 数量
state.backend.rocksdb.writebuffer.count: 4

# Write Buffer 大小
state.backend.rocksdb.writebuffer.size: 128mb

# 增量 Checkpoint
state.backend.incremental: true

四、Checkpoint 机制

4.1 Checkpoint 原理

4.2 Barrier 对齐

4.3 Unaligned Checkpoint (Flink 1.11+)

配置：

// 开启 Unaligned Checkpoint
env.getCheckpointConfig().enableUnalignedCheckpoints();

// 或者配置文件
// execution.checkpointing.unaligned: true

4.4 Checkpoint 配置

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 启用 Checkpoint，间隔 1 分钟
env.enableCheckpointing(60000);

CheckpointConfig config = env.getCheckpointConfig();

// Checkpoint 模式
config.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 最小间隔
config.setMinPauseBetweenCheckpoints(30000);

// 超时时间
config.setCheckpointTimeout(600000);

// 最大并发 Checkpoint
config.setMaxConcurrentCheckpoints(1);

// 失败容忍次数
config.setTolerableCheckpointFailureNumber(3);

// 取消时保留 Checkpoint
config.setExternalizedCheckpointCleanup(
    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

// Checkpoint 存储
config.setCheckpointStorage("hdfs:///flink/checkpoints");

// Unaligned Checkpoint
config.enableUnalignedCheckpoints();

五、Savepoint

5.1 Checkpoint vs Savepoint

5.2 Savepoint 操作

# 触发 Savepoint
bin/flink savepoint <jobId> [savepointPath]
bin/flink savepoint abc123 hdfs:///savepoints/

# 取消作业并触发 Savepoint
bin/flink cancel -s [savepointPath] <jobId>
bin/flink cancel -s hdfs:///savepoints/ abc123

# 从 Savepoint 恢复
bin/flink run -s hdfs:///savepoints/savepoint-abc123 myJob.jar

# 允许跳过无法恢复的状态
bin/flink run -s hdfs:///savepoints/savepoint-abc123 --allowNonRestoredState myJob.jar

5.3 状态兼容性

最佳实践：

// 为算子指定 UID，保证升级兼容性
stream
    .map(new MyMapper())
    .uid("my-mapper")  // 重要！
    .name("My Mapper")
    .keyBy(...)
    .process(new MyProcessor())
    .uid("my-processor")  // 重要！
    .name("My Processor");

六、Exactly-Once 语义

6.1 端到端 Exactly-Once

三个条件：

1. Source 可重放：如 Kafka（offset 可回溯）
2. Flink 内部：Checkpoint + Barrier
3. Sink 端：幂等或两阶段提交

6.2 两阶段提交 (2PC)

6.3 Kafka Exactly-Once 示例

// Kafka Producer 配置
Properties producerProps = new Properties();
producerProps.setProperty("bootstrap.servers", "kafka:9092");
producerProps.setProperty("transaction.timeout.ms", "600000");

// 创建 Kafka Sink（Exactly-Once）
KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setRecordSerializer(
        KafkaRecordSerializationSchema.builder()
            .setTopic("output-topic")
            .setValueSerializationSchema(new SimpleStringSchema())
            .build()
    )
    .setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)  // Exactly-Once
    .setTransactionalIdPrefix("my-app")  // 事务ID前缀
    .build();

stream.sinkTo(sink);

七、State 调优

7.1 State TTL

// 设置 State TTL
StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(7))  // 7天过期
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)  // 创建和写入时更新
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)  // 不返回过期数据
    .cleanupFullSnapshot()  // 全量快照时清理
    .cleanupInRocksdbCompactFilter(1000)  // RocksDB Compaction 时清理
    .build();

// 应用到 State
ValueStateDescriptor<Long> stateDesc = new ValueStateDescriptor<>("my-state", Long.class);
stateDesc.enableTimeToLive(ttlConfig);

7.2 大状态优化

7.3 常见问题排查

八、实战案例

8.1 案例：实时去重计数

/**
 * 实时统计独立用户数（精确去重）
 */
public class ExactUVCount extends KeyedProcessFunction<String, UserEvent, Tuple2<String, Long>> {
    
    // 使用 MapState 存储用户 ID
    private MapState<Long, Boolean> userIdState;
    private ValueState<Long> countState;
    
    @Override
    public void open(Configuration parameters) {
        // 配置 TTL
        StateTtlConfig ttlConfig = StateTtlConfig
            .newBuilder(Time.days(1))
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .cleanupInRocksdbCompactFilter(1000)
            .build();
        
        MapStateDescriptor<Long, Boolean> userIdDesc = new MapStateDescriptor<>("user-ids", Long.class, Boolean.class);
        userIdDesc.enableTimeToLive(ttlConfig);
        userIdState = getRuntimeContext().getMapState(userIdDesc);
        
        ValueStateDescriptor<Long> countDesc = new ValueStateDescriptor<>("count", Long.class);
        countDesc.enableTimeToLive(ttlConfig);
        countState = getRuntimeContext().getState(countDesc);
    }
    
    @Override
    public void processElement(UserEvent event, Context ctx, Collector<Tuple2<String, Long>> out) throws Exception {
        Long userId = event.getUserId();
        
        // 检查是否已存在
        if (!userIdState.contains(userId)) {
            userIdState.put(userId, true);
            
            Long count = countState.value();
            count = count == null ? 1 : count + 1;
            countState.update(count);
        }
        
        // 输出当前 UV
        out.collect(Tuple2.of(ctx.getCurrentKey(), countState.value()));
    }
}

8.2 案例：双流 Join

/**
 * 订单流和支付流 Join
 */
public class OrderPaymentJoin extends CoProcessFunction<Order, Payment, OrderWithPayment> {
    
    // 订单状态（等待支付）
    private MapState<String, Order> orderState;
    
    // 支付状态（订单未到）
    private MapState<String, Payment> paymentState;
    
    @Override
    public void open(Configuration parameters) {
        // 5分钟 TTL
        StateTtlConfig ttlConfig = StateTtlConfig
            .newBuilder(Time.minutes(5))
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .build();
        
        MapStateDescriptor<String, Order> orderDesc = new MapStateDescriptor<>("orders", String.class, Order.class);
        orderDesc.enableTimeToLive(ttlConfig);
        orderState = getRuntimeContext().getMapState(orderDesc);
        
        MapStateDescriptor<String, Payment> paymentDesc = new MapStateDescriptor<>("payments", String.class, Payment.class);
        paymentDesc.enableTimeToLive(ttlConfig);
        paymentState = getRuntimeContext().getMapState(paymentDesc);
    }
    
    @Override
    public void processElement1(Order order, Context ctx, Collector<OrderWithPayment> out) throws Exception {
        // 检查是否有匹配的支付
        Payment payment = paymentState.get(order.getOrderId());
        if (payment != null) {
            out.collect(new OrderWithPayment(order, payment));
            paymentState.remove(order.getOrderId());
        } else {
            // 缓存订单，等待支付
            orderState.put(order.getOrderId(), order);
        }
    }
    
    @Override
    public void processElement2(Payment payment, Context ctx, Collector<OrderWithPayment> out) throws Exception {
        // 检查是否有匹配的订单
        Order order = orderState.get(payment.getOrderId());
        if (order != null) {
            out.collect(new OrderWithPayment(order, payment));
            orderState.remove(payment.getOrderId());
        } else {
            // 缓存支付，等待订单
            paymentState.put(payment.getOrderId(), payment);
        }
    }
}

九、监控与运维

9.1 关键监控指标

9.2 Checkpoint 失败排查

十、总结

10.1 核心要点

10.2 一句话总结

State = Flink 有状态计算的基础

Checkpoint = Flink 容错的核心机制

两者结合，实现 Exactly-Once 语义