4. Flink时间语义与Watermark：EventTime与ProcessingTime实战
好，咱们今天聊一个在实时计算里绕不开的话题——时间语义和Watermark。
说实话，我刚接触Flink那会儿，也被这几个概念搞得有点晕。EventTime、ProcessingTime、IngestionTime，再加上Watermark，感觉像是一堆术语在打架。但后来我在一个量化交易项目里真正用上了它们，才明白这些东西到底有多重要。
你想想看，在量化交易里，数据是有严格的时间顺序的。比如一条成交数据，它发生的时间是10:00:01，但系统处理它的时间可能是10:00:03。如果你按处理时间算，那这条数据就“迟到”了，你的策略逻辑可能就全乱了。
所以，咱们得先搞清楚：你到底该用哪个时间？
4.1 时间语义：三种时间，三种视角
Flink里定义了三种时间语义，我一个个说。
| 时间语义 | 含义 | 典型场景 |
|---|---|---|
| EventTime | 事件实际发生的时间 | 交易数据、传感器数据 |
| ProcessingTime | 数据到达Flink算子的时间 | 实时监控、简单聚合 |
| IngestionTime | 数据进入Flink Source的时间 | 介于两者之间，较少用 |
EventTime 是最接近业务真相的。比如一条订单数据，它自带的timestamp字段就是EventTime。我个人习惯，只要业务允许，优先用EventTime。为什么？因为它能保证结果的可复现性。你重跑一遍历史数据，结果是一样的。
ProcessingTime 就简单多了。数据啥时候到，我就用啥时候的时间。好处是延迟低，坏处是结果不稳定。你想想看，如果网络抖动一下，数据晚到了几秒，那聚合结果可能就变了。这在量化交易里是致命的。
IngestionTime 其实是个折中方案。它比ProcessingTime稳定，但又不需要从数据里提取时间戳。我一般不太用，除非数据本身没有时间戳。
核心建议： 量化交易场景，99%的情况用EventTime。别偷懒。
4.2 Watermark：解决“迟到”问题的关键
好，问题来了。你用EventTime，但数据可能迟到。比如一条10:00:01的数据，因为网络延迟，10:00:05才到。这时候你的窗口已经计算完了，这条数据就被丢了。
怎么办？Watermark就是干这个的。
Watermark说白了就是一个“时间进度标记”。它告诉Flink：“到目前为止，我收到的数据里，时间戳小于等于这个值的数据，基本都到齐了。你可以放心地触发窗口计算了。”
举个例子。你设置Watermark延迟为3秒。那么当Flink收到一条时间戳为10:00:10的数据时，它会认为所有时间戳小于等于10:00:07的数据都已经到了。这时候，窗口[10:00:00, 10:00:05)就可以被触发了。
嗯，这里要注意：Watermark不是绝对的。它只是一个“大概率”的估计。你设置3秒延迟，不代表不会有迟到超过3秒的数据。那怎么办？后面会讲。
4.3 Watermark的生成策略
Flink提供了两种生成Watermark的方式：
- 周期性生成（Periodic）：每隔一段时间生成一次Watermark。适合数据量大的场景。
- 逐条生成（Punctuated）：每来一条数据就判断是否生成Watermark。适合数据量小、对延迟敏感的场景。
我直接给你看代码，这样更直观。
// 周期性生成Watermark
DataStream<Trade> stream = env.addSource(tradeSource)
.assignTimestampsAndWatermarks(
WatermarkStrategy.<Trade>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner((event, timestamp) -> event.getTimestamp())
);
// 自定义Watermark生成器
DataStream<Trade> stream = env.addSource(tradeSource)
.assignTimestampsAndWatermarks(
WatermarkStrategy.<Trade>forGenerator(ctx -> new MyWatermarkGenerator())
.withTimestampAssigner((event, timestamp) -> event.getTimestamp())
);
上面这个例子，forBoundedOutOfOrderness(Duration.ofSeconds(3)) 是最常用的。它假设数据最多迟到3秒。我在项目中一般先用这个，跑一段时间看看实际延迟情况，再调整。
我的经验： 别把Watermark延迟设得太小。我曾经设了1秒，结果线上经常丢数据。后来改成3秒，稳多了。当然，也别太大，否则窗口迟迟不触发，实时性就没了。
4.4 如何处理迟到数据？
Watermark不是万能的。总会有数据迟到得离谱。比如网络断了5分钟，恢复后一堆旧数据涌进来。
这时候，你有三个选择：
- 直接丢弃：默认行为。简单粗暴，但可能丢数据。
- 侧输出（Side Output）：把迟到数据单独收集起来，后续处理。
- 重新触发窗口：用
allowedLateness让窗口多等一会儿。
我个人最推荐第二种——侧输出。为什么呢？因为你可以把迟到数据存到另一个Topic或数据库里，然后人工分析。这样既不耽误主流程，又能保留数据。
看代码：
// 定义侧输出标签
OutputTag<Trade> lateTag = new OutputTag<Trade>("late-data") {};
// 在窗口操作中捕获迟到数据
DataStream<Trade> mainStream = stream
.keyBy(Trade::getSymbol)
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.sideOutputLateData(lateTag)
.process(new MyWindowFunction());
// 获取迟到数据流
DataStream<Trade> lateStream = mainStream.getSideOutput(lateTag);
你看，这样主流程照常运行，迟到数据被单独拎出来了。我有个项目就是这么干的，后来发现迟到数据里居然有套利机会——因为那些数据来自一个延迟较高的数据源，反而成了信号。
注意： 侧输出会占用额外内存。如果迟到数据量特别大，建议加个限流或直接丢弃。别让迟到数据拖垮整个作业。
4.5 实战：一个简单的量化交易窗口计算
咱们来写一个完整的例子。假设我们要计算每5秒内，每只股票的平均成交价。
public class TradeAvgPriceJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设置EventTime
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream<Trade> tradeStream = env.addSource(new TradeSource())
.assignTimestampsAndWatermarks(
WatermarkStrategy.<Trade>forBoundedOutOfOrderness(Duration.ofSeconds(3))
.withTimestampAssigner((trade, ts) -> trade.getTimestamp())
);
DataStream<AvgPriceResult> result = tradeStream
.keyBy(Trade::getSymbol)
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.allowedLateness(Time.seconds(2)) // 允许额外2秒的迟到数据
.sideOutputLateData(lateTag)
.aggregate(new AvgPriceAggregator(), new AvgPriceProcessWindowFunction());
result.print();
env.execute("Trade Avg Price Job");
}
}
这个例子里，我用了allowedLateness(Time.seconds(2))。意思是，窗口触发后，再等2秒。这2秒内来的迟到数据，还能更新窗口结果。2秒后，窗口彻底关闭，迟到数据就走侧输出了。
你可能会问：为什么不用更大的allowedLateness？嗯，因为窗口状态会一直保留，占用内存。而且，如果数据迟到太久，窗口结果早就被下游消费了，你再更新也没意义。
4.6 知识体系图
下面这张图，帮你理清本章的核心逻辑：
这张图把整个流程串起来了。从选择时间语义，到生成Watermark，再到处理迟到数据，每一步都有对应的策略。你在实际项目中，可以照着这个图来设计你的实时计算逻辑。
一句话总结： 用EventTime做业务逻辑，用Watermark控制窗口触发时机，用侧输出兜底迟到数据。这三板斧用好了，量化交易的实时计算基本就稳了。