Flink触发器Trigger

前言

在 Flink 窗口计算模型中，数据先经过 WindowAssigner 分配窗口，然后再经过触发器 Trigger，Trigger 决定了一个窗口何时被 ProcessFunction 处理。每个 WindowAssigner 都有一个默认的 Trigger，如果默认的不满足需求，可以通过 WindowedStream.trigger() 指定自定义的 Trigger。

认识Trigger

所有触发器都是org.apache.flink.streaming.api.windowing.triggers.Trigger的子类，父类定义了一个触发器应该具备的能力：

@PublicEvolving
public abstract class Trigger<T, W extends Window> implements Serializable {private static final long serialVersionUID = -4104633972991191369L;public Trigger() {}public abstract TriggerResult onElement(T var1, long var2, W var4, TriggerContext var5) throws Exception;public abstract TriggerResult onProcessingTime(long var1, W var3, TriggerContext var4) throws Exception;public abstract TriggerResult onEventTime(long var1, W var3, TriggerContext var4) throws Exception;public boolean canMerge() {return false;}public void onMerge(W window, OnMergeContext ctx) throws Exception {throw new UnsupportedOperationException("This trigger does not support merging.");}public abstract void clear(W var1, TriggerContext var2) throws Exception;
}

Trigger 抽象类提供了六个方法：

onElement 元素被加入到窗口时触发，返回值决定窗口是否计算
onProcessingTime 注册的ProcessingTime任务时间到达时触发
onEventTime 注册的EventTime任务时间到达时触发
canMerge 是否可以合并
onMerge Trigger合并
clear 窗口被移除时触发

如果数据本身携带窗口是否触发计算的标记，那么重写 onElement() 方法即可；但是在时间窗口计算模型下，并不是通过元素来判断窗口是否需要计算的，而是窗口的结束时间到达时才出发计算，这个时候就需要用到定时器 Timer。Timer 就像一个闹钟，我们可以在它上面注册一个未来的时间戳，当这个时间到达时，对应的事件就会被触发，就像闹钟喊醒沉睡的你一样。

Trigger 会有自己的 Timer，TriggerContext 提供了注册时间事件的方法，你可以根据自己采用的时间语义，调用对应的注册方法来注册事件。

triggerContext.registerProcessingTimeTimer();
triggerContext.registerEventTimeTimer();

在重写这三个方法时，要重点关注方法的返回值。方法的返回值有两个作用：1、窗口内的数据是否可以计算，2、窗口内的数据是否需要清理。

public enum TriggerResult {CONTINUE(false, false),FIRE_AND_PURGE(true, true),FIRE(true, false),PURGE(false, true);private final boolean fire;private final boolean purge;
}

TriggerResult 枚举有四个值，含义分别是：

CONTINUE 不做任何操作
FIRE 触发窗口计算，但是数据仍然保留
PURGE 不触发计算，只是清理窗口内数据
FIRE_AND_PURGE 触发窗口计算，同时清理数据

内置的Trigger

Flink 内置了许多常用的 Trigger，大多数情况下它们足以支撑我们的业务场景，只有当内置的Trigger不符合要求时，才需要开发自定义的 Trigger。

1、EventTimeTrigger

和事件时间窗口搭配使用的 Trigger，直到 Watermark 时间戳等于窗口的结束时间才会触发计算。

onElement 的逻辑是：当有元素加入到窗口时，先判断当前 Watermark 时间戳是否到达窗口的结束时间，如果到了就直接触发计算，否则注册一个时间事件，等待 Timer 触发窗口计算。

public class EventTimeTrigger extends Trigger<Object, TimeWindow> {private static final long serialVersionUID = 1L;private EventTimeTrigger() {}public TriggerResult onElement(Object element, long timestamp, TimeWindow window, Trigger.TriggerContext ctx) throws Exception {if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {return TriggerResult.FIRE;} else {ctx.registerEventTimeTimer(window.maxTimestamp());return TriggerResult.CONTINUE;}}public TriggerResult onEventTime(long time, TimeWindow window, Trigger.TriggerContext ctx) {return time == window.maxTimestamp() ? TriggerResult.FIRE : TriggerResult.CONTINUE;}。。。。。。
}

2、ProcessingTimeTrigger

和 EventTimeTrigger 差不多，区别是采用的处理时间语义，没有 Watermark 相关的判断，直接注册ProcessingTime 事件等待窗口触发计算。

public class ProcessingTimeTrigger extends Trigger<Object, TimeWindow> {private static final long serialVersionUID = 1L;private ProcessingTimeTrigger() {}public TriggerResult onElement(Object element, long timestamp, TimeWindow window, Trigger.TriggerContext ctx) {ctx.registerProcessingTimeTimer(window.maxTimestamp());return TriggerResult.CONTINUE;}public TriggerResult onProcessingTime(long time, TimeWindow window, Trigger.TriggerContext ctx) {return TriggerResult.FIRE;}
}

3、DeltaTrigger

DeltaTrigger 会计算新加入窗口的元素和上一个元素的差值，当这个差值超过给定的阈值时，窗口就会触发计算，元素之间的差值是通过指定的 DeltaFunction 计算出来的。

public class DeltaTrigger<T, W extends Window> extends Trigger<T, W> {private static final long serialVersionUID = 1L;private final DeltaFunction<T> deltaFunction;private final double threshold;private final ValueStateDescriptor<T> stateDesc;private DeltaTrigger(double threshold, DeltaFunction<T> deltaFunction, TypeSerializer<T> stateSerializer) {this.deltaFunction = deltaFunction;this.threshold = threshold;this.stateDesc = new ValueStateDescriptor("last-element", stateSerializer);}public TriggerResult onElement(T element, long timestamp, W window, Trigger.TriggerContext ctx) throws Exception {ValueState<T> lastElementState = (ValueState)ctx.getPartitionedState(this.stateDesc);if (lastElementState.value() == null) {lastElementState.update(element);return TriggerResult.CONTINUE;} else if (this.deltaFunction.getDelta(lastElementState.value(), element) > this.threshold) {lastElementState.update(element);return TriggerResult.FIRE;} else {return TriggerResult.CONTINUE;}}
}

自定义Trigger

通过子类继承 Trigger 重写相应的方法，即可自定义我们自己的触发器。

举个例子，我们自定义一个和时间不相关的 Trigger，我们等窗口积攒到一定数量的元素再出发计算。如下示例程序：

public static class CounterTrigger extends Trigger<Long, GlobalWindow> {private final int count;public CounterTrigger(int count) {this.count = count;}@Overridepublic TriggerResult onElement(Long element, long timestamp, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {// 通过Flink state来保存窗口内积攒的元素数量ValueState<Integer> countState = triggerContext.getPartitionedState(new ValueStateDescriptor<Integer>("count", Integer.class));int elementCount = Optional.ofNullable(countState.value()).orElse(0) + 1;if (elementCount >= this.count) {countState.update(0);return TriggerResult.FIRE_AND_PURGE;}countState.update(elementCount);return TriggerResult.CONTINUE;}@Overridepublic TriggerResult onProcessingTime(long timestamp, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {return null;}@Overridepublic TriggerResult onEventTime(long timestamp, GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {return null;}@Overridepublic void clear(GlobalWindow globalWindow, TriggerContext triggerContext) throws Exception {}}

接下来验证一下我们的Trigger是否生效。

我们编写一个Flink作业，数据源每秒生成10个随机数，数据会被统一划分到 GlobalWindow 窗口，然后指定我们自定义的 CounterTrigger，等窗口内积攒了十条数据就出发求和计算。

public static void main(String[] args) throws Exception {StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();environment.addSource(new SourceFunction<Long>() {@Overridepublic void run(SourceContext<Long> sourceContext) throws Exception {while (true) {Threads.sleep(100);sourceContext.collect(ThreadLocalRandom.current().nextLong(100));}}@Overridepublic void cancel() {}}).windowAll(GlobalWindows.create()).trigger(new CounterTrigger(10)).process(new ProcessAllWindowFunction<Long, Object, GlobalWindow>() {@Overridepublic void process(ProcessAllWindowFunction<Long, Object, GlobalWindow>.Context context, Iterable<Long> iterable, Collector<Object> collector) throws Exception {Iterator<Long> iterator = iterable.iterator();long sum = 0L;while (iterator.hasNext()) {sum += iterator.next();}System.err.println(sum);}});environment.execute();
}