Java Flink窗口触发器Trigger怎么使用

这篇文章主要介绍“Java flink窗口触发器Trigger怎么使用”，在日常操作中，相信很多人在Java Flink窗口触发器Trigger怎么使用问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”Java Flink窗口触发器Trigger怎么使用”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！

定义

Trigger确定窗口（由窗口分配器形成）何时准备好由窗口函数处理。每个WindowAssigner都带有一个默认值Trigger。如果默认触发器不符合您的需求，您可以使用trigger(…)。

Trigger 源码

public abstract class Trigger<T, W extends Window> implements Serializable {    public abstract TriggerResult onElement(T var1, long var2, W var4, Trigger.TriggerContext var5) throws Exception;     public abstract TriggerResult onProcessingTime(long var1, W var3, Trigger.TriggerContext var4) throws Exception;    public abstract TriggerResult onEventTime(long var1, W var3, Trigger.TriggerContext var4) throws Exception;     public void onMerge(W window, Trigger.OnMergeContext ctx) throws Exception {        throw new UnsupportedOperationException("This trigger does not support merging.");    }    public abstract void clear(W var1, Trigger.TriggerContext var2) throws Exception;    }

TriggerResult 源码

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;    private TriggerResult(boolean fire, boolean purge) {        this.purge = purge;        this.fire = fire;    }    public boolean isFire() {        return this.fire;    }    public boolean isPurge() {        return this.purge;    }}

一旦触发器确定窗口已准备好进行处理，就会触发，返回状态可以是FIRE或FIRE_AND_PURGE。其中FIRE是触发窗口计算并保留窗口内容，而FIRE_AND_PURGE是触发窗口计算并删除窗口内容。默认情况下，预实现的触发器只是简单地FIRE不清除窗口状态。

Flink 预置的Trigger

• EventTimeTrigger：通过对比EventTime和窗口的Endtime确定是否触发窗口计算，如果EventTime大于Window EndTime则触发，否则不触发，窗口将继续等待。

• ProcessTimeTrigger：通过对比ProcessTime和窗口EndTme确定是否触发窗口，如果ProcessTime大于EndTime则触发计算，否则窗口继续等待。

• ContinuousEventTimeTrigger：根据间隔时间周期性触发窗口或者Window的结束时间小于当前EndTime触发窗口计算。

• ContinuousProcessingTimeTrigger：根据间隔时间周期性触发窗口或者Window的结束时间小于当前ProcessTime触发窗口计算。

• CountTrigger：根据接入数据量是否超过设定的阙值判断是否触发窗口计算。

• DeltaTrigger：根据接入数据计算出来的Delta指标是否超过指定的Threshold去判断是否触发窗口计算。

• PurgingTrigger：可以将任意触发器作为参数转换为Purge类型的触发器，计算完成后数据将被清理。

• NeverTrigger：任何时候都不触发窗口计算

主要看看EventTimeTrigger和ProcessingTimeTrigger的源码。

EventTimeTrigger源码

public class EventTimeTrigger extends Trigger<Object, TimeWindow> {    private static final long serialVersionUID = 1L;    private EventTimeTrigger() {    }    public TriggerResult onElement(Object element, long timestamp, TimeWindow window, 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, TriggerContext ctx) {        return time == window.maxTimestamp() ? TriggerResult.FIRE : TriggerResult.CONTINUE;    }    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {        return TriggerResult.CONTINUE;    }    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {        ctx.deleteEventTimeTimer(window.maxTimestamp());    }    public boolean canMerge() {        return true;    }    public void onMerge(TimeWindow window, OnMergeContext ctx) {        long windowMaxTimestamp = window.maxTimestamp();        if (windowMaxTimestamp > ctx.getCurrentWatermark()) {            ctx.registerEventTimeTimer(windowMaxTimestamp);        }    }    public String toString() {        return "EventTimeTrigger()";    }    public static EventTimeTrigger create() {        return new EventTimeTrigger();    }}

ProcessingTimeTrigger源码

public class ProcessingTimeTrigger extends Trigger<Object, TimeWindow> {    private static final long serialVersionUID = 1L;    private ProcessingTimeTrigger() {    }    public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) {        ctx.registerProcessingTimeTimer(window.maxTimestamp());        return TriggerResult.CONTINUE;    }    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {        return TriggerResult.CONTINUE;    }    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) {        return TriggerResult.FIRE;    }    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {        ctx.deleteProcessingTimeTimer(window.maxTimestamp());    }    public boolean canMerge() {        return true;    }    public void onMerge(TimeWindow window, OnMergeContext ctx) {        long windowMaxTimestamp = window.maxTimestamp();        if (windowMaxTimestamp > ctx.getCurrentProcessingTime()) {            ctx.registerProcessingTimeTimer(windowMaxTimestamp);        }    }    public String toString() {        return "ProcessingTimeTrigger()";    }    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

滚动窗口

TumblingEventTimewindows ：EventTimeTriggerpublic class TumblingEventTimeWindows extends WindowAssigner<Object, TimeWindow> {    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {            return EventTimeTrigger.create();        }}

TumblingProcessingTimeWindows ：ProcessingTimeTrigger

public class TumblingProcessingTimeWindows extends WindowAssigner<Object, TimeWindow> {    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {        return ProcessingTimeTrigger.create();    }}

滑动窗口

SlidingEventTimeWindows：EventTimeTriggerpublic class SlidingEventTimeWindows extends WindowAssigner<Object, TimeWindow> {    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {        return EventTimeTrigger.create();    }}

SlidingProcessingTimeWindows ：ProcessingTimeTrigger

public class SlidingProcessingTimeWindows extends WindowAssigner<Object, TimeWindow> {    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {            return ProcessingTimeTrigger.create();        }}

会话窗口

EventTimeSessionWindows：EventTimeTriggerpublic class EventTimeSessionWindows extends MergingWindowAssigner<Object, TimeWindow> {    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {        return EventTimeTrigger.create();    }}

ProcessingTimeSessionWindows：ProcessingTimeTrigger

public class ProcessingTimeSessionWindows extends MergingWindowAssigner<Object, TimeWindow> {    public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {        return ProcessingTimeTrigger.create();    }}

全局窗口

GlobalWindows ：NeverTriggerpublic class GlobalWindows extends WindowAssigner<Object, GlobalWindow> {     public Trigger<Object, GlobalWindow> getDefaultTrigger(StreamExecutionEnvironment env) {            return new GlobalWindows.NeverTrigger();        }}

到此，关于“Java Flink窗口触发器Trigger怎么使用”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注编程网网站，小编会继续努力为大家带来更多实用的文章！