【实战】彻底理解flink水位和窗口关系
翻阅了很多的 水位和窗口 的博客文章，由于本来概念就很晦涩，光看很难一下子理解，所以本文用实际数据，做了比较详尽的梳理。可以跟着步骤一步步操作，亲自体会水位在处理乱序数据的时候发挥的重要作用，文末有代码地址，下载下来在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 领取