ToB企服应用市场:ToB评测及商务社交产业平台

标题: Flink 窗口触发器(Trigger)(一) [打印本页]

---

作者: 王國慶    时间: 2024-8-28 02:09
标题: Flink 窗口触发器(Trigger)(一)
Flink 窗口触发器(Trigger)(一)
Flink 窗口触发器(Trigger)(二)
Flink的窗口触发器（Trigger）是流处置惩罚中一个非常关键的概念，它界说了窗口何时被触发并决定触发后的行为（如进行窗口数据的计算或清算）。
一、基本概念

• 界说：触发器决定了窗口何时被触发以及触发后的行为。在Flink中，窗口的触发是通过设置定时器来实现的。
  
• 作用：控制窗口数据的聚合机会，确保数据在得当的时间点被处置惩罚和输出。
  
  
  
  

二、类型

Flink提供了多种内置的触发器，以下几种为常用类型：

• EventTimeTrigger
  

• 工作原理：基于事件时间和水印（Watermark）机制来触发窗口计算。当窗口的最大时间戳小于便是当前的水印时，立即触发窗口计算。
  
• 适用场景：适用于须要基于事件时间进行处置惩罚的场景，如金融生意业务、日记分析等。
  

• ProcessingTimeTrigger
  

• 工作原理：基于处置惩罚时间（即机器的系统时间）来触发窗口计算。当处置惩罚时间达到窗口的竣事时间时，触发窗口计算。
  
• 适用场景：适用于对时间精度要求不高的场景，或者当事件时间无法正确获取时。
  

• CountTrigger
  

• 工作原理：根据窗口内元素的数量来触发计算。当窗口内的元素数量达到预设的阈值时，触发窗口计算。
  
• 适用场景：适用于须要基于数据量进行处置惩罚的场景，如批量数据处置惩罚、流量分析等。
  

• ContinuousEventTimeTrigger和ContinuousProcessingTimeTrigger
  

• 工作原理：根据间隔时间周期性触发窗口计算，或者当窗口的竣事时间小于当前的时间（事件时间或处置惩罚时间）时触发计算。
  
• 适用场景：适用于须要周期性处置惩罚数据的场景，如实时监控、周期性报表等。
  

• DeltaTrigger
  

• 工作原理：根据接入数据计算出的Delta指标是否凌驾指定的阈值来触发窗口计算。
  
• 适用场景：适用于须要基于数据变化量进行处置惩罚的场景，如异常检测、趋势分析等。
  

• PurgingTrigger
  

• 工作原理：将其他触发器作为参数转换为Purge类型的触发器，在触发计算后清除窗口内的数据。
  
• 适用场景：适用于须要在计算完成后立即清除窗口数据的场景，以节流存储空间。
  

三、关键方法

触发器通常包含以下几个关键方法：

• onElement(T element, long timestamp, W window, TriggerContext ctx)
  当元素被添加到窗口时调用，用于注册定时器或更新窗口状态。
  
• onEventTime(long time, W window, TriggerContext ctx)
  当事件时间计时器触发时调用，用于处置惩罚事件时间相关的触发逻辑。
  
• onProcessingTime(long time, W window, TriggerContext ctx)
  当处置惩罚时间计时器触发时调用，用于处置惩罚处置惩罚时间相关的触发逻辑。
  
• onMerge(W window, OnMergeContext ctx)（可选）
  当两个窗口合并时调用，用于合并窗口的状态和定时器。
  
• clear(W window, TriggerContext ctx)
  当窗口被删除时调用，用于清算窗口的状态和定时器。
  

四、行为

触发器在触发时会返回一个TriggerResult枚举值，以决定窗口的后续行为。常见的TriggerResult值包括：

• CONTINUE：表现不进行任何操纵，等候下一个触发条件。
  
• FIRE：表现触发窗口计算并输出结果，但窗口状态保持稳固。
  
• PURGE：表现不触发窗口计算，只清除窗口内的数据和状态。
  
• FIRE_AND_PURGE：表现触发窗口计算并输出结果，然后清除窗口内的数据和状态。
  

   Flink的窗口触发器是流处置惩罚中非常机动且强盛的工具，它答应开发者根据现实需求界说窗口的触发条件和触发后的行为。通过选择合适的触发器和配置相应的参数，可以实现高效、正确的流数据处置惩罚。
  五、Trigger

EventTimeTrigger

1. //
2. // Source code recreated from a .class file by IntelliJ IDEA
3. // (powered by FernFlower decompiler)
4. //
6. package org.apache.flink.streaming.api.windowing.triggers;
8. import org.apache.flink.annotation.PublicEvolving;
9. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
11. @PublicEvolving
12. public class EventTimeTrigger extends Trigger<Object, TimeWindow> {
13.     private static final long serialVersionUID = 1L;
15.     private EventTimeTrigger() {
16.     }
18.     public TriggerResult onElement(Object element, long timestamp, TimeWindow window, Trigger.TriggerContext ctx) throws Exception {
19.         if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
20.             return TriggerResult.FIRE;
21.         } else {
22.             ctx.registerEventTimeTimer(window.maxTimestamp());
23.             return TriggerResult.CONTINUE;
24.         }
25.     }
27.     public TriggerResult onEventTime(long time, TimeWindow window, Trigger.TriggerContext ctx) {
28.         return time == window.maxTimestamp() ? TriggerResult.FIRE : TriggerResult.CONTINUE;
29.     }
31.     public TriggerResult onProcessingTime(long time, TimeWindow window, Trigger.TriggerContext ctx) throws Exception {
32.         return TriggerResult.CONTINUE;
33.     }
35.     public void clear(TimeWindow window, Trigger.TriggerContext ctx) throws Exception {
36.         ctx.deleteEventTimeTimer(window.maxTimestamp());
37.     }
39.     public boolean canMerge() {
40.         return true;
41.     }
43.     public void onMerge(TimeWindow window, Trigger.OnMergeContext ctx) {
44.         long windowMaxTimestamp = window.maxTimestamp();
45.         if (windowMaxTimestamp > ctx.getCurrentWatermark()) {
46.             ctx.registerEventTimeTimer(windowMaxTimestamp);
47.         }
49.     }
51.     public String toString() {
52.         return "EventTimeTrigger()";
53.     }
55.     public static EventTimeTrigger create() {
56.         return new EventTimeTrigger();
57.     }
58. }

复制代码

ProcessingTimeTrigger

1. //
2. // Source code recreated from a .class file by IntelliJ IDEA
3. // (powered by FernFlower decompiler)
4. //
6. package org.apache.flink.streaming.api.windowing.triggers;
8. import org.apache.flink.annotation.PublicEvolving;
9. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
11. @PublicEvolving
12. public class ProcessingTimeTrigger extends Trigger<Object, TimeWindow> {
13.     private static final long serialVersionUID = 1L;
15.     private ProcessingTimeTrigger() {
16.     }
18.     public TriggerResult onElement(Object element, long timestamp, TimeWindow window, Trigger.TriggerContext ctx) {
19.         ctx.registerProcessingTimeTimer(window.maxTimestamp());
20.         return TriggerResult.CONTINUE;
21.     }
23.     public TriggerResult onEventTime(long time, TimeWindow window, Trigger.TriggerContext ctx) throws Exception {
24.         return TriggerResult.CONTINUE;
25.     }
27.     public TriggerResult onProcessingTime(long time, TimeWindow window, Trigger.TriggerContext ctx) {
28.         return TriggerResult.FIRE;
29.     }
31.     public void clear(TimeWindow window, Trigger.TriggerContext ctx) throws Exception {
32.         ctx.deleteProcessingTimeTimer(window.maxTimestamp());
33.     }
35.     public boolean canMerge() {
36.         return true;
37.     }
39.     public void onMerge(TimeWindow window, Trigger.OnMergeContext ctx) {
40.         long windowMaxTimestamp = window.maxTimestamp();
41.         if (windowMaxTimestamp > ctx.getCurrentProcessingTime()) {
42.             ctx.registerProcessingTimeTimer(windowMaxTimestamp);
43.         }
45.     }
47.     public String toString() {
48.         return "ProcessingTimeTrigger()";
49.     }
51.     public static ProcessingTimeTrigger create() {
52.         return new ProcessingTimeTrigger();
53.     }
54. }

复制代码

ProcessingTimeoutTrigger

1. //
2. // Source code recreated from a .class file by IntelliJ IDEA
3. // (powered by FernFlower decompiler)
4. //
6. package org.apache.flink.streaming.api.windowing.triggers;
8. import java.time.Duration;
9. import org.apache.flink.annotation.PublicEvolving;
10. import org.apache.flink.api.common.state.ValueState;
11. import org.apache.flink.api.common.state.ValueStateDescriptor;
12. import org.apache.flink.api.common.typeutils.base.LongSerializer;
13. import org.apache.flink.streaming.api.windowing.windows.Window;
15. @PublicEvolving
16. public class ProcessingTimeoutTrigger<T, W extends Window> extends Trigger<T, W> {
17.     private static final long serialVersionUID = 1L;
18.     private final Trigger<T, W> nestedTrigger;
19.     private final long interval;
20.     private final boolean resetTimerOnNewRecord;
21.     private final boolean shouldClearOnTimeout;
22.     private final ValueStateDescriptor<Long> timeoutStateDesc;
24.     private ProcessingTimeoutTrigger(Trigger<T, W> nestedTrigger, long interval, boolean resetTimerOnNewRecord, boolean shouldClearOnTimeout) {
25.         this.nestedTrigger = nestedTrigger;
26.         this.interval = interval;
27.         this.resetTimerOnNewRecord = resetTimerOnNewRecord;
28.         this.shouldClearOnTimeout = shouldClearOnTimeout;
29.         this.timeoutStateDesc = new ValueStateDescriptor("timeout", LongSerializer.INSTANCE);
30.     }
32.     public TriggerResult onElement(T element, long timestamp, W window, Trigger.TriggerContext ctx) throws Exception {
33.         TriggerResult triggerResult = this.nestedTrigger.onElement(element, timestamp, window, ctx);
34.         if (triggerResult.isFire()) {
35.             this.clear(window, ctx);
36.             return triggerResult;
37.         } else {
38.             ValueState<Long> timeoutState = (ValueState)ctx.getPartitionedState(this.timeoutStateDesc);
39.             long nextFireTimestamp = ctx.getCurrentProcessingTime() + this.interval;
40.             Long timeoutTimestamp = (Long)timeoutState.value();
41.             if (timeoutTimestamp != null && this.resetTimerOnNewRecord) {
42.                 ctx.deleteProcessingTimeTimer(timeoutTimestamp);
43.                 timeoutState.clear();
44.                 timeoutTimestamp = null;
45.             }
47.             if (timeoutTimestamp == null) {
48.                 timeoutState.update(nextFireTimestamp);
49.                 ctx.registerProcessingTimeTimer(nextFireTimestamp);
50.             }
52.             return triggerResult;
53.         }
54.     }
56.     public TriggerResult onProcessingTime(long timestamp, W window, Trigger.TriggerContext ctx) throws Exception {
57.         TriggerResult triggerResult = this.nestedTrigger.onProcessingTime(timestamp, window, ctx);
58.         if (this.shouldClearOnTimeout) {
59.             this.clear(window, ctx);
60.         }
62.         return triggerResult.isPurge() ? TriggerResult.FIRE_AND_PURGE : TriggerResult.FIRE;
63.     }
65.     public TriggerResult onEventTime(long timestamp, W window, Trigger.TriggerContext ctx) throws Exception {
66.         TriggerResult triggerResult = this.nestedTrigger.onEventTime(timestamp, window, ctx);
67.         if (this.shouldClearOnTimeout) {
68.             this.clear(window, ctx);
69.         }
71.         return triggerResult.isPurge() ? TriggerResult.FIRE_AND_PURGE : TriggerResult.FIRE;
72.     }
74.     public void clear(W window, Trigger.TriggerContext ctx) throws Exception {
75.         ValueState<Long> timeoutTimestampState = (ValueState)ctx.getPartitionedState(this.timeoutStateDesc);
76.         Long timeoutTimestamp = (Long)timeoutTimestampState.value();
77.         if (timeoutTimestamp != null) {
78.             ctx.deleteProcessingTimeTimer(timeoutTimestamp);
79.             timeoutTimestampState.clear();
80.         }
82.         this.nestedTrigger.clear(window, ctx);
83.     }
85.     public String toString() {
86.         return "TimeoutTrigger(" + this.nestedTrigger.toString() + ")";
87.     }
89.     public static <T, W extends Window> ProcessingTimeoutTrigger<T, W> of(Trigger<T, W> nestedTrigger, Duration timeout) {
90.         return new ProcessingTimeoutTrigger(nestedTrigger, timeout.toMillis(), false, true);
91.     }
93.     public static <T, W extends Window> ProcessingTimeoutTrigger<T, W> of(Trigger<T, W> nestedTrigger, Duration timeout, boolean resetTimerOnNewRecord, boolean shouldClearOnTimeout) {
94.         return new ProcessingTimeoutTrigger(nestedTrigger, timeout.toMillis(), resetTimerOnNewRecord, shouldClearOnTimeout);
95.     }
96. }

复制代码

CountTrigger

1. //
2. // Source code recreated from a .class file by IntelliJ IDEA
3. // (powered by FernFlower decompiler)
4. //
6. package org.apache.flink.streaming.api.windowing.triggers;
8. import org.apache.flink.annotation.PublicEvolving;
9. import org.apache.flink.api.common.functions.ReduceFunction;
10. import org.apache.flink.api.common.state.ReducingState;
11. import org.apache.flink.api.common.state.ReducingStateDescriptor;
12. import org.apache.flink.api.common.typeutils.base.LongSerializer;
13. import org.apache.flink.streaming.api.windowing.windows.Window;
15. @PublicEvolving
16. public class CountTrigger<W extends Window> extends Trigger<Object, W> {
17.     private static final long serialVersionUID = 1L;
18.     private final long maxCount;
19.     private final ReducingStateDescriptor<Long> stateDesc;
21.     private CountTrigger(long maxCount) {
22.         this.stateDesc = new ReducingStateDescriptor("count", new Sum(), LongSerializer.INSTANCE);
23.         this.maxCount = maxCount;
24.     }
26.     public TriggerResult onElement(Object element, long timestamp, W window, Trigger.TriggerContext ctx) throws Exception {
27.         ReducingState<Long> count = (ReducingState)ctx.getPartitionedState(this.stateDesc);
28.         count.add(1L);
29.         if ((Long)count.get() >= this.maxCount) {
30.             count.clear();
31.             return TriggerResult.FIRE;
32.         } else {
33.             return TriggerResult.CONTINUE;
34.         }
35.     }
37.     public TriggerResult onEventTime(long time, W window, Trigger.TriggerContext ctx) {
38.         return TriggerResult.CONTINUE;
39.     }
41.     public TriggerResult onProcessingTime(long time, W window, Trigger.TriggerContext ctx) throws Exception {
42.         return TriggerResult.CONTINUE;
43.     }
45.     public void clear(W window, Trigger.TriggerContext ctx) throws Exception {
46.         ((ReducingState)ctx.getPartitionedState(this.stateDesc)).clear();
47.     }
49.     public boolean canMerge() {
50.         return true;
51.     }
53.     public void onMerge(W window, Trigger.OnMergeContext ctx) throws Exception {
54.         ctx.mergePartitionedState(this.stateDesc);
55.     }
57.     public String toString() {
58.         return "CountTrigger(" + this.maxCount + ")";
59.     }
61.     public static <W extends Window> CountTrigger<W> of(long maxCount) {
62.         return new CountTrigger(maxCount);
63.     }
65.     private static class Sum implements ReduceFunction<Long> {
66.         private static final long serialVersionUID = 1L;
68.         private Sum() {
69.         }
71.         public Long reduce(Long value1, Long value2) throws Exception {
72.             return value1 + value2;
73.         }
74.     }
75. }

复制代码

ContinuousEventTimeTrigger和ContinuousProcessingTimeTrigger

1. //
2. // Source code recreated from a .class file by IntelliJ IDEA
3. // (powered by FernFlower decompiler)
4. //
6. package org.apache.flink.streaming.api.windowing.triggers;
8. import org.apache.flink.annotation.PublicEvolving;
9. import org.apache.flink.annotation.VisibleForTesting;
10. import org.apache.flink.api.common.functions.ReduceFunction;
11. import org.apache.flink.api.common.state.ReducingState;
12. import org.apache.flink.api.common.state.ReducingStateDescriptor;
13. import org.apache.flink.api.common.typeutils.base.LongSerializer;
14. import org.apache.flink.streaming.api.windowing.time.Time;
15. import org.apache.flink.streaming.api.windowing.windows.Window;
17. @PublicEvolving
18. public class ContinuousEventTimeTrigger<W extends Window> extends Trigger<Object, W> {
19.     private static final long serialVersionUID = 1L;
20.     private final long interval;
21.     private final ReducingStateDescriptor<Long> stateDesc;
23.     private ContinuousEventTimeTrigger(long interval) {
24.         this.stateDesc = new ReducingStateDescriptor("fire-time", new Min(), LongSerializer.INSTANCE);
25.         this.interval = interval;
26.     }
28.     public TriggerResult onElement(Object element, long timestamp, W window, Trigger.TriggerContext ctx) throws Exception {
29.         if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
30.             return TriggerResult.FIRE;
31.         } else {
32.             ctx.registerEventTimeTimer(window.maxTimestamp());
33.             ReducingState<Long> fireTimestampState = (ReducingState)ctx.getPartitionedState(this.stateDesc);
34.             if (fireTimestampState.get() == null) {
35.                 this.registerNextFireTimestamp(timestamp - timestamp % this.interval, window, ctx, fireTimestampState);
36.             }
38.             return TriggerResult.CONTINUE;
39.         }
40.     }
42.     public TriggerResult onEventTime(long time, W window, Trigger.TriggerContext ctx) throws Exception {
43.         if (time == window.maxTimestamp()) {
44.             return TriggerResult.FIRE;
45.         } else {
46.             ReducingState<Long> fireTimestampState = (ReducingState)ctx.getPartitionedState(this.stateDesc);
47.             Long fireTimestamp = (Long)fireTimestampState.get();
48.             if (fireTimestamp != null && fireTimestamp == time) {
49.                 fireTimestampState.clear();
50.                 this.registerNextFireTimestamp(time, window, ctx, fireTimestampState);
51.                 return TriggerResult.FIRE;
52.             } else {
53.                 return TriggerResult.CONTINUE;
54.             }
55.         }
56.     }
58.     public TriggerResult onProcessingTime(long time, W window, Trigger.TriggerContext ctx) throws Exception {
59.         return TriggerResult.CONTINUE;
60.     }
62.     public void clear(W window, Trigger.TriggerContext ctx) throws Exception {
63.         ReducingState<Long> fireTimestamp = (ReducingState)ctx.getPartitionedState(this.stateDesc);
64.         Long timestamp = (Long)fireTimestamp.get();
65.         if (timestamp != null) {
66.             ctx.deleteEventTimeTimer(timestamp);
67.             fireTimestamp.clear();
68.         }
70.     }
72.     public boolean canMerge() {
73.         return true;
74.     }
76.     public void onMerge(W window, Trigger.OnMergeContext ctx) throws Exception {
77.         ctx.mergePartitionedState(this.stateDesc);
78.         Long nextFireTimestamp = (Long)((ReducingState)ctx.getPartitionedState(this.stateDesc)).get();
79.         if (nextFireTimestamp != null) {
80.             ctx.registerEventTimeTimer(nextFireTimestamp);
81.         }
83.     }
85.     public String toString() {
86.         return "ContinuousEventTimeTrigger(" + this.interval + ")";
87.     }
89.     @VisibleForTesting
90.     public long getInterval() {
91.         return this.interval;
92.     }
94.     public static <W extends Window> ContinuousEventTimeTrigger<W> of(Time interval) {
95.         return new ContinuousEventTimeTrigger(interval.toMilliseconds());
96.     }
98.     private void registerNextFireTimestamp(long time, W window, Trigger.TriggerContext ctx, ReducingState<Long> fireTimestampState) throws Exception {
99.         long nextFireTimestamp = Math.min(time + this.interval, window.maxTimestamp());
100.         fireTimestampState.add(nextFireTimestamp);
101.         ctx.registerEventTimeTimer(nextFireTimestamp);
102.     }
104.     private static class Min implements ReduceFunction<Long> {
105.         private static final long serialVersionUID = 1L;
107.         private Min() {
108.         }
110.         public Long reduce(Long value1, Long value2) throws Exception {
111.             return Math.min(value1, value2);
112.         }
113.     }
114. }

复制代码

1. //
2. // Source code recreated from a .class file by IntelliJ IDEA
3. // (powered by FernFlower decompiler)
4. //
6. package org.apache.flink.streaming.api.windowing.triggers;
8. import org.apache.flink.annotation.PublicEvolving;
9. import org.apache.flink.annotation.VisibleForTesting;
10. import org.apache.flink.api.common.functions.ReduceFunction;
11. import org.apache.flink.api.common.state.ReducingState;
12. import org.apache.flink.api.common.state.ReducingStateDescriptor;
13. import org.apache.flink.api.common.typeutils.base.LongSerializer;
14. import org.apache.flink.streaming.api.windowing.time.Time;
15. import org.apache.flink.streaming.api.windowing.windows.Window;
17. @PublicEvolving
18. public class ContinuousProcessingTimeTrigger<W extends Window> extends Trigger<Object, W> {
19.     private static final long serialVersionUID = 1L;
20.     private final long interval;
21.     private final ReducingStateDescriptor<Long> stateDesc;
23.     private ContinuousProcessingTimeTrigger(long interval) {
24.         this.stateDesc = new ReducingStateDescriptor("fire-time", new Min(), LongSerializer.INSTANCE);
25.         this.interval = interval;
26.     }
28.     public TriggerResult onElement(Object element, long timestamp, W window, Trigger.TriggerContext ctx) throws Exception {
29.         ReducingState<Long> fireTimestampState = (ReducingState)ctx.getPartitionedState(this.stateDesc);
30.         timestamp = ctx.getCurrentProcessingTime();
31.         if (fireTimestampState.get() == null) {
32.             this.registerNextFireTimestamp(timestamp - timestamp % this.interval, window, ctx, fireTimestampState);
33.         }
35.         return TriggerResult.CONTINUE;
36.     }
38.     public TriggerResult onEventTime(long time, W window, Trigger.TriggerContext ctx) throws Exception {
39.         return TriggerResult.CONTINUE;
40.     }
42.     public TriggerResult onProcessingTime(long time, W window, Trigger.TriggerContext ctx) throws Exception {
43.         ReducingState<Long> fireTimestampState = (ReducingState)ctx.getPartitionedState(this.stateDesc);
44.         if (((Long)fireTimestampState.get()).equals(time)) {
45.             fireTimestampState.clear();
46.             this.registerNextFireTimestamp(time, window, ctx, fireTimestampState);
47.             return TriggerResult.FIRE;
48.         } else {
49.             return TriggerResult.CONTINUE;
50.         }
51.     }
53.     public void clear(W window, Trigger.TriggerContext ctx) throws Exception {
54.         ReducingState<Long> fireTimestamp = (ReducingState)ctx.getPartitionedState(this.stateDesc);
55.         Long timestamp = (Long)fireTimestamp.get();
56.         if (timestamp != null) {
57.             ctx.deleteProcessingTimeTimer(timestamp);
58.             fireTimestamp.clear();
59.         }
61.     }
63.     public boolean canMerge() {
64.         return true;
65.     }
67.     public void onMerge(W window, Trigger.OnMergeContext ctx) throws Exception {
68.         ctx.mergePartitionedState(this.stateDesc);
69.         Long nextFireTimestamp = (Long)((ReducingState)ctx.getPartitionedState(this.stateDesc)).get();
70.         if (nextFireTimestamp != null) {
71.             ctx.registerProcessingTimeTimer(nextFireTimestamp);
72.         }
74.     }
76.     @VisibleForTesting
77.     public long getInterval() {
78.         return this.interval;
79.     }
81.     public String toString() {
82.         return "ContinuousProcessingTimeTrigger(" + this.interval + ")";
83.     }
85.     public static <W extends Window> ContinuousProcessingTimeTrigger<W> of(Time interval) {
86.         return new ContinuousProcessingTimeTrigger(interval.toMilliseconds());
87.     }
89.     private void registerNextFireTimestamp(long time, W window, Trigger.TriggerContext ctx, ReducingState<Long> fireTimestampState) throws Exception {
90.         long nextFireTimestamp = Math.min(time + this.interval, window.maxTimestamp());
91.         fireTimestampState.add(nextFireTimestamp);
92.         ctx.registerProcessingTimeTimer(nextFireTimestamp);
93.     }
95.     private static class Min implements ReduceFunction<Long> {
96.         private static final long serialVersionUID = 1L;
98.         private Min() {
99.         }
101.         public Long reduce(Long value1, Long value2) throws Exception {
102.             return Math.min(value1, value2);
103.         }
104.     }
105. }

复制代码

DeltaTrigger

1. //
2. // Source code recreated from a .class file by IntelliJ IDEA
3. // (powered by FernFlower decompiler)
4. //
6. package org.apache.flink.streaming.api.windowing.triggers;
8. import org.apache.flink.annotation.PublicEvolving;
9. import org.apache.flink.api.common.state.ValueState;
10. import org.apache.flink.api.common.state.ValueStateDescriptor;
11. import org.apache.flink.api.common.typeutils.TypeSerializer;
12. import org.apache.flink.streaming.api.functions.windowing.delta.DeltaFunction;
13. import org.apache.flink.streaming.api.windowing.windows.Window;
15. @PublicEvolving
16. public class DeltaTrigger<T, W extends Window> extends Trigger<T, W> {
17.     private static final long serialVersionUID = 1L;
18.     private final DeltaFunction<T> deltaFunction;
19.     private final double threshold;
20.     private final ValueStateDescriptor<T> stateDesc;
22.     private DeltaTrigger(double threshold, DeltaFunction<T> deltaFunction, TypeSerializer<T> stateSerializer) {
23.         this.deltaFunction = deltaFunction;
24.         this.threshold = threshold;
25.         this.stateDesc = new ValueStateDescriptor("last-element", stateSerializer);
26.     }
28.     public TriggerResult onElement(T element, long timestamp, W window, Trigger.TriggerContext ctx) throws Exception {
29.         ValueState<T> lastElementState = (ValueState)ctx.getPartitionedState(this.stateDesc);
30.         if (lastElementState.value() == null) {
31.             lastElementState.update(element);
32.             return TriggerResult.CONTINUE;
33.         } else if (this.deltaFunction.getDelta(lastElementState.value(), element) > this.threshold) {
34.             lastElementState.update(element);
35.             return TriggerResult.FIRE;
36.         } else {
37.             return TriggerResult.CONTINUE;
38.         }
39.     }
41.     public TriggerResult onEventTime(long time, W window, Trigger.TriggerContext ctx) {
42.         return TriggerResult.CONTINUE;
43.     }
45.     public TriggerResult onProcessingTime(long time, W window, Trigger.TriggerContext ctx) throws Exception {
46.         return TriggerResult.CONTINUE;
47.     }
49.     public void clear(W window, Trigger.TriggerContext ctx) throws Exception {
50.         ((ValueState)ctx.getPartitionedState(this.stateDesc)).clear();
51.     }
53.     public String toString() {
54.         return "DeltaTrigger(" + this.deltaFunction + ", " + this.threshold + ")";
55.     }
57.     public static <T, W extends Window> DeltaTrigger<T, W> of(double threshold, DeltaFunction<T> deltaFunction, TypeSerializer<T> stateSerializer) {
58.         return new DeltaTrigger(threshold, deltaFunction, stateSerializer);
59.     }
60. }

复制代码

PurgingTrigger

1. //
2. // Source code recreated from a .class file by IntelliJ IDEA
3. // (powered by FernFlower decompiler)
4. //
6. package org.apache.flink.streaming.api.windowing.triggers;
8. import org.apache.flink.annotation.PublicEvolving;
9. import org.apache.flink.annotation.VisibleForTesting;
10. import org.apache.flink.streaming.api.windowing.windows.Window;
12. @PublicEvolving
13. public class PurgingTrigger<T, W extends Window> extends Trigger<T, W> {
14.     private static final long serialVersionUID = 1L;
15.     private Trigger<T, W> nestedTrigger;
17.     private PurgingTrigger(Trigger<T, W> nestedTrigger) {
18.         this.nestedTrigger = nestedTrigger;
19.     }
21.     public TriggerResult onElement(T element, long timestamp, W window, Trigger.TriggerContext ctx) throws Exception {
22.         TriggerResult triggerResult = this.nestedTrigger.onElement(element, timestamp, window, ctx);
23.         return triggerResult.isFire() ? TriggerResult.FIRE_AND_PURGE : triggerResult;
24.     }
26.     public TriggerResult onEventTime(long time, W window, Trigger.TriggerContext ctx) throws Exception {
27.         TriggerResult triggerResult = this.nestedTrigger.onEventTime(time, window, ctx);
28.         return triggerResult.isFire() ? TriggerResult.FIRE_AND_PURGE : triggerResult;
29.     }
31.     public TriggerResult onProcessingTime(long time, W window, Trigger.TriggerContext ctx) throws Exception {
32.         TriggerResult triggerResult = this.nestedTrigger.onProcessingTime(time, window, ctx);
33.         return triggerResult.isFire() ? TriggerResult.FIRE_AND_PURGE : triggerResult;
34.     }
36.     public void clear(W window, Trigger.TriggerContext ctx) throws Exception {
37.         this.nestedTrigger.clear(window, ctx);
38.     }
40.     public boolean canMerge() {
41.         return this.nestedTrigger.canMerge();
42.     }
44.     public void onMerge(W window, Trigger.OnMergeContext ctx) throws Exception {
45.         this.nestedTrigger.onMerge(window, ctx);
46.     }
48.     public String toString() {
49.         return "PurgingTrigger(" + this.nestedTrigger.toString() + ")";
50.     }
52.     public static <T, W extends Window> PurgingTrigger<T, W> of(Trigger<T, W> nestedTrigger) {
53.         return new PurgingTrigger(nestedTrigger);
54.     }
56.     @VisibleForTesting
57.     public Trigger<T, W> getNestedTrigger() {
58.         return this.nestedTrigger;
59.     }
60. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4