您好,登錄后才能下訂單哦!
這篇“Java Flink窗口觸發(fā)器Trigger如何使用”文章的知識點(diǎn)大部分人都不太理解,所以小編給大家總結(jié)了以下內(nèi)容,內(nèi)容詳細(xì),步驟清晰,具有一定的借鑒價(jià)值,希望大家閱讀完這篇文章能有所收獲,下面我們一起來看看這篇“Java Flink窗口觸發(fā)器Trigger如何使用”文章吧。
Trigger確定窗口(由窗口分配器形成)何時(shí)準(zhǔn)備好由窗口函數(shù)處理。每個(gè)WindowAssigner都帶有一個(gè)默認(rèn)值Trigger。如果默認(rèn)觸發(fā)器不符合您的需求,您可以使用trigger。
public abstract class Trigger<T, W extends Window> implements Serializable { /** 只要有元素落?到當(dāng)前窗?, 就會(huì)調(diào)?該?法 * @param element 收到的元素 * @param timestamp 元素抵達(dá)時(shí)間. * @param window 元素所屬的window窗口. * @param ctx ?個(gè)上下?對象,通常?該對象注冊 timer(ProcessingTime/EventTime) 回調(diào). */ public abstract TriggerResult onElement(T var1, long var2, W var4, Trigger.TriggerContext var5) throws Exception; /** * processing-time 定時(shí)器回調(diào)函數(shù) * * @param time 定時(shí)器觸發(fā)的時(shí)間. * @param window 定時(shí)器觸發(fā)的窗口對象. * @param ctx ?個(gè)上下?對象,通常?該對象注冊 timer(ProcessingTime/EventTime) 回調(diào). */ public abstract TriggerResult onProcessingTime(long var1, W var3, Trigger.TriggerContext var4) throws Exception; /** * event-time 定時(shí)器回調(diào)函數(shù) * * @param time 定時(shí)器觸發(fā)的時(shí)間. * @param window 定時(shí)器觸發(fā)的窗口對象. * @param ctx ?個(gè)上下?對象,通常?該對象注冊 timer(ProcessingTime/EventTime) 回調(diào). */ public abstract TriggerResult onEventTime(long var1, W var3, Trigger.TriggerContext var4) throws Exception; /** * 當(dāng) 多個(gè)窗口合并到?個(gè)窗?的時(shí)候,調(diào)用該方法法,例如系統(tǒng)SessionWindow * * @param window 合并后的新窗口對象 * @param ctx ?個(gè)上下?對象,通常用該對象注冊 timer(ProcessingTime/EventTime)回調(diào)以及訪問狀態(tài) */ public void onMerge(W window, Trigger.OnMergeContext ctx) throws Exception { throw new UnsupportedOperationException("This trigger does not support merging."); } /** * 當(dāng)窗口被刪除后執(zhí)?所需的任何操作。例如:可以清除定時(shí)器或者刪除狀態(tài)數(shù)據(jù) */ public abstract void clear(W var1, Trigger.TriggerContext var2) throws Exception; }
public enum TriggerResult { // 表示對窗口不執(zhí)行任何操作。即不觸發(fā)窗口計(jì)算,也不刪除元素。 CONTINUE(false, false), // 觸發(fā)窗口計(jì)算,輸出結(jié)果,然后將窗口中的數(shù)據(jù)和窗口進(jìn)行清除。 FIRE_AND_PURGE(true, true), // 觸發(fā)窗口計(jì)算,但是保留窗口元素 FIRE(true, false), // 不觸發(fā)窗口計(jì)算,丟棄窗口,并且刪除窗口的元素。 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; } }
一旦觸發(fā)器確定窗口已準(zhǔn)備好進(jìn)行處理,就會(huì)觸發(fā),返回狀態(tài)可以是FIRE或FIRE_AND_PURGE。其中FIRE是觸發(fā)窗口計(jì)算并保留窗口內(nèi)容,而FIRE_AND_PURGE是觸發(fā)窗口計(jì)算并刪除窗口內(nèi)容。默認(rèn)情況下,預(yù)實(shí)現(xiàn)的觸發(fā)器只是簡單地FIRE不清除窗口狀態(tài)。
EventTimeTrigger:通過對比EventTime和窗口的Endtime確定是否觸發(fā)窗口計(jì)算,如果EventTime大于Window EndTime則觸發(fā),否則不觸發(fā),窗口將繼續(xù)等待。
ProcessTimeTrigger:通過對比ProcessTime和窗口EndTme確定是否觸發(fā)窗口,如果ProcessTime大于EndTime則觸發(fā)計(jì)算,否則窗口繼續(xù)等待。
ContinuousEventTimeTrigger:根據(jù)間隔時(shí)間周期性觸發(fā)窗口或者Window的結(jié)束時(shí)間小于當(dāng)前EndTime觸發(fā)窗口計(jì)算。
ContinuousProcessingTimeTrigger:根據(jù)間隔時(shí)間周期性觸發(fā)窗口或者Window的結(jié)束時(shí)間小于當(dāng)前ProcessTime觸發(fā)窗口計(jì)算。
CountTrigger:根據(jù)接入數(shù)據(jù)量是否超過設(shè)定的闕值判斷是否觸發(fā)窗口計(jì)算。
DeltaTrigger:根據(jù)接入數(shù)據(jù)計(jì)算出來的Delta指標(biāo)是否超過指定的Threshold去判斷是否觸發(fā)窗口計(jì)算。
PurgingTrigger:可以將任意觸發(fā)器作為參數(shù)轉(zhuǎn)換為Purge類型的觸發(fā)器,計(jì)算完成后數(shù)據(jù)將被清理。
NeverTrigger:任何時(shí)候都不觸發(fā)窗口計(jì)算
主要看看EventTimeTrigger和ProcessingTimeTrigger的源碼。
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(); } }
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())將會(huì)注冊一個(gè)ProcessingTime定時(shí)器,時(shí)間參數(shù)是window.maxTimestamp(),也就是窗口的最終時(shí)間,當(dāng)時(shí)間到達(dá)這個(gè)窗口最終時(shí)間,定時(shí)器觸發(fā)并調(diào)用 onProcessingTime()方法,在 onProcessingTime() 方法中,return TriggerResult.FIRE 即返回 FIRE,觸發(fā)窗口中數(shù)據(jù)的計(jì)算,但是會(huì)保留窗口元素。
需要注意的是ProcessingTimeTrigger類只會(huì)在窗口的最終時(shí)間到達(dá)的時(shí)候觸發(fā)窗口函數(shù)的計(jì)算,計(jì)算完成后并不會(huì)清除窗口中的數(shù)據(jù),這些數(shù)據(jù)存儲(chǔ)在內(nèi)存中,除非調(diào)用PURGE或FIRE_AND_PURGE,否則數(shù)據(jù)將一直存在內(nèi)存中。實(shí)際上,F(xiàn)link中提供的Trigger類,除了PurgingTrigger類,其他的都不會(huì)對窗口中的數(shù)據(jù)進(jìn)行清除。
TumblingEventTimeWindows :EventTimeTrigger public 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:EventTimeTrigger public 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:EventTimeTrigger public 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 :NeverTrigger public class GlobalWindows extends WindowAssigner<Object, GlobalWindow> { public Trigger<Object, GlobalWindow> getDefaultTrigger(StreamExecutionEnvironment env) { return new GlobalWindows.NeverTrigger(); } }
以上就是關(guān)于“Java Flink窗口觸發(fā)器Trigger如何使用”這篇文章的內(nèi)容,相信大家都有了一定的了解,希望小編分享的內(nèi)容對大家有幫助,若想了解更多相關(guān)的知識內(nèi)容,請關(guān)注億速云行業(yè)資訊頻道。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。