1. Watermark概念
watermark是一种衡量Event Time进展的机制,它是数据本身的一个隐藏属性。通常基于Event Time的数据,自身都包含一个timestamp.watermark是用于处理乱序事件的,而正确的处理乱序事件,通常用watermark机制结合window来实现。我们知道,流处理从事件产生,到流经source,再到operator,中间是有一个过程和时间的。虽然大部分情况下,流到operator的数据都是按照事件产生的时间顺序来的,但是也不排除由于网络、背压等原因,导致乱序的产生(out-of-order或者说late element)。但是对于late element,我们又不能无限期的等下去,必须要有个机制来保证一个特定的时间后,必须触发window去进行计算了。此时就是watermark发挥作用了,它表示当达到watermark到达之后,在watermark之前的数据已经全部达到(即使后面还有延迟的数据).watermark的示意图如下.
2. 生成EventTime和Watermark
以下这个程序的功能是实现计算相同时间窗口出现相同单词的统计.在这个过程中,自定义实现了时间戳和Watermark.
- maxOutOfOrderness 这个参数在设置的时候往往根据经验来.MaxOutOfOrderness设置的太小,而自身数据发送时由于网络等原因导致乱序或者late太多,那么最终的结果就是会有很多单条的数据在window中被触发,数据的正确性影响太大。如果设置太大,导致设置的Watermark太小,使得Watermark没有用,因为原本在很短的时间内,一个窗口的所有的数据都到达了,但是不得不等Watermark一点点变大, 才能触发计算.
3. EventTime按顺序的情况
运行上述程序.初始化的时候,由于没有输入,watermark为-10000
在9900监听端口,输入aa 1522827199000(2018-04-04 15:33:19),重复3次.得到如下图所示结果.现在的watermark是1522827189000(2018-04-04 15:33:09),即为最大的currentMaxTimestamp-10000.在这里生命下,aa 的时间2018-04-04 15:33:19在2018-04-04 15:33:0-2018-04-04 15:33:20这个窗口中.因为我设置的窗口为20s.不管怎样怎样,窗口是确定的.初始化设置后,就一直不会变.而且窗口是左闭右开的区间.
接着输入bb 1522827299000(2018-04-04 15:34:59),此时的watermark为1522827289000(2018-04-04 15:34:49),此时的watermark超过了aa所在窗口的endtime(2018-04-04 15:33:20).那么会触发计算,从而会有下面的输出.在这里强调下,触发计算的时间点是
- watermark超过了window的endtime.
- 在该window中有数据.
只有同时满足这两个条件,就会触发计算.
4. EventTime不按顺序的情况
输入如下数据:
aa 1522827261000(2018-04-04 15:34:21)
aa 1522827251000(2018-04-04 15:34:11)
aa 1522827252000(2018-04-04 15:34:12)
ee 1522827291000(2018-04-04 15:34:51)
由于最后一个ee的输入,改变了watermark,使得当前的时间戳为2018-04-04 15:34:41.那么此时会触发前面两个窗口的计算.计算结果如下.
Note:此时,如果还输入之前的时间窗口的aa 1522827261000.是不会触发计算的,它会丢弃掉数据,这在后面会引入allowedLateness参数.原窗口中的内容不会立即被删除,而是会再次等待一段时间,即watermark小于end-time + allowedLateness时,后续的该窗口的数据到达时会纳入到原窗口,再次触发计算.而watermark >= end_time + allowedLateness,后续的还有属于该窗口的数据到达时,那么这种数据只能被删除了,因为系统不会无限制的等下去,这既会增加window buffer的大小,也会引起不必要的性能下降.
5. 总结
通过以上例子,我们可以看到如何通过watermark对顺序以及乱序数据的处理.以及如何在watermark触发之后还能通过 allowedLateness对延迟做一些补偿.watermark的设计思想和用途很光,我也只是浅尝辄止.