四、flink--window、eventTime和wate
一、flink的window机制
1.1 window概述
streaming流式计算是一种被设计用于处理无限数据集的数据处理引擎,而无限数据集是指一种不断增长的本质上无限的数据集,而window是一种切割无限数据为有限块进行处理的手段。
Window是无限数据流处理的核心,Window将一个无限的stream拆分成有限大小的"buckets"桶,我们可以在这些桶上做计算操作。
1.2 window的类型
window可以分为两大类:
CountWindow:按照指定的数据条数生成一个Window,与时间无关。比较少用
TimeWindow:按照时间生成Window。非常常用,下面主要将时间窗口有哪些类型。主要有四类:滚动窗口(Tumbling Window)、滑动窗口(Sliding Window)、会话窗口(Session Window)和全局窗口(global window比较少用 )。
1.2.1 滚动窗口(Tumbling Windows)
概述:将数据依据固定的窗口长度对数据进行切片。只有一个工作参数,就是窗口大小
特点:时间对齐,窗口长度固定,没有重叠。
滚动窗口分配器将每个元素分配到一个指定窗口大小的窗口中,滚动窗口有一个固定的大小,并且不会出现重叠(前后时间点都是紧接着的)。例如:如果你指定了一个5分钟大小的滚动窗口,窗口的创建如下图所示:
图 1.2.1 滚动窗口
适用场景:适合做BI统计等(做每个时间段的聚合计算)。
1.2.2 滑动窗口(Sliding Windows)
概述:滑动窗口是固定窗口的更广义的一种形式,滑动窗口工作参数由固定的窗口长度和滑动间隔组成。
特点:时间对齐,窗口长度固定,有重叠。
滑动窗口分配器将元素分配到固定长度的窗口中,与滚动窗口类似,窗口的大小由窗口大小参数来配置,另一个窗口滑动参数控制滑动窗口开始的频率。因此,滑动窗口如果滑动参数小于窗口大小的话,窗口是可以重叠的,在这种情况下元素会被分配到多个窗口中。
例如,你有10分钟的窗口和5分钟的滑动,那么每个窗口中5分钟的窗口里包含着上个10分钟产生的数据,如下图所示:
图 1.2.2 滑动窗口
适用场景:对最近一个时间段内的统计(求某接口最近5min的失败率来决定是否要报警)。
1.2.3 会话窗口(Session Windows)
概述:由一系列事件组合一个指定时间长度的timeout间隙组成,类似于web应用的session,也就是一段时间没有接收到新数据就会生成新的窗口。
特点:时间无对齐。窗口无固定长度
session窗口分配器通过session活动来对元素进行分组,session窗口跟滚动窗口和滑动窗口相比,不会有重叠和固定的开始时间和结束时间的情况,相反,当它在一个固定的时间周期内不再收到元素,即非活动间隔产生,那个这个窗口就会关闭。一个session窗口通过一个session间隔来配置,这个session间隔定义了非活跃周期的长度,当这个非活跃周期产生,那么当前的session将关闭并且后续的元素将被分配到新的session窗口中去。
图1.2.3 会话窗口
1.3 window窗口api
1.3.1 window api分类
window数据源分为两种,一种是典型的KV类型(keyedStream),另一种是非KV类型(Non-keyedStream)。
区别:
keyedStream:
需要在使用窗口操作前,调用 keyBy对KV按照key进行分区,然后才可以调用window操作的api,比如 countWindow,timeWindow等
Non-keyedstream:
如果使用窗口操作前,没有使用keyBy算子,那么就认为是Non-keyedstream,调用的window api就是 xxxWindowAll,比如countWindowAll,timeWindowAll,而且因为是非KV,所以无法分区,也就是只有一个分区,那么这个窗口并行度只能是1。这个是要注意的。
1.3.2 countWindow
CountWindow根据窗口中相同key元素的数量来触发执行,执行时只计算元素数量达到窗口大小的key对应的结果。
有两个用法:countWindow(window_size):只指定窗口大小,此时窗口是滚动窗口countWindow(window_size, slide):指定窗口大小以及滑动间隔,此时窗口是滑动窗口
注意:CountWindow的window_size指的是相同Key的元素的个数,不是输入的所有元素的总数。
1、滚动窗口
默认的CountWindow是一个滚动窗口,只需要指定窗口大小即可,当元素数量达到窗口大小时,就会触发窗口的执行。
import org.apache.flink.api.common.functions.FlatMapFunction;import org.apache.flink.api.common.functions.ReduceFunction;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.streaming.api.datastream.DataStreamSource;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.util.Collector;public class WindowTest { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource source = env.readTextFile("/test.txt"); source.flatMap(new FlatMapFunction>() { @Override public void flatMap(String s, Collector> collector) throws Exception { for (String s1 : s.split(" ")) { collector.collect(new Tuple2<>(s1, 1)); } } }).keyBy(0).countWindow(5).reduce(new ReduceFunction>() { @Override public Tuple2 reduce(Tuple2 t1, Tuple2 t2) throws Exception { return new Tuple2<>(t1.f0, t1.f1 + t2.f1); } }).print(); env.execute("滚动窗口"); }}
2、滑动窗口
动窗口和滚动窗口的函数名是完全一致的,只是在传参数时需要传入两个参数,一个是window_size,一个是sliding_size。
下面代码中的sliding_size设置为了2,也就是说,每收到两个相同key的数据就计算一次,每一次计算的window范围是5个元素。
import org.apache.flink.api.common.functions.FlatMapFunction;import org.apache.flink.api.common.functions.ReduceFunction;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.streaming.api.datastream.DataStreamSource;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.util.Collector;public class WindowTest { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource source = env.readTextFile("/test.txt"); source.flatMap(new FlatMapFunction>() { @Override public void flatMap(String s, Collector> collector) throws Exception { for (String s1 : s.split(" ")) { collector.collect(new Tuple2<>(s1, 1)); } } }).keyBy(0).countWindow(5,2).reduce(new ReduceFunction>() { @Override public Tuple2 reduce(Tuple2 t1, Tuple2 t2) throws Exception { return new Tuple2<>(t1.f0, t1.f1 + t2.f1); } }).print(); env.execute("滑动窗口"); }}
1.3.3 timeWindow
TimeWindow是将指定时间范围内的所有数据组成一个window,一次对一个window里面的所有数据进行计算。同样支持类似上面的滚动窗口和滑动窗口模式。有两个工作参数:window_size和slide。只指定window_size时是滚动窗口。
1、滚动窗口
Flink默认的时间窗口根据Processing Time 进行窗口的划分,将Flink获取到的数据根据进入Flink的时间划分到不同的窗口中。
import org.apache.flink.api.common.functions.FlatMapFunction;import org.apache.flink.api.common.functions.ReduceFunction;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.streaming.api.datastream.DataStreamSource;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.windowing.time.Time;import org.apache.flink.util.Collector;public class WindowTest { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource source = env.readTextFile("/test.txt"); source.flatMap(new FlatMapFunction>() { @Override public void flatMap(String s, Collector> collector) throws Exception { for (String s1 : s.split(" ")) { collector.collect(new Tuple2<>(s1, 1)); } } }).keyBy(0).timeWindow(Time.seconds(2)).reduce(new ReduceFunction>() { @Override public Tuple2 reduce(Tuple2 t1, Tuple2 t2) throws Exception { return new Tuple2<>(t1.f0, t1.f1 + t2.f1); } }).print(); env.execute("滚动窗口"); }}
2、滑动窗口
和上面类似,就是参数里面增加了slide参数,也就是滑动时间间隔。时间间隔可以通过Time.milliseconds(x),Time.seconds(x),Time.minutes(x)等其中的一个来指定。
1.3.4 window reduce
也就是在窗口算子之后执行reduce算子,用法和普通的reduce一样,只不过reduce的单位是一个窗口。即每一个窗口返回一次reduce结果。程序在上面,不重复了。
1.3.5 window fold
也就是在窗口算子之后执行fold算子,用法和普通的fold一样,只不过fold的单位是一个窗口。即每一个窗口返回一次reduce结果。程序在上面,不重复了。
1.3.6 window聚合操作
指的是max、min等这些聚合算子,只不过是在window算子之后使用,以窗口为单位,每一个窗口返回一次聚合结果,而不是像普通那样,每一次聚合结果都返回。
二、time、watermark和window
2.1 flink中 time的分类
在flink中,time有不同分类,如下:
Event Time:
是事件创建的时间。它通常由事件中的时间戳描述,例如采集的日志数据中,每一条日志都会记录自己的生成时间,Flink通过时间戳分配器访问事件时间戳。
Ingestion Time:
是数据进入Flink的时间。
Processing Time:
是每一个执行基于时间操作的算子的本地系统时间,与机器相关,默认的时间属性就是Processing Time。也就是数据被处理时的当前时间。
这些时间有什么不同呢?因网络传输需要时间,所以Ingestion Time不一定和Event Time相等,很多情况下是不等的。同样Processing Time表示数据处理时的时间,如果数据是很久之前采集的,现在才处理,那么很明显,三个时间time都不会相等的。
图 2.1 flink--时间的概念
例子:
一条日志进入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
可以看到,三个time都不相等。而对于业务来说,要统计1min内的故障日志个数,哪个时间是最有意义的?-- eventTime,因为我们要根据日志的生成时间进行统计。但是flink默认的窗口的时间是Processing Time,那么如何引入eventTime呢?
2.2 eventTime的引入
在Flink的流式处理中,绝大部分的业务都会使用eventTime,一般只在eventTime无法使用时,才会被迫使用ProcessingTime或者IngestionTime。默认使用的是ProcessingTime。那么如何指定flink使用指定的time呢?
2.2.1 引入方式1:设置env时间类型
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setStreamTimeCharacteristic(时间类型);//三种类型的time对应如下:TimeCharacteristic.EventTime; eventtimeTimeCharacteristic.IngestionTime; 到达flink的时间TimeCharacteristic.ProcessingTime; 处理数据的时间
这种方式是整个env全局生效的,是直接将env默认的时间设置为eventtime。后面的窗口操作默认就会使用eventtime作为时间依据。如果想不同的窗口设置不同的时间类型,这种方式就行不通了。
2.2.2 引入方式2:单独设置window的实际类型
stream.window(TumblingEventTimeWindows.of(Time.seconds(5))).window这个api就是所有窗口总的api,其他窗口api都是通过这个api封装出来的。可以通过这个总api,参数直接窗口的类型,比如上面的就是指定eventtime 的timewindow,这样并不会影响整个env的时间类型。同样的,其他时间类型窗口,比如:SlidingEventTimeWindows 滑动eventtime窗口基本上看名字就知道是什么时间类型(三大时间类型)、以及什么类型(滑动、滚动、会话窗口)的窗口了。注意:eventtime没有session窗口,processingTime和
2.3 watermark的原理
2.3.1 引入背景
我们知道,流处理从事件产生,到流经source,再到operator,中间是有一个过程和时间的,虽然大部分情况下,流到operator的数据都是按照事件产生的时间顺序来的,但是也不排除由于网络、背压等原因,导致乱序的产生,所谓乱序,就是指Flink接收到的事件的先后顺序不是严格按照事件的Event Time顺序排列的。
图 2.3 数据的乱序
那么此时出现一个问题,一旦出现乱序,如果只根据eventTime决定window的运行,我们不能明确数据是否全部到位,但又不能无限期的等下去,此时必须要有个机制来保证一个特定的时间后,必须触发window去进行计算了,这个特别的机制,就是Watermark。
解释:
如果只按照到达的event的eventtime来触发窗口操作,假设有event1~5。如果到达顺序是乱的,比如event5最先达到,然后event1也达到了,那么flink这边怎么知道这中间还有没有数据呢?没办法的,不能确定数据是否完整到达,也不能无限制等待下去。所以需要一种机制来处理这种情况。
2.3.2 watermark机制原理
Watermark是一种衡量Event Time进展的机制,它是数据本身的一个隐藏属性,数据本身携带着对应的Watermark。Watermark是用于处理乱序事件的,而正确的处理乱序事件,通常用Watermark机制结合window来实现。
数据流中的Watermark用于表示timestamp小于Watermark的数据,都已经到达了,因此,window的执行也是由Watermark触发的。
Watermark可以理解成一个延迟触发机制,我们可以设置Watermark的延时时长t,每次系统会校验已经到达的数据中最大的maxEventTime,然后认定eventTime小于maxEventTime - t的所有数据都已经到达,如果有窗口的停止时间等于maxEventTime - t,那么这个窗口被watermark触发执行。
解释:
watermark是一种概率性的机制。假设event1~5,如果event5已经到达了,那么其实按照event产生的先后顺序,正常情况下,前面的event1~4应该也到达了。而为了保证前面的event1~4的到达(其实是更多的到达,但是不一定全部都到达),在event5到达了之后,提供一定的延迟时间t。当event5到达,且经过 t 时间之后,正常情况下,前面的event1~4 大概率会到达了,如果没有到达,属于少数情况,那么就认为event5之前的event都到达了,无论是否真的全部到达了。如果在延迟时间之后到达了,这个旧数据直接会被丢弃。所以其实watermark就是一种保障更多event乱序到达的机制,提供了一定的延时机制,而因为只会延迟一定的时间,所以也不会导致flink无限期地等待下去。
有序数据流的watermark如下:(watermark设置为0)
图 2.4 有序数据流的watermark
乱序数据流的watermark如下:(watermark设置为2)
图 2.5 乱序数据流的watermark
当Flink接收到每一条数据时,都会产生一条Watermark,这条Watermark就等于当前所有到达数据中的maxEventTime - 延迟时长t,也就是说,Watermark是由数据携带的,一旦数据携带的Watermark比当前未触发的窗口的停止时间要晚,那么就会触发相应窗口的执行。由于Watermark是由数据携带的,因此,如果运行过程中无法获取新的数据,那么没有被触发的窗口将永远都不被触发。
上图中,我们设置的允许最大延迟到达时间为2s,所以时间戳为7s的事件对应的Watermark是5s,时间戳为12s的事件的Watermark是10s,如果我们的窗口1是1s~5s,窗口2是6s~10s,那么时间戳为7s的事件到达时的Watermarker恰好触发窗口1,时间戳为12s的事件到达时的Watermark恰好触发窗口2。
Window会不断产生,属于这个Window范围的数据会被不断加入到Window中,所有未被触发的Window都会等待触发,只要Window还没触发,属于这个Window范围的数据就会一直被加入到Window中,直到Window被触发才会停止数据的追加,而当Window触发之后才接受到的属于被触发Window的数据会被丢弃。如果产生的窗口中没有新到的数据,也就不会有watermark,那么窗口就不会被触发计算。
2.3.3 watermark的触发计算的条件
watermark时间(max_eventTime-t) >= window_end_time;
在[window_start_time,window_end_time)中有数据存在。
2.3.4 watermark的产生方式
Punctuated:不间断产生
数据流中每一个递增的EventTime都会产生一个Watermark。
在实际的生产中Punctuated方式在TPS很高的场景下会产生大量的Watermark在一定程度上对下游算子造成压力,所以只有在实时性要求非常高的场景才会选择Punctuated的方式进行Watermark的生成。
Periodic:周期性产生
周期性的(一定时间间隔或者达到一定的记录条数)产生一个Watermark。
在实际的生产中Periodic的方式必须结合时间和积累条数两个维度继续周期性产生Watermark,否则在极端情况下会有很大的延时。
这两种有不同的api实现,下面会讲
2.4 watermark的引入以及接口
2.4.1 watermark引入
需要先引入eventime,然后引入watermark
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);DataStreamSource source = env.readTextFile("/test.txt");//引入的watermark的实现类source.assignTimestampsAndWatermarks(xx)
watermark的实现有两大类,对应上面的两种watermark的产生方式,有两个接口:
AssignerWithPeriodicWatermarks; 周期性产生watermark,即PeriodAssignerWithPunctuatedWatermarks; Punctuated:不间断产生
2.4.2 AssignerWithPeriodicWatermarks接口
看看AssignerWithPeriodicWatermarks这个接口的源码,主要用于周期性产生watermark
public interface AssignerWithPeriodicWatermarks extends TimestampAssigner { //获取当前的watermark @Nullable Watermark getCurrentWatermark();}//父接口===================public interface TimestampAssigner extends Function { //获取当前的时间戳 long extractTimestamp(T var1, long var2);}
主要就是有两个方法需要覆盖,getCurrentWatermark()用于生成watermark,extractTimestamp用于获取每个event的timestamp。
由于这是一个周期性产生watermark的接口,所以需要指定这个生成周期有多长,需要env的配置中指定,如:
env.getConfig().setAutoWatermarkInterval(n ms);记住间隔时间单位是毫秒
例子:
/*根据eventTime 创建处理watermark*/public class BoundedOutOfOrdernessGenerator implements AssignerWithPeriodicWatermarks { //watermark延迟时间 t,单位是毫秒 private final long maxOutOfOrderness = 3500; // 3.5 seconds //保存当前最大的时间戳 private long currentMaxTimestamp; //根据传递进来的event,获取time,然后如果比当前最大的time还大,就替换,否则保持。因为数据乱序到达是无法保证时间是递增的 @Override public long extractTimestamp(MyEvent element, long previousElementTimestamp) { long timestamp = element.getCreationTime(); currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp); return timestamp; } //返回watermark @Override public Watermark getCurrentWatermark() { // return the watermark as current highest timestamp minus the out-of-orderness bound return new Watermark(currentMaxTimestamp - maxOutOfOrderness); }}
再加上设置的setAutoWatermarkInterval(n ms),就可以周期性生成watermark。
2.4.3 AssignerWithPunctuatedWatermarks接口
看看AssignerWithPunctuatedWatermarks这个接口的源码,主要用于实时产生watermark
public interface AssignerWithPunctuatedWatermarks extends TimestampAssigner { //获取最新的watermark @Nullable Watermark checkAndGetNextWatermark(T var1, long var2);}//父接口public interface TimestampAssigner extends Function { //从event中获取timestamp long extractTimestamp(T var1, long var2);}
写法其实和上面的类似,只是这里不会设置生成watermark的时间间隔
2.4.4 flink自带的watermark实现类
1、BoundedOutOfOrdernessTimestampExtractor
继承了AssignerWithPeriodicWatermarks接口的一个类,看看它的源码
package org.apache.flink.streaming.api.functions.timestamps;import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;import org.apache.flink.streaming.api.watermark.Watermark;import org.apache.flink.streaming.api.windowing.time.Time;public abstract class BoundedOutOfOrdernessTimestampExtractor implements AssignerWithPeriodicWatermarks { private static final long serialVersionUID = 1L; private long currentMaxTimestamp; private long lastEmittedWatermark = -9223372036854775808L; private final long maxOutOfOrderness; //构造方法中接收一个参数,就是延迟时间 t public BoundedOutOfOrdernessTimestampExtractor(Time maxOutOfOrderness) { if (maxOutOfOrderness.toMilliseconds() < 0L) { throw new RuntimeException("Tried to set the maximum allowed lateness to " + maxOutOfOrderness + ". This parameter cannot be negative."); } else { this.maxOutOfOrderness = maxOutOfOrderness.toMilliseconds(); this.currentMaxTimestamp = -9223372036854775808L + this.maxOutOfOrderness; } } public long getMaxOutOfOrdernessInMillis() { return this.maxOutOfOrderness; } //需要重写的方法,用于获取timestamp public abstract long extractTimestamp(T var1); //获取watermark的方法已经写好了,用传递进来的延迟时间t来计算得出watermark public final Watermark getCurrentWatermark() { long potentialWM = this.currentMaxTimestamp - this.maxOutOfOrderness; if (potentialWM >= this.lastEmittedWatermark) { this.lastEmittedWatermark = potentialWM; } return new Watermark(this.lastEmittedWatermark); } public final long extractTimestamp(T element, long previousElementTimestamp) { long timestamp = this.extractTimestamp(element); if (timestamp > this.currentMaxTimestamp) { this.currentMaxTimestamp = timestamp; } return timestamp; }}
这个类就是实现了用户可以自定义设定延迟时间t 的一个watermark。
2、AscendingTimestampExtractor
也是继承了AssignerWithPeriodicWatermarks接口的一个类。具有稳定的递增时间戳的数据源,比如kafka的分区数据,每一条信息都是递增+1的,适用于这个类。只需要重写
extractAscendingTimestamp方法。
2.5 eventTime、window和watermark结合使用例子
package flinktest;import org.apache.flink.api.common.functions.FlatMapFunction;import org.apache.flink.api.common.functions.MapFunction;import org.apache.flink.api.common.functions.ReduceFunction;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.streaming.api.TimeCharacteristic;import org.apache.flink.streaming.api.datastream.DataStreamSource;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;import org.apache.flink.streaming.api.windowing.time.Time;import org.apache.flink.util.Collector;public class EventTimeTest { public static void main(String[] args) { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); env.getConfig().setAutoWatermarkInterval(1000); DataStreamSource source = env.readTextFile("/tmp/test.txt"); source.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor(Time.milliseconds(3000)) { @Override public long extractTimestamp(String s) { return Integer.valueOf(s.split(" ")[0]); } }).flatMap(new FlatMapFunction>() { @Override public void flatMap(String s, Collector> collector) throws Exception { Tuple2 tmpTuple = new Tuple2<>(); for (String s1 : s.split(" ")) { tmpTuple.setFields(s1, 1); collector.collect(tmpTuple); } } }).keyBy(0) .timeWindow(Time.seconds(10)) .reduce(new ReduceFunction>() { @Override public Tuple2 reduce(Tuple2 t1, Tuple2 t2) throws Exception { return new Tuple2<>(t1.f0, t1.f1 + t2.f1); } }) .print(); try { env.execute("eventtime test"); } catch (Exception e) { e.printStackTrace(); } }}
window api的类继承结构