Flink-时间语义以及WaterMark

我会带着你远行 2022-09-09 10:54 180阅读 0赞

# Flink 中的时间语义 #

在 Flink 的流式处理中,会涉及到时间的不同概念,如下图所示

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16]

**Event Time:是事件创建的时间**。它通常由事件中的时间戳描述,例如采集的日志数据中,每一条日志都会记录自己的生成时间,Flink 通过时间戳分配器访问事件时间戳。

**Ingestion Time:是数据进入 Flink 的时间**。

\*\*Processing Time:是每一个执行基于时间操作的算子的本地系统时间,与机器相关,默认的时间属性就是 Processing Time。\*\*处理数据的时间。

例如,一条日志进入 Flink 的时间为 2017-11-12 10:00:00.123,到达 Window 的系统时间为 2017-11-12 10:00:01.234,日志的内容如下:

2017-11-02 18:37:15.624 INFO Fail over to rm2

对于业务来说,要统计 1min 内的故障日志个数,哪个时间是最有意义的?——**eventTime,因为我们要根据日志的生成时间进行统计。**

## EventTime 的引入 ##

\*\*在 Flink 的流式处理 中,绝 大部分的业务 都会使用 eventTime,\*\*一般只在eventTime 无法使用时,才会被迫使用 ProcessingTime 或者 IngestionTime。

如果要使用 EventTime,那么需要引入 EventTime 的时间属性,引入方式如下所示:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment
    
    // 从调用时刻开始给 env 创建的每一个 stream 追加时间特征
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

# Watermark #

我们知道,流处理从事件产生,到流经 source,再到 operator,中间是有一个过程和时间的,虽然大部分情况下,流到 operator 的数据都是按照事件产生的时间顺序来的,但是也不排除由于网络、分布式等原因,导致乱序的产生,所谓乱序,就是指 Flink 接收到的事件的先后顺序不是严格按照事件的 Event Time 顺序排列的。

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 1]

那么此时出现一个问题,一旦出现乱序,如果只根据 eventTime 决定 window 的运行,我们不能明确数据是否全部到位,但又不能无限期的等下去,此时必须要有个机制来保证一个特定的时间后,必须触发 window 去进行计算了,这个特别的机制,就是 Watermark，遇到一个时间戳达到了窗口关闭时间,不应该立刻触发窗口计算,而是等待一段时间,等迟到的数据来了再关闭窗口。

当 Flink 以 Event Time 模式处理数据流时,它会根据数据里的时间戳来处理基于时间的算子；由于网络、分布式等原因,会导致乱序数据的产生；乱序数据会让窗口计算不准确。

Watermark 是一种衡量 Event Time 进展的机制，可以设定延迟触发。

**Watermark 是用 于处理乱序 事件的** ,而正确的处理乱序事件,通常用Watermark 机制结合 window 来实现。

数据流中的 Watermark 用于表示 timestamp 小 于 Watermark 的数据,都已经到达了,因此,window 的执行也是由 Watermark 触发的。

Watermark 可以理解成一个延迟触发机制,我们可以设置 Watermark 的延时时长 t,每次系统会校验已经到达的数据中最大的 maxEventTime,然后认定 eventTime小于 maxEventTime - t 的所有数据都已经到达,如果有窗口的停止时间等于maxEventTime – t,那么这个窗口被触发执行。

watermark 用来让程序自己平衡延迟和结果正确性。

watermark 的特点  
![在这里插入图片描述][e74fbdcbfb8d495b9880b0f077dc3439.png]

watermark 是一条特殊的数据记录

watermark 必须单调递增,以确保任务的事件时间时钟在向前推进,而不是在后退

watermark 与数据的时间戳相关

有序流的 Watermarker 如下图所示:(Watermark 设置为 0)  
![在这里插入图片描述][9f69ace472774136aceb2e3cd3d35a7c.png]

乱序流的 Watermarker 如下图所示:(Watermark 设置为 2)  
![在这里插入图片描述][4f4d7c3867874616b6130fa6ca95446d.png]

当 Flink 接收到数据时,会按照一定的规则去生成 Watermark,这条 Watermark就等于当前所有到达数据中的 maxEventTime - 延迟时长,也就是说,Watermark 是基于数据携带的时间戳生成的,一旦 Watermark 比当前未触发的窗口的停止时间要晚,那么就会触发相应窗口的执行。由于 event time 是由数据携带的,因此,如果运行过程中无法获取新的数据,那么没有被触发的窗口将永远都不被触发。

上图中,我们设置的允许最大延迟到达时间为 2s,所以时间戳为 7s 的事件对应的 Watermark 是 5s,时间戳为 12s 的事件的 Watermark 是 10s,如果我们的窗口 1是 1s~5s,窗口 2 是 6s~10s,那么时间戳为 7s 的事件到达时的 Watermarker 恰好触发窗口 1,时间戳为 12s 的事件到达时的 Watermark 恰好触发窗口 2。

Watermark 就是触发前一窗口的“关窗时间”,一旦触发关门那么以当前时刻为准在窗口范围内的所有所有数据都会收入窗中。

只要没有达到水位那么不管现实中的时间推进了多久都不会触发关窗。

watermark 的传递：

从上游到下游是广播传输，task的事件时钟就是该任务接收的所有分区WM中最小的那个，如果task的事件时钟改变，那么需要将其作为WM广播给下游任务。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_19_color_FFFFFF_t_70_g_se_x_16]

## Watermark 的引入 ##

/** * BoundedOutOfOrdernessTimestampExtractor的构造函数需要传入最大允许的乱序时间，也就是WM的最大延时时间。 */
    inputStream.map(line -> { 
        String[] fields = line.split(",");
        return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
    })
        // 升序数据设置事件时间和WM
        // .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<SensorReading>() { 
        // @Override
        // public long extractAscendingTimestamp(SensorReading event) { 
        // return event.getTimestamp() * 1000L;
        // }
        // })
        // 乱序数据设置时间戳和WM
        .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) { 
            @Override
            public long extractTimestamp(SensorReading event) { 
                // 需要毫秒
                return event.getTimestamp() * 1000L;
            }
        });

Event Time 的使用一定要指定**数据源中的时间戳** 。否则程序无法知道事件的事件时间是什么(数据源里的数据没有时间戳的话,就只能使用 Processing Time 了)。

我们看到上面的例子中创建了一个看起来有点复杂的类,这个类实现的其实就是分配时间戳的接口。Flink 暴露了 TimestampAssigner 接口供我们实现,使我们可以自定义如何从事件数据中抽取时间戳。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
    // 设置事件时间语义
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    
    DataStream<SensorReading> dataStream = env.addSource(new SensorSource())
    	.assignTimestampsAndWatermarks(new MyAssigner());

MyAssigner 有两种类型

AssignerWithPeriodicWatermarks

AssignerWithPunctuatedWatermarks

以上两个接口都继承自 TimestampAssigner。

**AssignerWithPeriodicWatermarks**

周期性的生成 watermark:系统会周期性的将 watermark 插入到流中(水位线也是一种特殊的事件!)。默认周期是 200 毫秒。可以使用ExecutionConfig.setAutoWatermarkInterval()方法 进行设置。

// 每隔 5 秒产生一个 watermark
    env.getConfig.setAutoWatermarkInterval(5000);

产生 watermark 的逻辑:每隔 5 秒钟,Flink 会调用AssignerWithPeriodicWatermarks 的 getCurrentWatermark()方法。如果方法返回一个时间戳大于之前水位的时间戳,新的 watermark 会被插入到流中。这个检查保证了水位线是单调递增的。如果方法返回的时间戳小于等于之前水 位的时间戳,则不会产生新的 watermark。

自定义一个周期性的时间戳抽取:

// 自定义周期性时间戳分配器
    public static class MyPeriodicAssigner implements AssignerWithPeriodicWatermarks<SensorReading> { 
        private Long bound = 60 * 1000L;
        // 延迟一分钟
        private Long maxTs = Long.MIN_VALUE;
        // 当前最大时间戳
        
        @Nullable
        @Override
        public Watermark getCurrentWatermark() { 
            return new Watermark(maxTs - bound);
        }
        
        @Override
        public long extractTimestamp(SensorReading element, long previousElementTimestamp){ 
            maxTs = Math.max(maxTs, element.getTimestamp());
            return element.getTimestamp();
        }
    }

一种简单的特殊情况是,如果我们事先得知数据流的时间戳是单调递增的,也就是说没有乱序,那我们可以使用 **AscendingTimestampExtractor** ,这个类会直接使用数据的时间戳生成 watermark。对于排好序的数据,不需要延迟触发,可以只指定时间戳就行了。

DataStream<SensorReading> dataStream = ...
    dataStream.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<SensorReading>() { 
        @Override
        public long extractAscendingTimestamp(SensorReading element) { 
            return element.getTimestamp() * 1000;
        }
    });

升序和前面乱序的处理 BoundedOutOfOrdernessTimestampExtractor,都是基于周期性 watermark 的

**AssigneWithPunctuatedWatermarks**

间断式地生成 watermark。和周期性生成的方式不同,这种方式不是固定时间的,而是可以根据需要对每条数据进行筛选和处理。直接上代码来举个例子,我们只给sensor\_1 的传感器的数据流插入 watermark，即只有sensor\_1数据来时才会去生成WM。

public static class MyPunctuatedAssigner implements AssignerWithPunctuatedWatermarks<SensorReading>{ 
        private Long bound = 60 * 1000L;
        // 延迟一分钟
        
        @Nullable
        @Override
        public Watermark checkAndGetNextWatermark(SensorReading lastElement, long
                                                  extractedTimestamp) { 
            if(lastElement.getId().equals("sensor_1"))
                return new Watermark(extractedTimestamp - bound);
            else
                return null;
        }
        
        @Override
        public long extractTimestamp(SensorReading element, long previousElementTimestamp)
        { 
            return element.getTimestamp();
        }
    }

watermark 的设定

如果watermark设置的延迟太久（实时性差）,收到结果的速度可能就会很慢,解决办法是在水位线到达之前输出一个近似结果，然后设置迟到时间，到达迟到时间之后再次进行数据整合。

而如果watermark到达得太早（准确性差）,则可能收到错误结果,不过 Flink 处理迟到数据的机制可以解决这个问题

数据稀疏的情况用AssigneWithPunctuatedWatermarks，数据稠密的情况用AssignerWithPeriodicWatermarks，默认是基于周期的处理。

## EventTime 在 window 中的使用(Java) ##

**基本测试**

public static void main(String[] args) throws Exception { 
    
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
    
        // 从Socket读取
        DataStream<String> inputStream = env.socketTextStream("localhost", 4444);
    
    
        /** * BoundedOutOfOrdernessTimestampExtractor的构造函数需要传入最大允许的乱序时间，也就是WM的最大延时时间。 */
        SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(line -> { 
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) { 
            @Override
            public long extractTimestamp(SensorReading event) { 
                // 需要毫秒
                return event.getTimestamp() * 1000L;
            }
        });
    
        // 基于事件时间的开窗聚合，统计15s内温度的最小值
        SingleOutputStreamOperator<SensorReading> minTempStream = dataStream.keyBy("id")
            .timeWindow(Time.seconds(15))
    // .window(TumblingEventTimeWindows.of(Time.seconds(15), Time.seconds(3))) // 设置偏移量为3的滚动窗口
            .minBy("temperature");
    
        minTempStream.print("minTemp");
    
        // 执行
        env.execute();
    }
    
    输入数据：
    lxj@lxj:~$ nc -lk 4444
    sensor_1,1547718199,35.8
    sensor_6,1547718201,15.4
    sensor_7,1547718202,6.7
    sensor_10,1547718205,38.1
    sensor_1,1547718206,36.3                
    sensor_1,1547718207,36.5
    sensor_1,1547718210,32.8
    sensor_1,1547718211,36.9
    sensor_1,1547718212,34.7
    sensor_1,1547718213,37
    sensor_1,1547718224,31
    sensor_1,1547718225,21
    sensor_1,1547718226,11
    sensor_1,1547718227,1
    
        
    打印数据：
    minTemp> SensorReading{ id='sensor_1', timestamp=1547718199, temperature=35.8}
    minTemp> SensorReading{ id='sensor_6', timestamp=1547718201, temperature=15.4}
    minTemp> SensorReading{ id='sensor_10', timestamp=1547718205, temperature=38.1}
    minTemp> SensorReading{ id='sensor_7', timestamp=1547718202, temperature=6.7}
    minTemp> SensorReading{ id='sensor_1', timestamp=1547718224, temperature=31.0}

**解析：**

窗口大小为15，前闭后开，WM最大延时为2

窗口的初始位置：195

public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) { 
        return timestamp - (timestamp - offset + windowSize) % windowSize;
    }
    
    // timestamp，时间戳，第一个时间戳
    // offset，偏移量，默认是0
    // windowSize，窗口大小
    // 初始位置的话：199 - (199 - 0 + 15) % 15 = 195
    // 初始位置是窗口大小的整数倍，是第一个时间戳中最大的窗口大小的整数倍

第一个窗口是\[195,210)，因此触发窗口时间为212,212来了之后就会将该窗口关闭；

第二个窗口是\[210,215)，因此触发窗口时间为227，而取值为210-214，因此做小温度为31。

偏移量的用法主要是在处理北京时间和标准时间的时候用的。

**以上是并行度为1的情况的，若并行度不为1，则需要考虑并行度情况，并且任务的事件时钟去的是上游所有分区的WM的最小值。**

因此WM生成的地方最好离source越近越好。最好是在读取数据的时候就直接生成好WM，如果是Kafka作为数据源的话，可以在Kafka就直接生成好WM。

**迟到测试**

在上面代码的基础上进行改造

// 基于事件时间的开窗聚合，统计15s内温度的最小值
    OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("late") { 
    };
    
    SingleOutputStreamOperator<SensorReading> minTempStream = dataStream.keyBy("id")
        .timeWindow(Time.seconds(15))
        .allowedLateness(Time.minutes(1))
        .sideOutputLateData(outputTag)
        .minBy("temperature");
    
    minTempStream.print("minTemp");
    
    minTempStream.getSideOutput(outputTag).print("late");
    
    lxj@lxj:~$ nc -lk 4444
    sensor_1,1547718199,35.8
    sensor_1,1547718206,36.3
    sensor_1,1547718210,32.8
    sensor_1,1547718211,36.9
    sensor_1,1547718209,31
    sensor_1,1547718212,34.7
    sensor_1,1547718213,37
    sensor_1,1547718209,30
    sensor_1,1547718209,31
    sensor_1,1547718270,20
    sensor_1,1547718206,29
    sensor_1,1547718272,60
    sensor_1,1547718206,28
        
    minTemp> SensorReading{ id='sensor_1', timestamp=1547718209, temperature=31.0}
    minTemp> SensorReading{ id='sensor_1', timestamp=1547718209, temperature=30.0}
    minTemp> SensorReading{ id='sensor_1', timestamp=1547718209, temperature=30.0}
    minTemp> SensorReading{ id='sensor_1', timestamp=1547718210, temperature=32.8}
    minTemp> SensorReading{ id='sensor_1', timestamp=1547718206, temperature=29.0}
    late> SensorReading{ id='sensor_1', timestamp=1547718206, temperature=28.0}

**解析**：

允许数据迟到1分钟，那么在窗口被激活关闭的时候不会立即关闭窗口，而是立即输出一个结果，并等待迟到数据，迟到数据来一条输出一条记录  
一分钟到了之后，窗口被关闭，那么之后来的数据放到侧输出流当中，进行处理

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/68ad76d86e5f4733ae7373ec1f14bde4.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/0897ebeaf26646d494c52edb08858739.png
[e74fbdcbfb8d495b9880b0f077dc3439.png]: /images/20220829/c14de5047240459a88988e868d03321a.png
[9f69ace472774136aceb2e3cd3d35a7c.png]: /images/20220829/a690a7a3bb0a4d83a58c75341a772279.png
[4f4d7c3867874616b6130fa6ca95446d.png]: /images/20220829/da40e41cc5f74f248e27e878ac69e6b4.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_19_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/6c9e6d42c44049aa803f4a0fef1e20c5.png