Flink源码分析系列文档目录

什么是Trigger

Trigger是用来决定是否触发窗格计算操作的。
窗格的定义为一组key相同（分区操作），并且位于同一个窗口中的元素。每个窗格都有自己的Trigger对象。
Trigger会被复用或重建，因此Trigger中不能存储状态数据。如果需要储存，可以写入到TriggerContext中。

TriggerResult

每个trigger事件方法调用后会返回TriggerResult，决定接下来的处理行为。

CONTINUE: 不做任何处理
FIRE：触发计算
PURGE：清除所有window中的元素，并且该window被废弃
FIRE_AND_PURGE：计算之后执行purge操作

Trigger有以下主要方法：

// 每个元素进入窗口时调用
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;

1. 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) {
    // 当前时间为window的最大时间戳，触发计算
    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;
}

2. 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;
}

3. 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;
}

4. PurgingTrigger

PurgingTrigger是一个trigger的包装类。具体作用为：如果被包装的trigger触发返回FIRE，则PurgingTrigger将返回修改为FIRE_AND_PURGE，其他的返回值不做处理。

5. 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 源码之Trigger