欢迎您的访问
专注架构,Java,数据结构算法,Python技术分享

11:Apache Flink EventTime与Window

1. EventTime的引入

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

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

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

2. Watermark

2.1 基本概念

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

100_1.png

那么此时出现一个问题,一旦出现乱序,如果只根据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)

100_2.png

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

100_3.png

当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 Watermark的引入

val env = StreamExecutionEnvironment.getExecutionEnvironment

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

val stream = env.readTextFile("eventTest.txt").assignTimestampsAndWatermarks(
  new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(200)) {
  override def extractTimestamp(t: String): Long = {
     // EventTime是日志生成时间,我们从日志中解析EventTime
     t.split(" ")(0).toLong
  }
})

3. EventTimeWindow API

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

[00:00:00,00:00:03)
[00:00:03,00:00:06)
...
[00:00:57,00:01:00)

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

[00:00:00,00:00:10)
[00:00:10,00:00:20)
...
[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会在以下的条件满足时被触发执行:

  • watermark时间 >= window_end_time;
  • 在[window_start_time,window_end_time)中有数据存在。

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

100_4.png

3.1 滚动窗口(TumblingEventTimeWindows)

// 获取执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

// 创建SocketSource
val stream = env.socketTextStream("localhost", 11111)

// 对stream进行处理并按key聚合
val streamKeyBy = stream.assignTimestampsAndWatermarks(
  new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(3000)) {
     override def extractTimestamp(element: String): Long = {
       val sysTime = element.split(" ")(0).toLong
       println(sysTime)
       sysTime
     }}).map(item => (item.split(" ")(1), 1)).keyBy(0)

// 引入滚动窗口
val streamWindow = streamKeyBy.window(TumblingEventTimeWindows.of(Time.seconds(10)))

// 执行聚合操作
val streamReduce = streamWindow.reduce(
  (item1, item2) => (item1._1, item1._2 + item2._2)
)

// 将聚合数据写入文件
streamReduce.print

// 执行程序
env.execute("TumblingWindow")

结果是按照Event Time的时间窗口计算得出的,而无关系统的时间(包括输入的快慢)。

3.2 滑动窗口(SlidingEventTimeWindows)

// 获取执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

// 创建SocketSource
val stream = env.socketTextStream("localhost", 11111)

// 对stream进行处理并按key聚合
val streamKeyBy = stream.assignTimestampsAndWatermarks(
  new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(0)) {
     override def extractTimestamp(element: String): Long = {
       val sysTime = element.split(" ")(0).toLong
       println(sysTime)
       sysTime
     }}).map(item => (item.split(" ")(1), 1)).keyBy(0)

// 引入滚动窗口
val streamWindow = streamKeyBy.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))

// 执行聚合操作
val streamReduce = streamWindow.reduce(
  (item1, item2) => (item1._1, item1._2 + item2._2)
)

// 将聚合数据写入文件
streamReduce.print

// 执行程序
env.execute("TumblingWindow")

3.3 会话窗口(EventTimeSessionWindows)

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

// 获取执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

// 创建SocketSource
val stream = env.socketTextStream("localhost", 11111)

// 对stream进行处理并按key聚合
val streamKeyBy = stream.assignTimestampsAndWatermarks(
 new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(0)) {
    override def extractTimestamp(element: String): Long = {
      val sysTime = element.split(" ")(0).toLong
      println(sysTime)
      sysTime
    }}).map(item => (item.split(" ")(1), 1)).keyBy(0)

// 引入滚动窗口
val streamWindow = streamKeyBy.window(EventTimeSessionWindows.withGap(Time.seconds(5)))

// 执行聚合操作
val streamReduce = streamWindow.reduce(
 (item1, item2) => (item1._1, item1._2 + item2._2)
)

// 将聚合数据写入文件
streamReduce.print

// 执行程序
env.execute("TumblingWindow")
赞(0) 打赏
版权归原创作者所有,任何形式转载请联系作者;码农code之路 博客站点 » 11:Apache Flink EventTime与Window

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏