3. 基于事件时间的交易流水水位线生成

任务目标

读取交易流水（含 txId、amount、eventTime），为每条记录分配事件时间戳，并生成周期性水位线（默认每 200ms 一次），水位线取当前已见最大事件时间减 5 秒，从而容忍最多 5 秒乱序。水位线是后续窗口触发的前提。

环境与依赖

• JDK 8+，Maven 项目
• Flink 1.17.x：flink-streaming-java、flink-clients

核心实现

关键点：用 BoundedOutOfOrdernessTimestampExtractor 的替代实现 WatermarkStrategy.forBoundedOutOfOrderness 绑定事件时间与乱序容忍；withTimestampAssigner 从记录中取毫秒时间戳；withPeriodicWatermarks 设置 200ms 周期。数据源用 fromElements 构造乱序样例，便于直接验证。

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

public class EventTimeWatermarkJob {

    // 交易流水记录
    public static class Tx {
        public String txId;
        public double amount;
        public long eventTime; // 毫秒
        public Tx(String txId, double amount, long eventTime) {
            this.txId = txId; this.amount = amount; this.eventTime = eventTime;
        }
        @Override public String toString() {
            return txId + "," + amount + "," + eventTime;
        }
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 乱序样例：eventTime 非单调递增
        DataStream<Tx> stream = env.fromElements(
                new Tx("t1", 100.0, 1000L),
                new Tx("t2", 200.0, 3000L),
                new Tx("t3", 150.0, 2000L), // 乱序：早于 t2
                new Tx("t4", 300.0, 5000L),
                new Tx("t5", 250.0, 4000L)  // 乱序：早于 t4
        );

        // 事件时间 + 5 秒乱序容忍 + 200ms 周期水位线
        WatermarkStrategy<Tx> strategy = WatermarkStrategy
                .<Tx>forBoundedOutOfOrderness(java.time.Duration.ofSeconds(5))
                .withTimestampAssigner((tx, ts) -> tx.eventTime)
                .withPeriodicWatermarks(); // 默认周期 200ms

        DataStream<Tx> withTs = stream.assignTimestampsAndWatermarks(strategy);

        // 打印每条记录及其当前水位线，验证分配与生成
        withTs.process(new ProcessFunction<Tx, String>() {
            @Override
            public void processElement(Tx tx, Context ctx, Collector<String> out) {
                long wm = ctx.timerService().currentWatermark();
                out.collect("tx=" + tx + " | watermark=" + wm);
            }
        }).print();

        env.execute("EventTimeWatermarkJob");
    }
}

运行方式

mvn -q compile exec:java -Dexec.mainClass=EventTimeWatermarkJob

或在 IDE 中直接运行 main。

验证方法与通过条件

• 每条记录都被分配了 eventTime 对应的时间戳（可在 processElement 中通过 ctx.timestamp() 校验）。
• 水位线随最大事件时间推进：处理到 t4（5000ms）后，水位线应达到 5000 - 5000 = 0；处理到 t5（4000ms）时最大时间仍为 5000，水位线保持 0，说明乱序记录未使水位线回退。
• 若把乱序容忍改为 0 秒，则 t3（2000ms）到达时水位线已推进到 3000，会被判定为迟到数据——这可用于对比验证容忍参数的作用。

SVG：事件时间与水位线推进

常见问题

• 水位线不推进：检查是否设置了并行度或数据源是否持续产生新记录；fromElements 数据量小，水位线可能停在初始值。
• 迟到数据被丢弃：默认容忍 5 秒，若业务乱序更大需调大 forBoundedOutOfOrderness 参数。
• 时间戳单位：Flink 事件时间以毫秒计，若源数据为秒需乘以 1000。