Flink EventTime 与 Window

た入场券 2024-04-20 09:12 89阅读 0赞

# 第七章 EventTime 与 Window #

## 7.1 EventTime 的引入 ##

在 Flink 的 流 式 处 理中 ， 绝 大 部 分 的 业务都 会 使 用 eventTime，一般只在

eventTime 无法使用时，才会被迫使用 ProcessingTime 或者 IngestionTime。

如果要使用 EventTime，那么需要引入 EventTime 的时间属性，

引入方式如下所示：

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

## 7.2 Watermark ##

### 7.2.1 基本概念 ###

我们知道，流处理从事件产生，到流经 source，再到 operator，中间是有一个过

程和时间的，虽然大部分情况下，流到 operator 的数据都是按照事件产生的时间顺

序来的，但是也不排除由于网络等原因，导致乱序的产生，所谓乱序，就是指 Flink

接收到的事件的先后顺序不是严格按照事件的 Event Time 顺序排列的。

![1604514-20190713172907263-1280894677.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）

![1604514-20190713172956964-273970204.png][]

![1604514-20190713173040897-1636127809.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。

### 7.2.2 Watermark 的引入 ###

val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 从调用时刻开始给 env 创建的每一个 stream 追加时间特征
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
    val stream = env.socketTextStream("localhost", 11111).assignTimestampsAndWatermarks(
    　　new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(200)) {
    　　override def extractTimestamp(t: String): Long = {
    　　　　// EventTime 是日志生成时间，我们从日志中解析 EventTime
    　　　　t.split(" ")(0).toLong
    　　}
    })

## 7.3 EvnetTimeWindow 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 的关系。

![1604514-20190713173423359-1147863605.png][]

### 7.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 的时间窗口计算得出的，而无关系统的时间（包括输入的快慢）。

### 7.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")

### 7.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")

测试代码：

package eventtimewindow
    
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, SlidingEventTimeWindows, TumblingEventTimeWindows}
    import org.apache.flink.streaming.api.windowing.time.Time
    
    object EventTimeWindow01 {
    
      def main(args: Array[String]): Unit = {
    
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        //修改时间特性为 EventTime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
        val stream = env.socketTextStream("localhost", 11111).assignTimestampsAndWatermarks(
          new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(3000)) {
            override def extractTimestamp(element: String): Long = {
              // eventTime word
              val eventTime = element.split(" ")(0).toLong
              println(eventTime)
              eventTime
            }
          }
        ).map(item => (item.split(" ")(1), 1L)).keyBy(0)
    
        /* TumblingEventTimeWindows */
    //    val streamWindow = stream.window(TumblingEventTimeWindows.of(Time.seconds(5)))
    //
    //    val streamReduce = streamWindow.reduce(
    //      (item1, item2) => (item1._1, item2._2 + item1._2)
    //    )
    //
    //    streamReduce.print()
    
        /* SlidingEventTimeWindows */
    //    val streamWindow = stream.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
    //
    //    val streamReduce = streamWindow.reduce(
    //      (item1, item2) => (item1._1, item2._2 + item1._2)
    //    )
    //
    //    streamReduce.print()
    
        /* EventTimeSessionWindows */
        val streamWindow = stream.window(EventTimeSessionWindows.withGap(Time.seconds(5)))
    
        val streamReduce = streamWindow.reduce(
          (item1, item2) => (item1._1, item2._2 + item1._2)
        )
    
        streamReduce.print()
    
    
        env.execute("EventTimeJob")
      }
    
    }

# 总结 #

Flink 是一个真正意义上的流计算引擎，在满足低延迟和低容错开销的基础之上，完美

的解决了 exactly-once 的目标，真是由于 Flink 具有诸多优点，越来越多的企业开始使用 Flink

作为流处理框架，逐步替换掉了原本的 Storm 和 Spark 技术框架。

转载于:https://www.cnblogs.com/LXL616/p/11181488.html

[1604514-20190713172907263-1280894677.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/c6680f97bcca4f22b27c2659c7956a66.png
[1604514-20190713172956964-273970204.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/7bf40fd22c6c4279bc45127a3642b59b.png
[1604514-20190713173040897-1636127809.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/55397e381e364a8c984729dbf6e6ca4c.png
[1604514-20190713173423359-1147863605.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/3ab6bab839ae4d4ab59e81104f5f6fdb.png