Flink 窗口触发器(Trigger)(一)

王國慶  金牌会员 | 2024-8-28 02:09:51 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 572|帖子 572|积分 1720

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企服之家，中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看，没有账号？立即注册

x
回复

使用道具 举报

0 个回复

正序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

王國慶

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表