Java Flink窗口触发器Trigger怎么使用 |
您所在的位置：网站首页 › java函数用法 › Java Flink窗口触发器Trigger怎么使用 |
Java Flink窗口触发器Trigger怎么使用
| 定义 Trigger确定窗口（由窗口分配器形成）何时准备好由窗口函数处理。每个WindowAssigner都带有一个默认值Trigger。如果默认触发器不符合您的需求，您可以使用trigger(…)。 Trigger 源码public abstract class Trigger implements Serializable { /** 只要有元素落⼊到当前窗⼝， 就会调⽤该⽅法 * @param element 收到的元素 * @param timestamp 元素抵达时间. * @param window 元素所属的window窗口. * @param ctx ⼀个上下⽂对象，通常⽤该对象注册 timer（ProcessingTime/EventTime） 回调. */ public abstract TriggerResult onElement(T var1, long var2, W var4, Trigger.TriggerContext var5) throws Exception; /** * processing-time 定时器回调函数 * * @param time 定时器触发的时间. * @param window 定时器触发的窗口对象. * @param ctx ⼀个上下⽂对象，通常⽤该对象注册 timer（ProcessingTime/EventTime） 回调. */ public abstract TriggerResult onProcessingTime(long var1, W var3, Trigger.TriggerContext var4) throws Exception; /** * event-time 定时器回调函数 * * @param time 定时器触发的时间. * @param window 定时器触发的窗口对象. * @param ctx ⼀个上下⽂对象，通常⽤该对象注册 timer（ProcessingTime/EventTime） 回调. */ public abstract TriggerResult onEventTime(long var1, W var3, Trigger.TriggerContext var4) throws Exception; /** * 当 多个窗口合并到⼀个窗⼝的时候，调用该方法法，例如系统SessionWindow * * @param window 合并后的新窗口对象 * @param ctx ⼀个上下⽂对象，通常用该对象注册 timer（ProcessingTime/EventTime）回调以及访问状态 */ 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 预置的TriggerEventTimeTrigger：通过对比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 { 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()) { ctx.registerEventTimeTimer(windowMaxTimestamp); } } public String toString() { return "EventTimeTrigger()"; } public static EventTimeTrigger create() { return new EventTimeTrigger(); } }登录后复制ProcessingTimeTrigger源码public class ProcessingTimeTrigger extends Trigger { 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 ：EventTimeTrigger public class TumblingEventTimeWindows extends WindowAssigner { public Trigger getDefaultTrigger(StreamExecutionEnvironment env) { return EventTimeTrigger.create(); } }登录后复制TumblingProcessingTimeWindows ：ProcessingTimeTrigger public class TumblingProcessingTimeWindows extends WindowAssigner { public Trigger getDefaultTrigger(StreamExecutionEnvironment env) { return ProcessingTimeTrigger.create(); } }登录后复制滑动窗口SlidingEventTimeWindows：EventTimeTrigger public class SlidingEventTimeWindows extends WindowAssigner { public Trigger getDefaultTrigger(StreamExecutionEnvironment env) { return EventTimeTrigger.create(); } }登录后复制SlidingProcessingTimeWindows ：ProcessingTimeTrigger public class SlidingProcessingTimeWindows extends WindowAssigner { public Trigger getDefaultTrigger(StreamExecutionEnvironment env) { return ProcessingTimeTrigger.create(); } }登录后复制会话窗口EventTimeSessionWindows：EventTimeTrigger public class EventTimeSessionWindows extends MergingWindowAssigner { public Trigger getDefaultTrigger(StreamExecutionEnvironment env) { return EventTimeTrigger.create(); } }登录后复制ProcessingTimeSessionWindows：ProcessingTimeTrigger public class ProcessingTimeSessionWindows extends MergingWindowAssigner { public Trigger getDefaultTrigger(StreamExecutionEnvironment env) { return ProcessingTimeTrigger.create(); } }登录后复制全局窗口GlobalWindows ：NeverTrigger public class GlobalWindows extends WindowAssigner { public Trigger getDefaultTrigger(StreamExecutionEnvironment env) { return new GlobalWindows.NeverTrigger(); } }登录后复制以上就是Java Flink窗口触发器Trigger怎么使用的详细内容，更多请关注php中文网其它相关文章！ |
【本文地址】
今日新闻 |
点击排行 |
|
推荐新闻 |
图片新闻 |
|
专题文章 |