时间语义与窗口

我不是女神ヾ 2023-10-15 15:30 35阅读 0赞

##### 时间语义 #####

在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中，窗口在同一时刻是可以存在多个，而不是只有一个。

![在这里插入图片描述][22cdb55d23944d30bd416e7f9e134c4c.png]

##### 数据流的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);

[22cdb55d23944d30bd416e7f9e134c4c.png]: https://img-blog.csdnimg.cn/22cdb55d23944d30bd416e7f9e134c4c.png