在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() 方法。
