【深入浅出flink】第10篇：聊聊flink窗口计算的触发器（Trigger）

Trigger（触发器）决定了什么时候窗口准备就绪了，一旦窗口准备就绪就可以使用WindowFunction（窗口计算操作）进行计算。每一个 WindowAssigner（窗口分配器）都会有一个默认的Trigger。如果默认的Trigger不满足用户的需求，用户可以自定义Trigger。

雷恩Layne

6474人浏览 · 2022-01-17 20:59:21

雷恩Layne · 2022-01-17 20:59:21 发布

大家好，我是雷恩Layne，这是《深入浅出flink》系列的第十篇文章，希望能对您有所收获O(∩_∩)O

每一种窗口分配器对应的默认触发器如下：

现在，就让我们一探Flink Trigger的究竟吧！

文章目录

Flink中预置的Trigger

窗口的计算触发依赖于窗口触发器，每种类型的窗口都有对应的窗口触发机制，都有一个默认的窗口触发器，触发器的作用就是去控制什么时候来触发计算。flink内部定义多种触发器，每种触发器对应于不同的WindowAssigner。常见的触发器如下：

EventTimeTrigger：通过对比EventTime和窗口的Endtime确定是否触发窗口计算，如果EventTime大于Window EndTime则触发，否则不触发，窗口将继续等待。
ProcessTimeTrigger：通过对比ProcessTime和窗口EndTme确定是否触发窗口，如果ProcessTime大于EndTime则触发计算，否则窗口继续等待。
ContinuousEventTimeTrigger：根据间隔时间周期性触发窗口或者Window的结束时间小于当前EndTime触发窗口计算。
ContinuousProcessingTimeTrigger：根据间隔时间周期性触发窗口或者Window的结束时间小于当前ProcessTime触发窗口计算。
CountTrigger：根据接入数据量是否超过设定的阙值判断是否触发窗口计算。
DeltaTrigger：根据接入数据计算出来的Delta指标是否超过指定的Threshold去判断是否触发窗口计算。
PurgingTrigger：可以将任意触发器作为参数转换为Purge类型的触发器，计算完成后数据将被清理。
NeverTrigger：任何时候都不触发窗口计算

需要注意的是，在EventTime时间语义下，如果设有Watermark，触发的时间还要再加上Watermark的延迟时长。比如EventTimeTrigger，在设置有Watermark的情况下，只有EventTime加上Watermark的延迟时长大于 Window EndTime触发，否则不触发，窗口将继续等待。

注意，在EventTime时间语义，一定会有Watermark机制，如果EventTime没有设置Watermark，会报错。ProcessingTime是根据系统时间判断的，没有Watermark机制。

上面也提到了，因为每一个WindowAssigner（窗口分配器）都会有⼀个默认的Trigger，所以下面两段代码的执行效果完全一样：

代码1：

DataStream<String> dataStream = env.socketTextStream("localhost", 7777);
DataStream<Integer> mapDataStream = dataStream.map(value -> Integer.parseInt(value));
//定义一个大小为10s的滚动时间窗口
AllWindowedStream<Integer, TimeWindow> allWindowedStream = mapDataStream
		.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10)))
//求窗口内的最小值
DataStream<Integer> minDataStream = allWindowedStream.min(0);
minDataStream.print();

代码2：

DataStream<String> dataStream = env.socketTextStream("localhost", 7777);
DataStream<Integer> mapDataStream = dataStream.map(value -> Integer.parseInt(value));
//定义一个大小为10s的滚动时间窗口
AllWindowedStream<Integer, TimeWindow> allWindowedStream = mapDataStream
		.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(10)))
		.trigger(ProcessingTimeTrigger.create());
//求窗口内的最小值
DataStream<Integer> minDataStream = allWindowedStream.min(0);
minDataStream.print();

代码中窗口分配器TumblingProcessingTimeWindows中默认的触发器就是ProcessingTimeTrigger，所以我们定不定义触发器效果都一样。

如果我们在代码中指定了窗口的trigger，默认的 trigger 将会被覆盖，不会起作用。所以，上述代码2中默认的trigger没有起作用，起作用的是我们指定的trigger。

另外，像GlobalWindows这样的窗口分配器，其本身默认的触发器是NeverTrigger，即永远不会触发，所以使用时一般要配合触发器使用。

举例：使用GlobalWindows实现一个滚动计数窗口，并计算窗口内的最小值

DataStream<String> dataStream = env.socketTextStream("localhost", 7777);
DataStream<Integer> mapDataStream = dataStream.map(value -> Integer.parseInt(value));

//定义一个大小为15的滚动计数窗口
AllWindowedStream<Integer, GlobalWindow> allWindowedStream = mapDataStream.windowAll(GlobalWindows.create()).trigger(PurgingTrigger.of(CountTrigger.of(15)));
//求窗口内的最小值
DataStream<Integer> minDataStream = allWindowedStream.min(0);

Trigger的抽象类

触发器抽象类具有四种抽象方法，这些方法允许触发器对不同事件做出反应：

onElement：在窗口中每进入一条数据的时候调用一次
onProcessingTime：根据窗口中最新的ProcessingTime判断是否满足定时器的条件，如果满足，将触发ProcessingTime定时器，并执行定时器的回调函数，即执行onProcessingTime方法里的逻辑
onEventTime：根据窗口中最新的EventTim判断是否满足定时器的条件，如果满足，将触发EventTime定时器，并执行定时器的回调函数，即onEventTime方法里的逻辑
clear：在窗口清除的时候调用

触发器接口的源码如下：

public abstract class Trigger<T, W extends Window> implements Serializable {
/**
 只要有元素落⼊到当前窗⼝， 就会调⽤该⽅法
 * @param element 收到的元素
 * @param timestamp 元素抵达时间.
 * @param window 元素所属的window窗口.
 * @param ctx ⼀个上下⽂对象，通常⽤该对象注册 timer（ProcessingTime/EventTime） 回调.
 */
 public abstract TriggerResult onElement(T element, long timestamp, W window,
TriggerContext ctx) throws Exception;
 /**
 * processing-time 定时器回调函数
 *
 * @param time 定时器触发的时间.
 * @param window 定时器触发的窗口对象.
 * @param ctx ⼀个上下⽂对象，通常⽤该对象注册 timer（ProcessingTime/EventTime） 回调.
 */
 public abstract TriggerResult onProcessingTime(long time, W window, TriggerContext
ctx) throws Exception;

 /**
 * event-time 定时器回调函数
 *
 * @param time 定时器触发的时间.
 * @param window 定时器触发的窗口对象.
 * @param ctx ⼀个上下⽂对象，通常⽤该对象注册 timer（ProcessingTime/EventTime） 回调.
 */
 public abstract TriggerResult onEventTime(long time, W window, TriggerContext ctx)
throws Exception;

 /**
 * 当 多个窗口合并到⼀个窗⼝的时候，调用该方法法，例如系统SessionWindow
 *
 * @param window 合并后的新窗口对象
 * @param ctx ⼀个上下⽂对象，通常用该对象注册 timer（ProcessingTime/EventTime）回调以及访问
状态
 */
 public void onMerge(W window, OnMergeContext ctx) throws Exception {
 throw new UnsupportedOperationException("This trigger does not support merging.");
 }
 /**
 * 当窗口被删除后执⾏所需的任何操作。例如：可以清除定时器或者删除状态数据
 */
 public abstract void clear(W window, TriggerContext ctx) throws Exception;
}

关于上述方法，需要注意三件事：

（1）前三个方法返回TriggerResult枚举类型，其包含四个枚举值：

CONTINUE：表示对窗口不执行任何操作。即不触发窗口计算，也不删除元素。
FIRE：触发窗口计算，但是保留窗口元素
PURGE：不触发窗口计算，丢弃窗口，并且删除窗口的元素。
FIRE_AND_PURGE：触发窗口计算，输出结果，然后将窗口中的数据和窗口进行清除。

源码如下：

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;//是否清除窗口元素
 ...
 }

（2）这些方法法中的任何一种都可以用于注册计时器以用于将来的操作。

（3）每一个窗口分配器都拥有一个属于自己的 Trigger，Trigger上会有定时器，用来决定一个窗口何时能够被计算或清除，当定时器触发后，会调用对应的回调返回，返回TriggerResult。Trigger的返回结果可以是 continue（不做任何操作），fire（处理窗口数据），purge（移除窗口和窗口中的数据），或者 fire + purge。一个Trigger的调用结果只是fire的话，那么会计算窗口并保留窗口原样，也就是说窗口中的数据仍然保留不变，等待下次Trigger fire的时候再次执行计算。一个窗口可以被重复计算多次知道它被 purge 了。在purge之前，窗口会一直占用着内存。

分析Trigger源码实现

ProcessingTimeTrigger源码实现如下：

public class ProcessingTimeTrigger extends Trigger<Object, TimeWindow> {
	private static final long serialVersionUID = 1L;

	private ProcessingTimeTrigger() {}

	@Override
	public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) {
		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) {
		return TriggerResult.FIRE;
	}

	@Override
	public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
		ctx.deleteProcessingTimeTimer(window.maxTimestamp());
	}

	@Override
	public boolean canMerge() {
		return true;
	}

	@Override
	public void onMerge(TimeWindow window,
			OnMergeContext ctx) {
		// only register a timer if the time is not yet past the end of the merged window
		// this is in line with the logic in onElement(). If the time is past the end of
		// the window onElement() will fire and setting a timer here would fire the window twice.
		long windowMaxTimestamp = window.maxTimestamp();
		if (windowMaxTimestamp > ctx.getCurrentProcessingTime()) {
			ctx.registerProcessingTimeTimer(windowMaxTimestamp);
		}
	}

	@Override
	public String toString() {
		return "ProcessingTimeTrigger()";
	}

	/**
	 * Creates a new trigger that fires once system time passes the end of the window.
	 */
	public static ProcessingTimeTrigger create() {
		return new ProcessingTimeTrigger();
	}

}

在 onElement()方法中，ctx.registerProcessingTimeTimer(window.maxTimestamp())将会注册一个ProcessingTime定时器，时间参数是window.maxTimestamp()，也就是窗口的最终时间，当时间到达这个窗口最终时间，定时器触发并调用 onProcessingTime()方法，在 onProcessingTime() 方法中，return TriggerResult.FIRE 即返回 FIRE，触发窗口中数据的计算，但是会保留窗口元素。

需要注意的是ProcessingTimeTrigger类只会在窗口的最终时间到达的时候触发窗口函数的计算，计算完成后并不会清除窗口中的数据，这些数据存储在内存中，除非调用PURGE或FIRE_AND_PURGE，否则数据将一直存在内存中。实际上，Flink中提供的Trigger类，除了PurgingTrigger类，其他的都不会对窗口中的数据进行清除。

Trigger中的定时器

Trigger中的定时器是由Flink Timer实现的，其实是一种用于感知并利用处理时间(ProcessingTime)或事件时间(EventTime)变化的机制。

Timer会由Flink按key+timestamp自动去重的，也就是说如果你的key有N个，并且注册的timestamp相同的话，那么实际只会注册N个Timer。

只有在KeyedStream才会有多个key，如果没有KeyedStream，此时一定是windowAll开窗的，并行度为1。

ProcessingTimeTrigger在onElement设置的定时器：

	public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) {
		ctx.registerProcessingTimeTimer(window.maxTimestamp());
		return TriggerResult.CONTINUE;
	}

ProcessingTime通过registerProcessingTimeTimer注册定时器，在系统时间戳达到Timer设定的时间戳时触发。

EventTimeTriggerr在onElement设置的定时器：

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