当前位置:   article > 正文

Flink(window time watermark)_flink 计算同一个事件的时间差

flink 计算同一个事件的时间差

https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/stream/operators/windows.html

1 Time与Window

1.1 Flink中涉及的时间

  • Event Time:是事件创建的时间。它通常由事件中的时间戳描述,例如采集的日志数据中,每⼀条⽇志都会记录⾃己的生成时间,Flink通过时间戳分配器访问事件时间戳。
  • Ingestion Time:是数据进入Flink的时间。
  • Processing Time:是每⼀个执行基于时间操作的算子的本地系统时间,与机器相关,默认的时间属性就是Processing Time。 

1.2 Window

1.2.1 Window描述

streaming流式计算是⼀种被设计用于处理⽆限数据集的数据处理引擎,而⽆限数据集是指一种不断增长的本质上无限数据集,⽽window是一种切割无限数据为有限块进行处理的手段。Window是无限数据流处理的核心,Window将⼀个⽆限stream拆分成有限大小的”buckets”桶,我们可以在这些桶上做计算操作。

1.2.2 Window类型

Window可以分成两类:

  • CountWindow:按照指定的数据条数生成⼀个Window,与时间无关。
  • TimeWindow:按照时间生成Window。

对于TimeWindow,可以根据窗⼝实现原理的不同分成三类:滚动窗⼝(Tumbling Window)、滑动
窗⼝(Sliding Window)和会话窗⼝(Session Window)。

  • 滚动窗⼝(Tumbling Windows)

将数据依据固定的窗⼝⻓度对数据进行切片。
特点:时间对⻬,窗口⻓度固定,没有重叠。

滚动窗⼝分配器将每个元素分配到⼀个指定窗⼝⼤小的窗口中,滚动窗口有一个固定的大小,并且不会出现重叠。例如:如果你指定了一个5分钟大小的滚动窗口,⼝口的创建如下图所示:

适用场景:适合做BI统计等(做每个时间段的聚合计算)。

  • 滑动窗⼝(Sliding Windows)

滑动窗⼝是固定窗口的更⼴义的⼀种形式,滑动窗口由固定的窗口长度和滑动间隔组成。
特点:时间对齐,窗口长度固定,有重叠

滑动窗口分配器分配元件以固定长度的窗口。与翻滚窗口分配器类似,窗口大小窗口大小参数配置。附加的窗口滑动参数控制滑动窗口的启动频率。因此,如果幻灯片小于窗口大小,则滑动窗口可以重叠。在这种情况下,元素被分配给多个窗口。

例如,您可以将大小为10分钟的窗口滑动5分钟。有了这个,你每隔5分钟就会得到一个窗口,其中包含过去10分钟内到达的事件,如下图所示。

适⽤场景:对最近⼀个时间段内的统计(求某接口最近5min的失败率来决定是否要报警)。

  • 会话窗⼝(Session Windows)

由⼀系列事件组合⼀个指定时间长度的timeout间隙组成,类似于web应用的session,也就是一段时间没有接收到新数据就会生成新的窗口。
特点:时间⽆对⻬。

会话窗口中按活动会话分配器组中的元素。会话窗口不重叠,没有固定的开始和结束时间,与翻滚窗口滑动窗口相反。相反,当会话窗口在一段时间内没有接收到元素时,当发生不活动的间隙时,会关闭会话窗口。会话窗口分配器可以配置静态会话间隙或 会话间隙提取器功能,该功能定义不活动时间段的长度。当此期限到期时,当前会话将关闭,后续元素将分配给新的会话窗口。

1.3 Window API

1.3.1 Count Window

Count Window根据窗⼝中相同key元素的数量来触发执行,执行时只计算元素数量达到窗⼝⼤小的key对应的结果**。

注意:CountWindow的window_size指的是相同Key的元素的个数,不是输入的所有元素的总数。

  • 滚动窗口

默认的CountWindow是⼀个滚动窗⼝,只需要指定窗⼝⼤小即可,当元素数量达到窗口⼤小时,就会触发窗⼝的执⾏。

  1. package com.wedoctor.flink.window
  2. import org.apache.flink.api.java.tuple.Tuple
  3. import org.apache.flink.streaming.api.scala._
  4. import org.apache.flink.streaming.api.windowing.windows.GlobalWindow
  5. /**
  6. * 目的:测试分组后的CountWindow
  7. * CountWindow是以条数划分Window(逻辑划分的一个部分数据,然后对这些数据进行处理)
  8. *
  9. */
  10. object CountWindowDemo2 {
  11. def main(args: Array[String]): Unit = {
  12. val env = StreamExecutionEnvironment.getExecutionEnvironment
  13. val lines: DataStream[String] = env.socketTextStream("localhost", 8888)
  14. val wordAndOne = lines.map((_, 1))
  15. val keyed: KeyedStream[(String, Int), Tuple] = wordAndOne.keyBy(0)
  16. //KeyedStream才可以调用countWindow
  17. val window = keyed.countWindow(5)
  18. val result: DataStream[(String, Int)] = window.sum(1)
  19. result.print()
  20. env.execute(this.getClass.getSimpleName)
  21. }
  22. }
  • 滑动窗口

 滑动窗⼝和滚动窗口的函数名是完全一致的,只是在传参数时需要传入两个参数,一个是window_size,一个是sliding_size。
下⾯代码中的sliding_size设置为了2,也就是说,每收到两个相同key的数据就计算一次,每一次计算的window范围是5个元素。

  1. // 当相同key的元素个数达到2个时,触发窗口计算,计算的窗口范围为5
  2. val streamWindow = streamKeyBy.countWindow(5,2)

 

1.3.2 TimeWindow

TimeWindow是将指定时间范围内的所有数据组成⼀个window,⼀次对一个window⾥面的所有数据进行计算。

  • 滚动窗口

Flink默认的时间窗⼝根据Processing Time 进⾏窗⼝的划分,将Flink获取到的数据根据进入Flink的时间划分到不同的窗口中。

  1. package com.wedoctor.flink.window
  2. import org.apache.flink.api.java.tuple.Tuple
  3. import org.apache.flink.streaming.api.scala._
  4. import org.apache.flink.streaming.api.windowing.time.Time
  5. import org.apache.flink.streaming.api.windowing.windows.TimeWindow
  6. import org.apache.log4j.{Level, Logger}
  7. object TumblingTimeWindow {
  8. Logger.getLogger("org").setLevel(Level.ERROR)
  9. def main(args: Array[String]): Unit = {
  10. val env = StreamExecutionEnvironment.getExecutionEnvironment
  11. val lines: DataStream[String] = env.socketTextStream("localhost",8888)
  12. val words: DataStream[String] = lines.flatMap(_.split(" "))
  13. var wordAndOne: DataStream[(String, Int)] = words.map((_, 1))
  14. val keyed = wordAndOne.keyBy(0)
  15. //分组后可以调用timeWindow
  16. val window: WindowedStream[(String, Int), Tuple, TimeWindow] = keyed.timeWindow(Time.seconds(5))
  17. val result = window.sum(1)
  18. result.print()
  19. env.execute(this.getClass.getSimpleName)
  20. }
  21. }

时间间隔可以通过Time.milliseconds(x),Time.seconds(x),Time.minutes(x)等其中的⼀个来指定。

  • 滑动窗口

滑动窗口和滚动窗口的函数名是完全一致的,只是在传参数时需要传入两个参数,一个是window_size,一个是sliding_size。

下⾯代码中的sliding_size设置为了2s,也就是说,窗口每2s就计算一次,每一次计算的window范围是5s内的所有元素。

  1. package com.wedoctor.flink.window
  2. import org.apache.flink.streaming.api.scala._
  3. import org.apache.flink.streaming.api.windowing.time.Time
  4. import org.apache.flink.streaming.api.windowing.windows.TimeWindow
  5. import org.apache.log4j.{Level, Logger}
  6. object SlideTimeWindow {
  7. Logger.getLogger("org").setLevel(Level.ERROR)
  8. def main(args: Array[String]): Unit = {
  9. val env = StreamExecutionEnvironment.getExecutionEnvironment
  10. val lines = env.socketTextStream("localhost",8888)
  11. val nums = lines.map(_.toInt)
  12. //如果是timeWindowAll 则不需要分组
  13. val window: AllWindowedStream[Int, TimeWindow] = nums.timeWindowAll(Time.seconds(10),Time.seconds(5))
  14. val result: DataStream[Int] = window.sum(0)
  15. result.print()
  16. env.execute(this.getClass.getSimpleName)
  17. }
  18. }

时间间隔可以通过Time.milliseconds(x),Time.seconds(x),Time.minutes(x)等其中的一个来指定。

 

1.3.3 Window Reduce

WindowedStream 转换成 DataStream:给window赋一个reduce功能的函数,并返回⼀个聚合的结果。

  1. // 引⼊时间窗⼝
  2. val streamWindow = streamKeyBy.timeWindow(Time.seconds(5))
  3. val streamReduce = streamWindow.reduce(
  4. (a, b) => (a._1, a._2 + b._2)
  5. )

 

2 EventTime与Window

2.1 EventTime的引入

在Flink的流式处理中,绝⼤部分的业务都会使⽤eventTime,一般只在eventTime无法使用时,才会被迫使用ProcessingTime或者IngestionTime。
如果要使用EventTime,那么需要引入EventTime的时间属性,引入方式如下所示:

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

2.2 Watermark

2.2.1 基本概念

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

那么此时出现⼀个问题,⼀旦出现乱序,如果只根据eventTime决定window的运⾏,我们不能明确数据是否全部到位,但⼜不能⽆限期的等下去,此时必须要有个机制来保证⼀个特定的时间后,必须触发window去进⾏计算了,这个特别的机制,就是Watermark。Watermark是⼀种衡量Event Time进展的机制,它是数据本身的⼀个隐藏属性,数据本身携带着对应Watermark。
Watermark是⽤于处理乱序事件的,⽽正确的处理乱序事件,通常⽤Watermark机制结合window来实现。数据流中的Watermark⽤于表示timestamp⼩于Watermark的数据,都已经到达了,因此,window的执⾏也是由Watermark触发的。
Watermark可以理解成⼀个延迟触发机制,我们可以设置Watermark的延时时⻓t,每次系统会校验已经到达的数据中最⼤的maxEventTime,然后认定eventTime⼩于maxEventTime- t的所有数据都已经到达,如果有窗⼝的停⽌时间等于maxEventTime – t,那么这个窗⼝被触发执⾏。

有序流的Watermarker如下图所示:(Watermark设置为0)

乱序流的Watermarker如下图所示:(Watermark设置为2)

当Flink接收到每一条数据时,都会产⽣一条Watermark,这条Watermark就等于当前所有到达数据中的maxEventTime - 延迟时⻓长,也就是说,Watermark是由数据携带的,一旦数据携带的Watermark比当前未触发的窗口的停止时间要晚,那么就会触发相应窗口的执行。由于Watermark是由数据携带的,因此,如果运行过程中⽆法获取新的数据,那么没有被触发的窗口将永远都不不被触发。
上图中,我们设置的允许最大延迟到达时间为2s,所以时间戳为7s的事件对应的Watermark是5s,时间戳为12s的事件的Watermark是10s,如果我们的窗口1是1s~5s,窗口2是6s~10s,那么时间戳为7s的事件到达时的Watermarker恰好触发窗口1,时间戳为12s的事件到达时的Watermark恰好触发窗口2。

 

2.2.2 Watermark的引入

  1. val env = StreamExecutionEnvironment.getExecutionEnvironment
  2. // 从调⽤时刻开始给env创建的每⼀个stream追加时间特征
  3. env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  4. val stream = env.readTextFile("eventTest.txt").assignTimestampsAndWatermarks(
  5. new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(200)) {
  6. override def extractTimestamp(t: String): Long = {
  7. // EventTime是⽇志⽣成时间,我们从⽇志中解析EventTime
  8. t.split(" ")(0).toLong
  9. }
  10. })

2.3 EventTimeWindow API

当使⽤EventTimeWindow时,所有的Window在EventTime的时间轴上进⾏划分,也就是说,在Window启动后,会根据初始的EventTime时间每隔⼀段时间划分⼀个窗⼝,如果Window⼤⼩是3秒,那么1分钟内会把Window划分为如下的形式:

  1. [00:00:00,00:00:03)
  2. [00:00:03,00:00:06)
  3. ...
  4. [00:00:57,00:01:00)

如果Window⼤⼩是10秒,则Window会被分为如下的形式:

  1. [00:00:00,00:00:10)
  2. [00:00:10,00:00:20)
  3. ...
  4. [00:00:50,00:01:00)

注意,窗⼝是左闭右开的,形式为:[window_start_time,window_end_time)。
Window的设定⽆关数据本身,⽽是系统定义好了的,也就是说,Window会⼀直按照指定的时间间隔进⾏划分,不论这个Window中有没有数据,EventTime在这个Window期间的数据会进⼊这个Window。
Window会不断产⽣,属于这个Window范围的数据会被不断加⼊到Window中,所有未被触发的Window都会等待触发,只要Window还没触发,属于这个Window范围的数据就会⼀直被加⼊到Window中,直到Window被触发才会停⽌数据的追加,⽽当Window触发之后才接受到的属于被触发Window的数据会被丢弃。

Window会在以下的条件满⾜时被触发执⾏:
l watermark时间 >= window_end_time;
l 在[window_start_time,window_end_time)中有数据存在。


我们通过下图来说明Watermark、EventTime和Window的关系。

2.3.1 滚动窗口(TumblingEventTimeWindows)

  1. val env = StreamExecutionEnvironment.getExecutionEnvironment
  2. env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  3. // 创建SocketSource
  4. val stream = env.socketTextStream("localhost", 8888)
  5. // 对stream进⾏处理并按key聚合
  6. val streamKeyBy = stream.assignTimestampsAndWatermarks(
  7. new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(3000)) {
  8. override def extractTimestamp(element: String): Long = {
  9. val sysTime = element.split(" ")(0).toLong
  10. println(sysTime)
  11. sysTime
  12. }}).map(item => (item.split(" ")(1), 1)).keyBy(0)
  13. // 引⼊滚动窗⼝
  14. val streamWindow = streamKeyBy.window(TumblingEventTimeWindows.of(Time.seconds(10)))
  15. // 执⾏聚合操作
  16. val streamReduce = streamWindow.reduce(
  17. (a, b) => (a._1, a._2 + b._2)
  18. )
  19. // 将聚合数据写⼊⽂件
  20. streamReduce.print
  21. // 执⾏程序
  22. env.execute("TumblingWindow")

结果是按照Event Time的时间窗⼝计算得出的,⽽⽆关系统的时间(包括输⼊的快慢)。 

2.3.2  滑动窗⼝(SlidingEventTimeWindows)

  1. // 获取执⾏环境
  2. val env = StreamExecutionEnvironment.getExecutionEnvironment
  3. env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  4. // 创建SocketSource
  5. val stream = env.socketTextStream("localhost", 11111)
  6. // 对stream进⾏处理并按key聚合
  7. val streamKeyBy = stream.assignTimestampsAndWatermarks(
  8. new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(0)) {
  9. override def extractTimestamp(element: String): Long = {
  10. val sysTime = element.split(" ")(0).toLong
  11. println(sysTime)
  12. sysTime
  13. }}).map(item => (item.split(" ")(1), 1)).keyBy(0)
  14. // 引⼊滚动窗⼝
  15. val streamWindow = streamKeyBy.window(SlidingEventTimeWindows.of(Time.seconds(10),
  16. Time.seconds(5)))
  17. // 执⾏聚合操作
  18. val streamReduce = streamWindow.reduce(
  19. (a,b) => (a._1, a._2 + b._2)
  20. )
  21. // 将聚合数据写⼊⽂件
  22. streamReduce.print
  23. // 执⾏程序
  24. env.execute("TumblingWindow")

 

2.3.3 会话窗⼝(EventTimeSessionWindows)

相邻两次数据的EventTime的时间差超过指定的时间间隔就会触发执⾏。如果加⼊Watermark,那么当触发执⾏时,所有满⾜时间间隔⽽还没有触发的Window会同时触发执⾏。

  1. // 获取执⾏环境
  2. val env = StreamExecutionEnvironment.getExecutionEnvironment
  3. env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  4. // 创建SocketSource
  5. val stream = env.socketTextStream("localhost", 11111)
  6. // 对stream进⾏处理并按key聚合
  7. val streamKeyBy = stream.assignTimestampsAndWatermarks(
  8. new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(0)) {
  9. override def extractTimestamp(element: String): Long = {
  10. val sysTime = element.split(" ")(0).toLong
  11. println(sysTime)
  12. sysTime
  13. }}).map(item => (item.split(" ")(1), 1)).keyBy(0)
  14. // 引⼊滚动窗⼝
  15. val streamWindow =
  16. streamKeyBy.window(EventTimeSessionWindows.withGap(Time.seconds(5)))
  17. // 执⾏聚合操作
  18. val streamReduce = streamWindow.reduce(
  19. (a, b) => (a._1, a._2 + b._2)
  20. )
  21. // 将聚合数据写⼊⽂件
  22. streamReduce.print
  23. // 执⾏程序

 

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/从前慢现在也慢/article/detail/751914
推荐阅读
相关标签
  

闽ICP备14008679号