时间语义

在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>() {
            @Override
            public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return null;
            }

            @Override
            public 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>{

    @Override
    public long extractTimestamp(Event event, long l) {
        return event.timestamp;
    }
}

• WaterMark生成器定义

//水位线提取器，基于事件或者周期性的生成水位线
class MyWaterMarkGenerator implements WatermarkGenerator<Event>{

    private long maxTimeStamp = 0;

    @Override
    public void onEvent(Event event, long l, WatermarkOutput watermarkOutput) {
        maxTimeStamp = Math.max(event.timestamp,maxTimeStamp);
    }

    @Override
    public 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());

        //设置数据流的watermark
        SingleOutputStreamOperator<Event> stream = dataStream.assignTimestampsAndWatermarks(new MyWaterMarkStrategy());

        env.execute();

    }

}

class MyWaterMarkStrategy implements WatermarkStrategy<Event>{

    @Override
    public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
        return new MyWaterMarkGenerator();
    }

    @Override
    public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
        return new MyTimeStampAss();
    }
}

默认的周期性生成WaterMark为200ms

 env.getConfig().setAutoWatermarkInterval(100);