溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
這篇“Java Flink窗口觸發器Trigger如何使用”文章的知識點大部分人都不太理解，所以小編給大家總結了以下內容，內容詳細，步驟清晰，具有一定的借鑒價值，希望大家閱讀完這篇文章能有所收獲，下面我們一起來看看這篇“Java Flink窗口觸發器Trigger如何使用”文章吧。
Trigger確定窗口（由窗口分配器形成）何時準備好由窗口函數處理。每個WindowAssigner都帶有一個默認值Trigger。如果默認觸發器不符合您的需求，您可以使用trigger。
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 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;
}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不清除窗口狀態。
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的源碼。
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())將會注冊一個ProcessingTime定時器，時間參數是window.maxTimestamp()，也就是窗口的最終時間，當時間到達這個窗口最終時間，定時器觸發并調用 onProcessingTime()方法，在 onProcessingTime() 方法中，return TriggerResult.FIRE 即返回 FIRE，觸發窗口中數據的計算，但是會保留窗口元素。
需要注意的是ProcessingTimeTrigger類只會在窗口的最終時間到達的時候觸發窗口函數的計算，計算完成后并不會清除窗口中的數據，這些數據存儲在內存中，除非調用PURGE或FIRE_AND_PURGE，否則數據將一直存在內存中。實際上，Flink中提供的Trigger類，除了PurgingTrigger類，其他的都不會對窗口中的數據進行清除。
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();
}
}以上就是關于“Java Flink窗口觸發器Trigger如何使用”這篇文章的內容，相信大家都有了一定的了解，希望小編分享的內容對大家有幫助，若想了解更多相關的知識內容，請關注億速云行業資訊頻道。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
