彻底理解flink水位和窗口关系

小鱼儿 2022-11-01 10:54 238阅读 0赞

[https://cloud.tencent.com/developer/article/1539213][https_cloud.tencent.com_developer_article_1539213]

翻阅了很多的 水位和窗口 的博客文章，由于本来概念就很晦涩，光看很难一下子理解，所以本文用实际数据，做了比较详尽的梳理。可以跟着步骤一步步操作，亲自体会水位在处理乱序数据的时候发挥的重要作用，文末有代码地址，下载下来在idea中跑起来运行。

### **一、数据乱序的现象** ###

实时计算中，对数据时间比较敏感，有 EventTime 和 ProcessTime 之分，一般来说 EventTime 是从原始消息中提取出来的，ProcessTime 是 Flink 自己提供的。

在实际应用中，数据源往往很多个且时钟无法严格同步，数据汇集过程中传输的距离和速度也不尽相同，在上游多个节点处理过程的处理速度也有差异，这些因素使得 Event Time 的乱序基本是一个必然现象。

所以 Flink 提供了窗口和水位线的功能，使其在一定时间范围内可以正确处理数据乱序的现象。

数据正序

数据乱序

### **二、Window 的概念** ###

在 Flink 中，window 可以分为 基于时间（Time-based）的 window 以及基于数量（Count-based）的 window，另外还有基于 session 的 window，同时由于某些特殊需要，还可以自定义 window。

1、Tumbling window （翻滚窗口）

比如每多长时间统计一次（基于时间）

比如每多少数量统计一次（基于数量）

2、Sliding window （滑动窗口）

比如每隔30秒统计过去1分钟的数据量（基于时间）

比如每隔10个元素统计过去100个元素的数据量（基于数量）

（window 就简单的介绍一下概念，本次重点是讲 watermark 和 window 处理乱序数据）

3、Flink 的窗口划分，不是基于数据的时间来划分的，而是基于自然时间来划分的。比如我们设置窗口大小为3s，事件时间为 2019-11-12 15:00:05

那窗口的时间范围并不是想象中的：

\[2019-11-12 15:00:05,2019-11-12 15:00:08\]

而是 一个前闭后开的区间：

\[2019-11-12 15:00:00 , 2019-11-12 15:00:03 )

\[2019-11-12 15:00:03 , 2019-11-12 15:00:06 )

\[2019-11-12 15:00:06 , 2019-11-12 15:00:09 )

### **三、WaterMark 的概念** ###

watermark 是描述 EventTime 进展的机制。通常基于 EventTime 的数据，时间属性都在事件的内容里面。

比如这条数据：“01,2019-11-12 11:27:12,你好”

EventTime 是 ：

2019-11-12 11:27:12 （1573529232） -- 括号里是时间戳

这条数据的水印的时间戳可能是：

2019-11-12 11:27:02 （1573529222），少了10s

这个水印的时间戳意义就是：

timestamp 小于

2019-11-12 11:27:02 （1573529222）

的数据，都已经到达了。

那这种比 EventTime 时间少一点的时间戳有什么意义呢？

我们知道，流处理从事件产生，到流经 source，再到 operator ，中间是有一个过程和时间的。由于一些网络、错误等原因，会有一些数据是乱序的（out-of-date 或者 late element）。

但是对于 late element，我们又不能无限期的等待下去，必须有一个机制来保证等待一段时间后，必须触发 window 计算了。这个特别的机制，就是 warkmark。

Flink 中有一段源码，展示了什么情况下需要触发计算 window 数据了：

@Override
    public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
      if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
        // if the watermark is already past the window fire immediately
        return TriggerResult.FIRE;
      } else {
        ctx.registerEventTimeTimer(window.maxTimestamp());
        return TriggerResult.CONTINUE;
      }
    }

当水印时间 大于等于 窗口的结束时间，开始触发窗口的计算。

注意：水印不是和某一条数据绑定的，而是一个全局的概念。也就是水印时间是一直往上的，不会下降

**四、代码逻辑展示**

1、主要逻辑代码展示（完整代码见文末的 gitee 地址）

设置了一个 3s 的窗口，并生成水印，并处理每一个窗口的数据，并打印出来

// 运行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置为事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    
    // kafka 相关参数
    val prop = new Properties
    prop.setProperty("bootstrap.servers", "192.168.17.26:9092,192.168.17.27:9092,192.168.17.28:9092")
    prop.setProperty("zookeeper.connect", "192.168.17.24,192.168.17.25,192.168.17.26")
    prop.setProperty("group.id", "default_group_id")
    val consumer = new FlinkKafkaConsumer010[String]("flink-test", new SimpleStringSchema, prop)
    // 从上一次读取的位置开始读
    consumer.setStartFromGroupOffsets()
    
    // 输入数据的格式：0001,2019-11-12 11:25:00
    env.addSource(consumer)
        // 过滤空数据
        .filter(!_.isEmpty)  
        .map(f => {
          val arr = f.split(",")
          val code = arr(0)
          val time = parseDateNewFormat(arr(1)).getTime
          //        val time = arr(1).toLong
          (code, time)
        })
        // 指定水印生成的逻辑
        .assignTimestampsAndWatermarks(new TimeStampExtractor)
        // 按照 code 来逻辑划分窗口，并行计算
        .keyBy(_._1)
        // 指定 翻滚窗口，3s生成一个窗口
        .window(TumblingEventTimeWindows.of(Time.seconds(3)))
        // 允许延迟5s之后才销毁计算过的窗口
        //.allowedLateness(Time.seconds(5))
        // 处理窗口数据
        .process(new MyProcessWindowFunction)
        // 打印处理完的数据
        .print()

2、生成水位的代码

class TimeStampExtractor extends AssignerWithPeriodicWatermarks[(String, Long)] with Serializable {
        var currentMaxTimestamp = 0L
        val maxOutOfOrderness = 10000L // 最大允许的乱序时间是10s
        var a: Watermark = null
    
        val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")
    
        // 水印 是 当前时间减去10s
        override def getCurrentWatermark: Watermark = {
          //        println("getCurrentWatermark")
          a = new Watermark(currentMaxTimestamp - maxOutOfOrderness)
          a
        }
    
        override def extractTimestamp(element: (String, Long), previousElementTimestamp: Long): Long = {
          //        println("extractTimestamp")
          val timestamp = element._2
          currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp)
          println("timestamp:" + element._1 + "," + element._2 + "," + format.format(element._2) + " ||| " + currentMaxTimestamp + "," + format.format(currentMaxTimestamp) + " ||| " + DateUtils.getNewFormatDateString(new Date(a.getTimestamp)))
          timestamp
        }
      }

核心的两行代码是：

currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp)
    new Watermark(currentMaxTimestamp - maxOutOfOrderness)

拿当前的时间和上一次的时间的最大时间，减去10s作为每次处理到来时的水印。

就是全局水印时间，是事件时间减去10s。

3、处理窗口数据的逻辑

class MyProcessWindowFunction extends ProcessWindowFunction[(String, Long), String, String, TimeWindow] {
      override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[String]): Unit = {
    
        val arr = ArrayBuffer[(String, Long)]()
        val iterator = elements.iterator
        while (iterator.hasNext) {
          val value = iterator.next()
          println(value._1, DateUtils.getNewFormatDateString(new Date(value._2)))
          arr += value
        }
        println(arr)
        val format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")
        val timeWindow = context.window
        out.collect(key + "," + arr.size + "," + format.format(arr.head._2) + "," + format.format(arr.last._2) + "," + format.format(timeWindow.getStart) + "," + format.format(timeWindow.getEnd))
      }
    
    }

就是把窗口的数据都收集了起来，并打印出来

**五、实战**

1、下面我们来输入第一条数据：01,2019-11-12 11:00:12

如果这是一个时间轴，上面的箭头是产生的数据位置，下面的箭头是生成的水印位置

控制台打印的数据（事件时间，最大时间，水印时间）

可以看到水印是1970年，转换成时间戳等于 -10000，为 0 - 10000 的结果，也就是控制台打印的是上一条数据的时间戳

timestamp:01,1573527612000,2019-11-12 11:00:12.000 ||| 1573527612000,2019-11-12 11:00:12.000 ||| 1970-01-01 07:59:50

没有触发窗口计算，触发窗口计算是会有打印的

2、输入第二条数据：01,2019-11-12 11:00:14

控制台输出：

timestamp:01,1573527614000,2019-11-12 11:00:14.000 ||| 1573527614000,2019-11-12 11:00:14.000 ||| 2019-11-12 11:00:02

没有触发窗口计算

3、输入第2条数据：01,2019-11-12 11:00:15

控制台输出：

timestamp:01,1573527615000,2019-11-12 11:00:15.000 ||| 1573527615000,2019-11-12 11:00:15.000 ||| 2019-11-12 11:00:04

没有触发窗口计算

4、输入第4条数据：01,2019-11-12 11:00:22

控制台输出：

timestamp:01,1573527622000,2019-11-12 11:00:22.000 ||| 1573527622000,2019-11-12 11:00:22.000 ||| 2019-11-12 11:00:05

此时距离第一条数据已经过去了10s了，仍然没有触发计算

此时水印位置在第一条数据所在窗口的开始时间，上面我们说水印大于等于窗口结束时间的时候，才会触发窗口计算，那我们继续输入数据

5、输入第5条数据：01,2019-11-12 11:00:23

控制台输出：

timestamp:01,1573527623000,2019-11-12 11:00:23.000 ||| 1573527623000,2019-11-12 11:00:23.000 ||| 2019-11-12 11:00:12

没有触发计算

6、输入第6条数据：01,2019-11-12 11:00:24

控制台输出：

timestamp:01,1573527624000,2019-11-12 11:00:24.000 ||| 1573527624000,2019-11-12 11:00:24.000 ||| 2019-11-12 11:00:13

没有触发计算，预感输入下一条数据后，水印位置到达了窗口的结束，此时会触发计算了。继续输入

7、输入第7条数据：01,2019-11-12 11:00:25

控制台输出：

timestamp:01,1573527625000,2019-11-12 11:00:25.000 ||| 1573527625000,2019-11-12 11:00:25.000 ||| 2019-11-12 11:00:14
    
    (01,2019-11-12 11:00:12)
    
    (01,2019-11-12 11:00:14)
    
    ArrayBuffer((01,1573527612000), (01,1573527614000))
    
    1> 01,2,2019-11-12 11:00:12.000,2019-11-12 11:00:14.000,2019-11-12 11:00:12.000,2019-11-12 11:00:15.000

此时已经触发了窗口计算，打印出了我们自定义的输出内容。

到此，我们可以总结出第一个知识点：

**watermark 到达 窗口的结束时间点的时候，会触发计算**

我们再看看具体的输出，只输出了两个元素，上图的 1 和 2 ，3 并没有输出

可以总结出第二个知识点：

**时间窗口是前闭后开的，不包括结束时间**

8、我们尝试输入一个乱序的数据：01,2019-11-12 11:00:24

注意这时候，这行代码是注释掉的：

控制台输出：

timestamp:01,1573527624000,2019-11-12 11:00:24.000 ||| 1573527625000,2019-11-12 11:00:25.000 ||| 2019-11-12 11:00:15

并没有触发计算。

可以总结出第三个知识点：**窗口计算完毕后，就直接销毁了**。

9、尝试输入一个很久之前的数据，比如输入一个第一个窗口之间的数据，可以预期，应该是没有输出的

输入：01,2019-11-12 11:00:12

确实没有计算。也就是验证了，窗口计算完毕之后，是即时销毁的。

10、然后我们把刚刚那行注释打开，重新执行，并重新输入以上数据到第七步

然后输入：01,2019-11-12 11:00:12

控制台输出：

发现刚刚输入的被重新计算了。

延迟5s之后才销毁窗口的意思是：水位位置 - window\_end\_time < 5 的时候，窗口是保留的，此时落在任何水位之前的窗口的数据都是被计算的;

当 水位位置 - window\_end\_time = 5 的时候，窗口被销毁

当前水位线的时间戳在 15。

我们输入一个：01,2019-11-12 11:00:29，使水位线到达19的位置，图中8的位置

当水位线到19的时候，输入01,2019-11-12 11:00:12的时候，窗口都会被计算。

输入一个：01,2019-11-12 11:00:30，使水位线到达20的位置

当水位线到20的时候，输入 01,2019-11-12 11:00:12

控制台输出：

timestamp:01,1573527612000,2019-11-12 11:00:12.000 ||| 1573527630000,2019-11-12 11:00:30.000 ||| 2019-11-12 11:00:20

窗口已经被销毁，没有执行窗口计算。

由此，我们得出第三个结论：

**如果设置了 allowedLateness，水位位置 - window\_end\_time < 5 的时候，落在窗口的数据都会被计算。**

**当 水位位置 - window\_end\_time = 5 的时候，窗口被销毁，落在窗口的数据不再被计算。**

### **六、总结** ###

1、窗口的时间范围是自然时间决定的，和数据无关

2、水印不是和某一条数据绑定的，而是一个全局的概念。我们设置水位的时候，需要把水位设置成持续上升，不然没有作用

3、水位到达窗口结束时间的时候，会触发这个窗口的计算

4、窗口计算完毕后，会立即销毁

5、如果设置了 allowedLateness，水位位置 - window\_end\_time < 5 的时候，落在窗口的数据都会被计算。

当 水位位置 - window\_end\_time = 5 的时候，窗口被销毁，落在窗口的数据不再被计算。

### **七、代码** ###

本次 水位和窗口的关系 的实战代码，请在公众号后台回复 20191112 领取

本文分享自微信公众号 - kk大数据（kkbigdata），作者：kk大数据

原文出处及转载信息见文内详细说明，如有侵权，请联系 yunjia\_community@tencent.com 删除。

原始发表时间：2019-11-12

本文参与[腾讯云自媒体分享计划][Link 1]，欢迎正在阅读的你也加入，一起分享。

[https_cloud.tencent.com_developer_article_1539213]: https://cloud.tencent.com/developer/article/1539213
[Link 1]: https://cloud.tencent.com/developer/support-plan