Flink窗口触发器

窗口触发器概念

窗口的触发器定义了窗口是何时被触发并同时决定触发行为（对窗口进行清理或者计算）。触发器确定窗口(由窗口分配程序形成)何时准备由窗口函数处理。每个WindowAssigner都带有一个默认触发器。
注意：窗口的触发在内部是设置定时器来实现的。

触发器相关类

triggers包

触发器相关类

Trigger抽象类

Trigger类

触发器接口有五种方法，允许触发器对不同的事件作出反应

onElement()添加到每个窗口的元素都会调用此方法。
onEventTime()当注册的事件时间计时器触发时，将调用此方法。
onProcessingTime()当注册的处理时间计时器触发时，将调用此方法。
onMerge()与有状态触发器相关，并在两个触发器对应的窗口合并时合并它们的状态，例如在使用会话窗口时。(目前没使用过，了解不多)
clear()执行删除相应窗口时所需的任何操作。(一般是删除定义的状态、定时器等)

TriggerContext接口（定义在Trigger类中），用于维持状态，注册定时器等：

TriggerContext.png

registerXXXTimeTimer：注册（处理/事件）时间定时器；
deleteXXXTimeTimer：删除（处理/事件）时间定时器；
getPartitionedState：从Flink状态存储终端获取状态；

TriggerResult枚举类，用于决定窗口在触发后的行为：

image.png

CONTINUE：不作任何处理；
FIRE_AND_PURGE：触发窗口计算并输出结果同时清理并释放窗口（该值只会被清理触发器PurgingTrigger使用）；
FIRE：触发窗口计算并输出结果，但窗口并没有被释放并且数据仍然保留；
PURGE：不触发窗口计算，不输出结果，只清除窗口中的所有数据并释放窗口

Flink内置的触发器:

EventTimeTrigger：基于事件时间和watermark机制来对窗口进行触发计算。
ProcessingTimeTrigger:基于处理时间触发。
CountTrigger:窗口元素数超过预先给定的限制值的话会触发计算。
PurgingTrigger作为其它trigger的参数，将其转化为一个purging触发器。

源码剖析

Trigger有以下主要方法：

// 先去注册一个ProcessingTime定时器，触发时间点就是当前窗口的最大时间戳；
//触发结果就是不做任何操作。
public abstract TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception;

// ProcessingTime到来之时调用
public abstract TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception;

// EventTime到来之时调用
public abstract TriggerResult onEventTime(long time, W window, TriggerContext ctx) throws Exception;

EventTimeTrigger
当窗口的最大时间戳小于等于水位线立即触发窗口计算；否则的话就去注册EventTime定时器，结果就是不做任何操作。

// EventTimeTrigger.java

// 针对每个元素触发
@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
        // window的最大时间戳比watermark小，该window需要立刻进行计算
        return TriggerResult.FIRE;
    } else {
        // 注册一个event time事件，当watermark超过window.maxTimestamp时，会调用onEventTime方法
        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;
}

@Override
public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
    // 对于processing time，不做任何处理
    return TriggerResult.CONTINUE;
}

ProcessingTimeTrigger
依照ProcessingTime判断是否触发计算。源码如下：

@Override
public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) {
    // 和eventtime的类似，注册processing time定时器
    ctx.registerProcessingTimeTimer(window.maxTimestamp());
    return TriggerResult.CONTINUE;
}

@Override
public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
    return TriggerResult.CONTINUE;
}

@Override
public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) {
    // 返回FIRE，执行计算
    return TriggerResult.FIRE;
}

CountTrigger
每maxCount个元素触发一次计算。

@Override
public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception {
    // 获取目前的count状态值
    ReducingState<Long> count = ctx.getPartitionedState(stateDesc);
    count.add(1L);
    // 如果count值大于等于maxCount，清空状态值并且触发计算，否则不进行操作
    if (count.get() >= maxCount) {
        count.clear();
        return TriggerResult.FIRE;
    }
    return TriggerResult.CONTINUE;
}

@Override
public TriggerResult onEventTime(long time, W window, TriggerContext ctx) {
    // 不进行操作
    return TriggerResult.CONTINUE;
}

@Override
public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {
    // 不进行操作
    return TriggerResult.CONTINUE;
}

PurgingTrigger
PurgingTrigger是一个trigger的包装类。具体作用为：如果被包装的trigger触发返回FIRE，则PurgingTrigger将返回修改为FIRE_AND_PURGE，其他的返回值不做处理。
DeltaTrigger
DeltaTrigger具有一个DeltaFunction，该函数的逻辑需要用户自己定义。该函数比较上一次触发计算的元素，和目前到来的元素。比较结果为一个double类型阈值。如果阈值超过DeltaTrigger配置的阈值，会返回TriggerResult.FIRE

public TriggerResult onElement(T element, long timestamp, W window, TriggerContext ctx) throws Exception {
    // 获取上个元素的储存状态
    ValueState<T> lastElementState = ctx.getPartitionedState(stateDesc);
    // 确保存入第一个到来的元素
    if (lastElementState.value() == null) {
        lastElementState.update(element);
        return TriggerResult.CONTINUE;
    }
    if (deltaFunction.getDelta(lastElementState.value(), element) > this.threshold) {
        // 重点：只要触发条件满足的时候，才会更新lastElementState，使用新的element替代上一个element
        lastElementState.update(element);
        // 触发计算
        return TriggerResult.FIRE;
    }
    return TriggerResult.CONTINUE;
}

Flink窗口触发器

窗口触发器概念

触发器相关类

Flink内置的触发器:

源码剖析

推荐阅读更多精彩内容