时间语义
在Flink中，时间语义分为两种 ： 处理时间和事件时间。时间语义与窗口函数是密不可分的。以窗口为单位进行某一段时间内指标统计，例如想要统计8点-9点的某个页面的访问量，此时就需要用到了窗口函数，这里的关键时间点是8点到9点，而这个时间点指的是事件时间，不是处理时间，也就是某个时间发生的时间。
一条日志数据发生的时间是8:59，而经过一系列转换到达系统开始处理的时间可能是9:01，很显然，如果想要统计8点到9点的数据，应该是根据事件的发生时间，而不是处理时间。
- WaterMark
上面说到，在窗口函数开窗以及关窗中，不应该以当前系统的处理时间，而应该取当前的事件时间，这个事件时间就是WaterMark 用于标识当前流数据的时间到什么时候了，这个时间不是由系统去推进的，而是基于事件的到来去推进的。
- WaterMark的延时
在流处理中，数据并不是一批一批到达的而是流式到达，且到达分布式系统处理的时间可能是乱序的
8:57 、 9:01 、 8:59 、 9:02 …
如果 想要统计8点到9点的数据统计，按照事件时间，当 9:01的数据到达之后，便认为，8点到9点的窗口可以关闭进行运算了，但是8:59 这条数据就 不在统计范围内了，这样就存在一定意义上的数据丢失。要解决这个问题，就是让Watermark的时间进行延时，也就是说，本来是计划9点关窗计算的，让Watermark的时间减少2分钟，那9:01这条数据到来之后，WaterMark的时间其实是8:59,9:02的数据来到以后，WaterMark才会到9:00，此时8点-9点才会关窗计算。这个等多久的时间就是WaterMark的延时。
窗口
窗口函数的统计基于时间语义的watermark,当窗口接收到数据以后，首先需要根据数据身上的时间戳来判断此条数据属于哪个窗口，将属于某个窗口的事件分配给窗口，而关窗计算的时间取决于WaterMark。在Flink中，窗口在同一时刻是可以存在多个，而不是只有一个。
数据流的WaterMark定义
- 在DataStream上游数据流通过assignTimestampsAndWatermarks 分配watermark
new WatermarkStrategy<Event>() {@Overridepublic WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {return null;}@Overridepublic TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {return WatermarkStrategy.super.createTimestampAssigner(context);}}
createWatermarkGenerator 方法用于返回 WatermarkGenerator
@Public
public interface WatermarkGenerator<T> {void onEvent(T var1, long var2, WatermarkOutput var4);void onPeriodicEmit(WatermarkOutput var1);
}
WatermarkGenerator 两个方法分别对基于事件的WaterMark以及周期性的生成watermark两种生成策略发射watermark。
而每条数据上的时间戳该如何提前，是通过 TimestampAssigner 这个时间戳分配器提取的。
@FunctionalInterface
@Public
public interface TimestampAssigner<T> {long NO_TIMESTAMP = -9223372036854775808L;long extractTimestamp(T var1, long var2);
}extractTimestamp 返回了一个时间戳，获取到这个时间以后，便可以根据这个时间去推进WaterMark的前进。
- WaterMark时间分配器定义
//时间戳分配器,获取到事件时间后用于推进WaterMark的前进
class MyTimeStampAss implements TimestampAssigner<Event>{@Overridepublic long extractTimestamp(Event event, long l) {return event.timestamp;}
}
- WaterMark生成器定义
//水位线提取器，基于事件或者周期性的生成水位线
class MyWaterMarkGenerator implements WatermarkGenerator<Event>{private long maxTimeStamp = 0;@Overridepublic void onEvent(Event event, long l, WatermarkOutput watermarkOutput) {maxTimeStamp = Math.max(event.timestamp,maxTimeStamp);}@Overridepublic void onPeriodicEmit(WatermarkOutput watermarkOutput) {watermarkOutput.emitWatermark(new Watermark(maxTimeStamp)); //发射Watermark，让下游窗口感知}
}
- 给DataStream设置WaterMark
public class WaterMarkTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = Env.getEnv();DataStream<Event> dataStream = env.addSource(new ClickSource());//设置数据流的watermarkSingleOutputStreamOperator<Event> stream = dataStream.assignTimestampsAndWatermarks(new MyWaterMarkStrategy());env.execute();}}class MyWaterMarkStrategy implements WatermarkStrategy<Event>{@Overridepublic WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {return new MyWaterMarkGenerator();}@Overridepublic TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {return new MyTimeStampAss();}
}
默认的周期性生成WaterMark为200ms
env.getConfig().setAutoWatermarkInterval(100);
