第七章 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 顺序排列的。

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

　　当 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 的关系。

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 技术框架。

Flink EventTime 与 Window