3. Flink 基础回顾：核心概念与时间语义

在构建实时 PSI 漂移监测系统之前，必须牢固掌握 Flink DataStream API 的三大支柱：核心抽象、时间语义与水位线、状态后端。本章将系统梳理这些基础，为后续实现分布式的 PSI 计算与漂移检测铺平道路。

3.1 Flink DataStream API 核心概念

DataStream API 是 Flink 处理无界数据流的主要编程接口。理解其核心抽象是编写任何实时作业的前提。

3.1.1 核心编程模型

一个典型的 DataStream 作业包含以下步骤：

1. 创建 StreamExecutionEnvironment
2. 添加 Source 获取数据流
3. 对数据流应用 Transformation（map、filter、keyBy、window 等）
4. 调用 Sink 输出结果
5. 调用 env.execute() 启动作业

// 示例：PSI 监测作业骨架
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<FeatureDistribution> inputStream = env
    .addSource(new FlinkKafkaConsumer<>("feature-topic", ...));

DataStream<PSIResult> psiStream = inputStream
    .keyBy(FeatureDistribution::getFeatureName)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .process(new PSICalculationProcessFunction());

psiStream.addSink(new InfluxDBSink<>("psi-metrics"));

env.execute("Real-time PSI Drift Monitor");

3.2 Flink 时间语义与水位线

时间语义是流处理正确性的基石。Flink 支持三种时间概念，其中 EventTime 是 PSI 漂移监测的首选语义。

3.2.1 水位线（Watermark）机制

水位线是 Flink 处理乱序数据的核心机制。它表示“在此时间戳之前的数据已经全部到达”的断言。

• 单调递增：水位线时间戳只能向前推进，不能回退。
• 触发窗口计算：当水位线超过窗口结束时间时，触发该窗口的聚合计算。
• 容忍乱序：通过设置最大乱序时间（maxOutOfOrderness），允许一定范围内的延迟数据。

// 设置 EventTime 并配置水位线
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

DataStream<FeatureDistribution> withWatermarks = inputStream
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<FeatureDistribution>forBoundedOutOfOrderness(Duration.ofMinutes(5))
            .withTimestampAssigner((event, timestamp) -> event.getEventTimestamp())
    );

3.3 Flink 状态后端简介

状态后端决定了 Flink 作业的状态存储位置、访问性能和容错机制。在 PSI 漂移监测中，状态后端直接影响基准分布存储、窗口聚合效率和故障恢复速度。

3.3.1 状态后端选择对 PSI 监测的影响

• 基准分布存储：PSI 监测需要长期保存每个特征的基准分布（如 10 个分桶的计数）。当特征数量达到数千甚至数万时，状态规模可达 GB 级别，必须使用 RocksDBStateBackend。
• 增量 Checkpoint：RocksDB 支持增量 Checkpoint，只上传变更的 SST 文件，大幅减少 Checkpoint 耗时，适合秒级窗口的实时监测。
• 本地状态访问：RocksDB 将状态存储在本地磁盘，避免全量状态驻留堆内存，降低 GC 压力，保障高吞吐下的稳定性。

// 推荐配置：RocksDB 状态后端 + 增量 Checkpoint
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

RocksDBStateBackend rocksBackend = new RocksDBStateBackend(
    "hdfs://namenode:8020/flink/checkpoints", true); // true 表示增量 Checkpoint
env.setStateBackend(rocksBackend);

// 同时启用 Checkpoint
env.enableCheckpointing(60000); // 每分钟 Checkpoint 一次
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);

3.4 本章小结

• DataStream API 提供了 KeyedStream、Window、State 等核心抽象，是构建 PSI 监测作业的基础。
• EventTime + 水位线 是处理乱序特征分布数据的正确选择，确保 PSI 计算窗口的准确性。
• RocksDBStateBackend 是 PSI 监测场景的首选状态后端，支持大规模特征基准分布的高效存储与容错。