4. Flink 实时计算基础：DataStream API、时间语义、Watermark 机制

在支付风控批次偏移预警场景中，我们需要在毫秒级延迟内处理海量交易事件，并准确判断“批次是否发生偏移”。Flink 的 DataStream API 提供了构建此类实时管道的核心抽象，而 时间语义 与 Watermark 机制 则是解决乱序事件与延迟数据的关键技术。本节将围绕这三个基础概念展开，为后续实现 PSI 实时计算打下坚实的技术地基。

4.1 DataStream API：实时数据流的编程模型

DataStream API 是 Flink 中处理无界数据流（如交易事件、日志流）的核心接口。它将数据流抽象为 DataStream<T>，支持丰富的转换算子（Transformation），形成有向无环图（DAG）执行计划。

// 示例：从 Kafka 读取交易事件，按批次 ID 分组，并应用 10 秒滚动窗口
DataStream<TransactionEvent> transactionStream = env
    .fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source");

transactionStream
    .keyBy(TransactionEvent::getBatchId)
    .window(TumblingEventTimeWindows.of(Time.seconds(10)))
    .process(new BatchShiftWarningProcessFunction())
    .sinkTo(warningSink);

4.2 时间语义：事件时间 vs 处理时间

Flink 支持三种时间语义，在支付风控中，事件时间（Event Time） 是最符合业务逻辑的选择，因为它基于交易实际发生的时间戳，而非数据到达 Flink 算子的系统时间。

// 正确设置事件时间提取器
WatermarkStrategy<TransactionEvent> watermarkStrategy = WatermarkStrategy
    .<TransactionEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
    .withTimestampAssigner((event, timestamp) -> event.getEventTimestamp());

DataStream<TransactionEvent> streamWithTimestamps = env
    .fromSource(kafkaSource, watermarkStrategy, "Kafka Source");

4.3 Watermark 机制：处理乱序与延迟的核心

Watermark 是 Flink 中衡量事件时间进展的机制，它本质上是一个带时间戳的流式信号，表示“在此时间戳之前的事件已经全部到达”。Watermark 推动窗口的触发与计算。

4.4 Watermark 生成策略与配置

Flink 提供了两种内置的 Watermark 生成策略，开发者也可以自定义。在支付风控场景中，推荐使用 周期性 Watermark 配合 最大乱序容忍度。

// 推荐配置：允许 5 秒乱序，每 100ms 生成一次 Watermark
WatermarkStrategy<TransactionEvent> strategy = WatermarkStrategy
    .<TransactionEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
    .withTimestampAssigner((event, timestamp) -> event.getEventTimestamp())
    .withIdleness(Duration.ofSeconds(1)); // 处理空闲分区

DataStream<TransactionEvent> stream = env
    .fromSource(kafkaSource, strategy, "Payment Source")
    .setParallelism(4);

4.5 时间语义与 Watermark 在批次偏移预警中的协同

在批次偏移预警中，事件时间与 Watermark 的协同工作流程如下：

1. 事件提取： 从 Kafka 消费交易事件，提取业务时间戳（如支付完成时间）。
2. Watermark 推进： 基于最大乱序容忍度（如 5 秒）周期性生成 Watermark。
3. 窗口划分： 按事件时间将数据分配到滚动窗口（如 10 秒窗口）。
4. 窗口触发： 当 Watermark 超过窗口结束时间时，触发窗口计算。
5. PSI 计算： 在窗口内计算当前批次与参考批次的分布偏移。
6. 延迟处理： 使用 allowedLateness() 处理窗口关闭后到达的延迟事件。

// 完整示例：事件时间 + Watermark + 窗口 + 延迟处理
DataStream<BatchShiftWarning> warnings = transactionStream
    .keyBy(TransactionEvent::getBatchId)
    .window(TumblingEventTimeWindows.of(Time.seconds(10)))
    .allowedLateness(Time.seconds(5))  // 允许 5 秒延迟
    .sideOutputLateData(lateOutputTag) // 捕获超延迟事件
    .process(new PSICalculationProcessFunction());

// 处理超延迟事件（可单独告警或落盘）
DataStream<TransactionEvent> lateStream = warnings.getSideOutput(lateOutputTag);