4. Flink状态管理与容错：Keyed State与Operator State、Checkpoint与Savepoint、Exactly-Once语义保障

在反洗钱PSI漂移监测场景中，模型特征分布（PSI）的计算需要跨多个事件窗口累积统计量。Flink的状态管理机制是保障这些累积数据不丢失、不重复、可恢复的核心。本章深入剖析Flink的两种状态类型、两大容错机制以及端到端的一致性语义。

4.1 Keyed State 与 Operator State

Flink将状态分为两大类：Keyed State（键控状态）和Operator State（算子状态）。理解它们的区别是设计容错应用的基础。

4.1.1 Keyed State 实战：PSI累积统计

在反洗钱PSI监测中，我们需要按模型版本ID + 特征名称作为Key，累积每个时间窗口内的期望频次和观测频次。以下代码展示了如何使用ValueState存储累积计数：

public class PsiAccumulator extends RichFlatMapFunction<Transaction, PsiEvent> {
    // 键控状态：存储当前Key的累积计数
    private transient ValueState<PsiAccumulatorState> accumulatorState;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<PsiAccumulatorState> descriptor =
            new ValueStateDescriptor<>(
                "psi-accumulator",          // 状态名称
                TypeInformation.of(PsiAccumulatorState.class),
                new PsiAccumulatorState()   // 默认值
            );
        accumulatorState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void flatMap(Transaction transaction, Collector<PsiEvent> out) throws Exception {
        PsiAccumulatorState current = accumulatorState.value();
        // 更新累积统计量
        current.addObservation(transaction.getFeatureValue());
        accumulatorState.update(current);

        // 判断是否触发PSI计算窗口
        if (current.getCount() >= WINDOW_SIZE) {
            double psi = computePsi(current);
            out.collect(new PsiEvent(transaction.getModelId(), psi));
            accumulatorState.clear(); // 重置状态
        }
    }
}

4.1.2 Operator State 实战：Kafka偏移量管理

当使用FlinkKafkaConsumer时，偏移量管理默认使用Operator State。每个并行子任务独立记录其消费的分区偏移量，并在故障恢复时从最近一次Checkpoint的偏移量重新消费。

// 自定义Operator State示例：记录已处理的事件ID，用于去重
public class DeduplicationSink implements SinkFunction<Transaction> {
    private transient ListState<String> processedIds;

    @Override
    public void invoke(Transaction value, Context context) throws Exception {
        // 检查是否已处理
        if (processedIds.get().contains(value.getEventId())) {
            return; // 跳过重复事件
        }
        // 处理事件...
        processedIds.add(value.getEventId());
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        ListStateDescriptor<String> descriptor =
            new ListStateDescriptor<>("processed-ids", String.class);
        processedIds = context.getOperatorStateStore().getListState(descriptor);
    }
}

4.2 Checkpoint 与 Savepoint

Flink提供两种快照机制来保障状态的可恢复性：Checkpoint（自动检查点）和Savepoint（手动保存点）。

4.3 Exactly-Once 语义保障

Flink通过Checkpoint + 两阶段提交协议（Two-Phase Commit）实现端到端的Exactly-Once语义。这要求Source和Sink都支持事务或幂等写入。

4.3.1 语义层级对比

4.3.2 两阶段提交在PSI场景的实现

当PSI计算结果需要写入Kafka或数据库时，Flink的TwoPhaseCommitSinkFunction确保每个Checkpoint周期内的数据要么全部提交，要么全部回滚。

4.3.3 配置示例：启用Exactly-Once

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 启用Checkpoint，间隔30秒
env.enableCheckpointing(30000);

// 设置语义为Exactly-Once
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 设置超时时间
env.getCheckpointConfig().setCheckpointTimeout(60000);

// 设置并发Checkpoint数量
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// 启用外部化持久化（保留Checkpoint到外部存储）
env.getCheckpointConfig().enableExternalizedCheckpoints(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

// Kafka Source配置Exactly-Once语义
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
kafkaProps.setProperty("group.id", "psi-monitor-group");
kafkaProps.setProperty("enable.auto.commit", "false"); // 必须关闭自动提交

FlinkKafkaConsumer<Transaction> consumer = new FlinkKafkaConsumer<>(
    "transactions",
    new TransactionDeserializationSchema(),
    kafkaProps
);
consumer.setStartFromLatest();
consumer.setCommitOffsetsOnCheckpoints(true); // 依赖Checkpoint提交偏移量

4.4 反洗钱PSI场景的状态容错配置清单