Flink中watermark为什么选择最小一条(源码分析)

昨天在社区群看到有人问，为什么水印取最小的一条？这里分享一下自己的理解

首先水印一般是设置为：(事件时间 - 指定的值)  这里的作用是解决迟到数据的问题，从源码来看一下它如何解决的

先来看下windowoperator.java接收到数据以后做了什么

在processelement方法中

因为这里是事件时间窗口所以会默认注册一个事件时间trigger,这是默认trigger的onelement方法

当返回continue时，也就是说水印还没有达到,这条数据属于的窗口的右边界,也就是说窗口还没有到触发的时机

可以看到这里他把这个数据属于的窗口的右边界注册成为一个触发器(timer)

这个timer有什么用呢，来看一下窗口触发的逻辑

所有的上游数据会从这里接收，在streaminputprocessor.java的processinput()方法中有这样一段逻辑，当接收到水印

里面又调用了

从名字就可以知道是取了一个最小的水印，具体更新最小水印时间逻辑如下

这里就是我们的问题了，为什么他选取了最小的一个水印？

看看这段代码的后面他又做了什么

这个方法里面有很多的调用，这里就不列出来了，可以自己点进去看一下，最后会走到这里

这里会判断定时器时间是否小于最小水印时间（是触发定时器的条件）

在oneventtime()方法中

这里看到当返回fire时，会调用emitwindowcontents()这个方法里面就会调用我们真正用户的process()方法了，而那个windowstate.get()则是拿到了一个窗口中的所有数据

而，是否触发窗口就看oneventtime()方法是否返回fire，具体实现如下

判断定时器的时间，变量time(前面我们将数据属于的窗口的右边界作为定时器的时间)是否等于窗口右边界的时间，来决定窗口是否触发

！！！那既然最小水印是触发定时器的条件，定时器到时会触发窗口，那我们为什么会选择最小的水印来作为触发条件呢？看下面这张图

可以看到一个窗口可能会有接收到许多的上游，每一个上游的流都会带有事件时间，那我们哪知道选用哪个流的水印时间作为窗口触发的条件呢？

有个最简单的办法就是：如果我上游每个流中取最小的水印，那就证明其他的水印时间肯定是大于最小的这个，我最小的一条流都达到了窗口的触发时间，那其他来自上游的流肯定都已经超过这个触发时间了，那我就可以触发这个窗口了