Flink中EventTimeTrigger的理解

作者：耳边的火
链接：https://www.jianshu.com/p/c8c789ff5570
来源：简书

在Flink中，使用event-time模式时，默认提供的window有TumblingEventTimeWindows，SlidingEventTimeWindows，EventTimeSessionWindow等，其中这些是属于window operator中的一部分，称作 window assigner。window operator包含四个组件，除了 window assigner外，还包括 trigger , evictor, window process。其作用分别如下：

• window assigner 指明数据流中的数据属于哪个window

• trigger 指明在哪些条件下触发window计算，基于处理数据时的时间以及事件的特定属性、

• evictor 可选组件，在window执行计算前或后，将window中的数据移除，如使用globalWindow时，由于该window的默认trigger为永不触发，所以既需要实现自定义trigger，也需要实现evictor，移除部分已经计算完毕的数据。

• window process flink默认提供的有 ReduceFunction,AggragateFunction.还可以自定义实现 windowProcessFunction

而上述TumblingEventTimeWindows，SlidingEventTimeWindows，EventTimeSessionWindow三个默认提供的window operator中，都提供了默认的trigger，因此我们在初次接触flink，使用以上三个方法时，没有写trigger，直接写window process，如 .reduce()。这是因为这三个window中的getDefaultTrigger()方法使用的是EventTimeTrigger,也就是它给我们提供了默认的trigger。

@Override
    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
        return EventTimeTrigger.create();
    }

而我们知道，在event-time模式中，window的计算是在watermark到达后再执行计算，watermark在计算延迟与结果的完整性之间提供了一个权衡，flink也提供了其他机制来处理迟到数据。所以，我们来看看EventTimeTrigger中是如何实现这样的逻辑的：

@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;
        }
    }
 
 
    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) {
        return time == window.maxTimestamp() ?
            TriggerResult.FIRE :
            TriggerResult.CONTINUE;
    }
 
 

此处，仅贴出两个相关的方法，onElement以及onEventTime方法。其中onElement方法在每次数据进入该window时会触发，而onEventTime是在之前注册的eventTime回调函数到达时间时，进行触发。需要注意的是，根据Trigger的接口，onElement方法中的timestamp参数，指的是处理时的机器时间，如下：

 /**
     * Called for every element that gets added to a pane. The result of this will determine
     * whether the pane is evaluated to emit results.
     *
     * @param element The element that arrived.
     * @param timestamp The timestamp of the element that arrived.
     * @param window The window to which the element is being added.
     * @param ctx A context object that can be used to register timer callbacks.
     */
    public abstract TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception;
 
 

修改内容：此处按照注释的理解好像是说的是机器时间，但是跟进代码，在 WindowOperator.Context.onElement 方法中，写明了：

 
 
public TriggerResult onElement(StreamRecord<IN> element) throws Exception {
            return trigger.onElement(element.getValue(), element.getTimestamp(), window, this);
        }
而 onEventTime方法中的 timestamp 参数，则是指注册回调函数时写入的 timestamp 数值。下面详细看一下，EventTimeTrigger中的onElement方法：@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;
        }
    }

我们假设watermark使用默认值，200ms产生一个，为了简便，我们假设第二个窗口在第一个窗口结束后，立即有第一个数据流入，按照时间先后顺序依次套入上面的代码，看一下执行结果：第一种情况：watermark不处理迟到数据，其时间戳与数据的时间戳相同。

• window.maxTimestamp()指的是当前窗口的最大时间戳的数值，而ctx.getCurrentWatermark()此时指的是window上下文所接收到的，最新到达的watermark，该watermark触发了上一个窗口的计算。此时新窗口尽管还未填满，仅有一个数据，但是其最大时间边界已经确定，一般来说要比此时的watermark的值要大。

• 因此不入if语句，而进入else语句。

• else语句会注册一个回调函数，当未来某个watermark的时间戳大于该trigger的注册时间时，就会触发trigger，执行该trigger所在的window operator的window process进行计算。

• 每进入一个新数据，就会注册一遍这个回调函数。但是由于trigger会添加到set集合中，因此重复添加相同的trigger不会真的注册重复的trigger。详见：WindowOperator.registerEventTimeTimer() -> HeapInternalTimerService.registerEventTimeTimer -> InternalTimer.equals()

• 200ms后，新的watermark跟随新的数据到达，新的数据进入该window，再次判断，仍然进入else语句，新的watermark的时间戳没有超过window.maxTimestamp，不会触发trigger，执行计算。

• 如此持续，直到一个时间戳大于等于window.maxTimestamp的事件到达并跟随着watermark。假设到达的数据时间戳恰好为window.maxTimestamp，也就是该window理论上所能容纳的最大的时间戳。此时watermark有两种情况，一为紧跟着数据到达，二为最多等待200ms后才到达。不论哪种情况，都不会在此时进入onElement方法了，watermark后没有新的数据进入window，就不会走该方法。而是会触发时间小于该watermark的所有trigger，就会触发该window执行计算。

那么onElement方法中的if条件什么时候进入呢？难道是处理迟到数据用的？，我们看一下第二种情况，即：watermark处理5秒内的延迟数据

• 第一个数据进入新的window，但此时上一个window还未计算，因为watermark处理5秒内延迟，此时window的上下文还未收到大于上一个window的maxTimestamp的watermark。因此上一个window还未执行，在等待迟到数据。

• 假设在新的window处理了4.9秒后，来了个新数据，该数据从时间戳上来看，应该属于上一个window，因此window assigner将其放入上一个window中，执行onElement方法，但是此时才过了4.9秒而已，还未到第5秒，也就是当前window的上下文保存的watermark的时间戳，还未大于上一个window的maxTimestamp，（但是要注意，windwo上下文中保存的watermark是200ms增长一次的，因为watermark就是200ms产生一次，但是由于处理5秒内的迟到数据，因此即便4.9秒内，watermark不断增长，但是仍未能触发上一个window的trigger），在onElement方法中，还是不会进入if语句，仍然是else语句，待5秒过后，新的watermark到达，其时间戳大于等于上一个window的maxTimestamp，才会触发trigger，执行上一个window的计算。

因此，在watermark处理迟到数据情境下，我们看到确实是处理了迟到数据，但是并没有走if条件。

那if条件存在的意义是什么呢？

我的理解是Flink在处理迟到数据时，watermark只是其中一个选择，它只是权衡了结果完整性与延迟。通过上面的例子也可以看出，假设这个系统的数据真的最多延迟5秒，使用watermark后，确实能处理5秒内延迟的数据，保证了结果的准确性，但是window在5秒后才触发执行，系统就有最少5秒的延迟。
但是，现实世界的情况更加复杂，我们无法预估系统的数据是否真的最多迟到5秒。因此flink还提供了其他处理迟到数据的策略。现在还没有阅读到这一部分内容，不过我估计onElement方法中的if条件分支就是为其提供的。
当window在watermark触发其计算后，该回调函数就从set集合中移除了。但是后面若来了比5秒还要延迟的数据，如迟到1分钟的数据。当我们选择了flink的其他处理迟到数据的策略后，window assginer将该迟到了1分钟的数据，正确的放入已经触发了计算，得出了结果的window中，执行onElement方法，此时，window.maxTimestamp肯定是小于ctx.currentWatermark的，会再次触发该旧的窗口执行计算，保证了结果的完整性。

------------------------ 2018/12/12 更新 ---------------------------

今天看了 WindowOperator 类，明白了if条件确实是为迟到数据提供的，以下着重分析 WindowOperator 类中的 processElement() 方法，该方法大致逻辑为：在每个元素进入时，使用 windowAssigner 指定该元素属于哪个或者哪几个window（如，slidingWindow，同一个元素属于多个window），然后每个window都执行以下逻辑：

• 首先分析，该元素属于的window是否已经过期了，如果程序没有指定处理迟到数据的策略，则迟到数据进入后就会被判定为该window已失效，该迟到数据不处理，交由GC慢慢回收

• 如果window还有效，则根据该window获取该window的状态，将新进入的元素写入该window的状态，window就是这样在状态中缓存数据的，用于后面 processWindowFunction 处理 window 的数据。

• 根据 key 与 window 找到相应的 trigger，调用 trigger 的 onElement 方法，如果此时这个数据是一个迟到的数据，则可以进入上面所说的if条件，直接出发trigger

• 根据trigger的结果，执行相应逻辑

• 注册该window的失效回调函数

下面贴上WindowOperator中 processElement() 方法的相关代码：

for (W window: elementWindows) {
 
 
                // drop if the window is already late
              if (isWindowLate(window)) {
                    continue;
                }
                isSkippedElement = false;
 
 
                windowState.setCurrentNamespace(window);
                windowState.add(element.getValue());
 
 
                triggerContext.key = key;
                triggerContext.window = window;
 
 
                TriggerResult triggerResult = triggerContext.onElement(element);
 
 
                if (triggerResult.isFire()) {
                    ACC contents = windowState.get();
                    if (contents == null) {
                        continue;
                    }
                    emitWindowContents(window, contents);
                }
 
 
                if (triggerResult.isPurge()) {
                    windowState.clear();
                }
                registerCleanupTimer(window);
            }

注意的是，在 processElement 方法内，会先判断 windowAssigner 是否是 MergingWindowAssigner，我们常用的 TumblingEventTimeWindows 是正常的 windowAssigner 接口，而 EventTimeSessionWindows 是 MergingWindowAssigner。因此这里贴的代码是针对 windowAssigner 的。

其中 isWindowLate(window) 方法，其判断条件是 该window的最大时间戳加上设置允许的最大迟到时间与当前watermark的比较。也就是 window.maxTimestamp() + allowedLateness <= watermark 。若小于，则window过期，需要删除window对象，删除window状态，触发为给window注册的onEventTime回调函数。

再再提醒注意一点：

在EventTimeTrigger中返回的triggerResult要么是 continue 要么是 fire，就是没有 purge 或者 fire_and_purge。现在看来，也是为了配合flink提供的 allowedLateness 处理迟到数据的策略。这是因为，如果trigger的结果是 purge 或者 fire_and_purge，就会在 WindowOperator 中触发 windowState.clear() 动作，这样的话，真正迟到的数据加入该window后，该window的状态已经被删除，无法更新了。但是 windowState.clear() 是为了清除该 window 的状态的，如果trigger的状态不指定为  purge 或者 fire_and_purge，该window的状态会不会删除？什么时候删除？若不删除岂不是会造成内存问题？答案就在之前说的，当判定该window过期，需要删除时，会同时删除window的状态，可以参考 WindowOperator 中的 clearAllState() 方法。

扫码关注我们

我知道你“在看”哟～