4. Flink 状态编程：Keyed State 与 Operator State

在反欺诈实时特征服务中，状态（State）是维持跨事件计算逻辑的核心。Flink 提供两种主要状态类型：Keyed State（键控状态）和 Operator State（算子状态）。理解它们的区别与适用场景，是构建高可靠反欺诈特征 pipeline 的基础。

4.1 Keyed State 与 Operator State 对比

4.2 三种核心状态类型详解

4.2.1 ValueState：单值状态

适用场景：存储单个可更新的特征值，如用户最近一次交易时间、当前累计风险分、设备首次出现时间戳。

// 反欺诈场景：记录用户最近一次交易时间戳
public class LastTransactionTimeFunction extends KeyedProcessFunction<String, TransactionEvent, FeatureOutput> {
    private ValueState<Long> lastTxTimeState;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<Long> descriptor = new ValueStateDescriptor<>(
            "lastTxTime",           // 状态名称
            TypeInformation.of(new TypeHint<Long>() {})  // 类型信息
        );
        // 设置 TTL：保留 24 小时
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(24))
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
            .build();
        descriptor.enableTimeToLive(ttlConfig);
        lastTxTimeState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(TransactionEvent event, Context ctx, Collector<FeatureOutput> out) throws Exception {
        Long lastTime = lastTxTimeState.value();
        long currentTime = event.getEventTime();

        // 计算时间间隔特征（秒）
        long interval = (lastTime == null) ? -1L : (currentTime - lastTime) / 1000;

        // 更新状态
        lastTxTimeState.update(currentTime);

        // 输出特征
        out.collect(new FeatureOutput(event.getUserId(), "last_tx_interval_sec", interval));
    }
}

4.2.2 MapState：键值映射状态

适用场景：存储 key-value 结构的聚合数据，如用户近 1 小时各渠道交易次数、设备关联的 IP 集合、商户历史交易金额分布。

// 反欺诈场景：统计用户近 1 小时各渠道交易次数
public class ChannelCountFunction extends KeyedProcessFunction<String, TransactionEvent, FeatureOutput> {
    private MapState<String, Long> channelCountState;

    @Override
    public void open(Configuration parameters) {
        MapStateDescriptor<String, Long> descriptor = new MapStateDescriptor<>(
            "channelCount",
            TypeInformation.of(new TypeHint<String>() {}),
            TypeInformation.of(new TypeHint<Long>() {})
        );
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(1)).build();
        descriptor.enableTimeToLive(ttlConfig);
        channelCountState = getRuntimeContext().getMapState(descriptor);
    }

    @Override
    public void processElement(TransactionEvent event, Context ctx, Collector<FeatureOutput> out) throws Exception {
        String channel = event.getChannel();
        Long count = channelCountState.get(channel);
        count = (count == null) ? 1L : count + 1;
        channelCountState.put(channel, count);

        // 输出各渠道计数特征
        for (Map.Entry<String, Long> entry : channelCountState.entries()) {
            out.collect(new FeatureOutput(event.getUserId(),
                "channel_count_" + entry.getKey(), entry.getValue()));
        }
    }
}

4.2.3 ListState：列表状态

适用场景：存储有序的事件序列，如用户最近 10 笔交易金额列表、设备最近 5 次登录 IP、滑动窗口内的原始事件快照。

// 反欺诈场景：维护用户最近 5 笔交易金额，用于计算均值/方差
public class RecentAmountListFunction extends KeyedProcessFunction<String, TransactionEvent, FeatureOutput> {
    private ListState<Double> recentAmountState;
    private static final int MAX_SIZE = 5;

    @Override
    public void open(Configuration parameters) {
        ListStateDescriptor<Double> descriptor = new ListStateDescriptor<>(
            "recentAmount",
            TypeInformation.of(new TypeHint<Double>() {})
        );
        recentAmountState = getRuntimeContext().getListState(descriptor);
    }

    @Override
    public void processElement(TransactionEvent event, Context ctx, Collector<FeatureOutput> out) throws Exception {
        recentAmountState.add(event.getAmount());

        // 维护固定大小列表
        Iterable<Double> allAmounts = recentAmountState.get();
        List<Double> amountList = new ArrayList<>();
        allAmounts.forEach(amountList::add);

        if (amountList.size() > MAX_SIZE) {
            // 移除最早的元素
            amountList.remove(0);
            recentAmountState.update(amountList);
        }

        // 计算均值特征
        double avg = amountList.stream().mapToDouble(Double::doubleValue).average().orElse(0.0);
        out.collect(new FeatureOutput(event.getUserId(), "recent_5_avg_amount", avg));
    }
}

4.3 状态后端与性能考量

4.4 状态一致性保证

Flink 通过 Checkpoint 机制保证状态的一致性。在反欺诈场景中，建议配置：

• Exactly-Once 语义：确保特征不重复、不丢失
• Checkpoint 间隔：30s~60s，避免频繁 checkpoint 影响吞吐
• 最小间隔：10s，防止反压时 checkpoint 风暴
• 超时时间：10min，大状态场景适当调大

// 生产环境推荐配置
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(30000);  // 30s 间隔
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000);  // 最小间隔 10s
env.getCheckpointConfig().setCheckpointTimeout(600000);  // 超时 10min
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);