Flink DataStream Trigger触发器

野性酷女 2021-12-20 12:09 462阅读 0赞

## Flink DataStream Trigger触发器 ##

触发器是指当`Window`（window 分配器产生）窗口准备通过`WindowFunction`计算时触发。每一个Window分配器都会有一个默认的`Trigger(触发器)`。如果默认的`Trigger（触发器）`不满足你的需要，你可以自定义触发器。

### 触发器的方法 ###

触发器的有4个抽象方法，2个具有实现方法。

抽象方法：

*  onElement()：每个元素进入窗口都会被调用，有4个参数
    
     *  T element：每个元素实例对象
     *  long timestamp：每个元素抵达时间
     *  W window：元素添加的窗口
     *  TriggerContext ctx：上下文对象，用于注册timer回调
 *  onProcessingTime()：当使用Processing Time的定时器被trigger触发时调用。
    
    参数
    
     *  long time：timer定时器触发时间戳
     *  W window：timer定时器触发的Window窗口
     *  TriggerContext ctx：触发器上下文对象
 *  onEventTime()：当使用Processing Time的定时器被trigger触发时调用。
    
    参数
    
     *  long time：timer定时器触发时间戳
     *  W window：timer定时器触发的Window窗口
     *  TriggerContext ctx：触发器上下文对象
 *  clear()：清除窗口
    
    参数：
    
     *  W Window：timer定时器触发的Window窗口
     *  TriggerContext ctx：触发器上下文对象

具有实现的方法：

*  canMerge()：默认返回false

如果需要用到Merge窗口，需要重写方法

*  onMerge()：多个窗口合并时，调用此方法
    
    参数有
    
     *  W Window：Window窗口对象
     *  OnMergeContext ctx：TriggerContext的子类。

例如：

**ProcessingTimeTrigger**示例：

@PublicEvolving
    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();
    	}
    }

#### TriggerResult触发结果值 ####

*  **CONTIUNE**：啥都不做，不触发，也不清除
 *  **FIRE**：触发窗口计算，例如上方示例，在onProcessingTime方法调用时，就触发计算。
 *  **PURGE**：清除窗口内容
 *  **FIRE\_AND\_PURGE**：触发之后并清除窗口的内容

PURGE只会删除窗口的内容，并保留有关窗口和任何触发器状态的任何潜在元信息。

#### 常见的Trigger ####

对于每一个Window Assigners(窗口分配器)都有一个默认的Trigger（触发器）。

例如：所有的Event Window的默认触发器是EventTimeTrigger。

**GlobalWindow的默认触发器是NeverTriger**。NerverTrigger的所有返回TriggerResult对象是CONTINUE。

下面是常见的Trigger：

*  **EventTimeTrigger**：基于EventTime Window的默认触发器
 *  **ProcessingTimeTrigger**：基于ProcessingTime Window的默认触发器
 *  **CountTrigger**：基于Count Window的默认触发器。
 *  PurgingTrigger：内部使用，用于清除窗口内容
 *  **NeverTrigger**：永不触发的触发器。

如果不满足需求可以自定义。需要在onElement方法中注册timer，在对应的onEventTime或onProcessingTime方法实现相应的逻辑。

--------------------

顺便备注下在**基于Event Time 的Session Window窗口**使用Trigger时的一些问题。

1.基于Event Time的Session窗口，使用EventTimeTrigger触发（默认），如果下一条水印不出现，到了窗口指定的时间，onEventTime方法不会被触发。

2.改用ProcessingTimeTrigger，可以到点进行触发。

代码片段：

...
    .window(EventTimeSessionWindows.withGap(Time.minutes(15)))
    // 通过Processing Time触发，可以解决长时间没有Event Time的watermark进入，无法触发窗口的问题
    .trigger(ProcessingTimeTrigger.create())
    ...