聊聊flink的Triggers

本文主要探讨flink中的triggers。

Trigger类接受两个泛型参数，一个是元素类型，另一个是窗口类型。它定义了onElement、onProcessingTime、onEventTime、canMerge、onMerge、clear等方法，其中onElement、onProcessingTime、onEventTime方法均需返回TriggerResult。onElement方法在每个元素被添加到窗口时被调用；onProcessingTime方法在注册的处理时间计时器触发时被调用；onEventTime方法在注册的事件时间计时器触发时被调用。canMerge方法用于指示是否支持触发器状态的合并，默认返回false；onMerge方法在多个窗口合并时被触发；clear方法用于清除TriggerContext中存储的相关状态。Trigger还定义了TriggerContext和OnMergeContext接口。TriggerContext定义了注册和删除事件时间和处理时间计时器的方法，同时还定义了getCurrentProcessingTime、getMetricGroup、getCurrentWatermark、getPartitionedState、getKeyValueState等方法。OnMergeContext继承自TriggerContext，并额外定义了mergePartitionedState方法。

TriggerResult枚举用于表示触发器在onElement、onProcessingTime、onEventTime方法被调用时返回的动作。它有fire和purge两个属性，包含CONTINUE、FIRE_AND_PURGE、FIRE、PURGE四个枚举值。fire属性指示是否需要触发窗口的计算操作，而purge属性指示是否需要清理窗口的数据。CONTINUE表示不对窗口进行任何操作；FIRE_AND_PURGE表示触发窗口的计算操作并清理窗口数据；FIRE表示仅触发窗口的计算操作而不清理窗口数据；PURGE表示不触发窗口的计算操作但清理窗口数据。

EventTimeTrigger继承自Trigger，元素类型为Object，窗口类型为TimeWindow。它被SlidingEventTimeWindows、TumblingEventTimeWindows、EventTimeSessionWindows和DynamicEventTimeSessionWindows默认使用。onElement方法在window.maxTimestamp()小于或等于ctx.getCurrentWatermark()时返回TriggerResult.FIRE，否则执行ctx.registerEventTimeTimer(window.maxTimestamp())并返回TriggerResult.CONTINUE。onEventTime方法在time等于window.maxTimestamp()时返回TriggerResult.FIRE，否则返回TriggerResult.CONTINUE。onProcessingTime方法返回TriggerResult.CONTINUE。canMerge方法返回true。onMerge方法在window.maxTimestamp()大于ctx.getCurrentWatermark()时执行ctx.registerEventTimeTimer(window.maxTimestamp())。clear方法执行ctx.deleteEventTimeTimer(window.maxTimestamp())。

ProcessingTimeTrigger同样继承自Trigger，元素类型为Object，窗口类型为TimeWindow。它被SlidingProcessingTimeWindows、TumblingProcessingTimeWindows、ProcessingTimeSessionWindows和DynamicProcessingTimeSessionWindows默认使用。onElement方法执行ctx.registerProcessingTimeTimer(window.maxTimestamp())并返回TriggerResult.CONTINUE。onEventTime方法返回TriggerResult.CONTINUE。onProcessingTime方法返回TriggerResult.FIRE。canMerge方法返回true。onMerge方法在window.maxTimestamp()大于ctx.getCurrentProcessingTime()时执行ctx.registerProcessingTimeTimer(window.maxTimestamp())。clear方法执行ctx.deleteProcessingTimeTimer(window.maxTimestamp())。

NeverTrigger的onElement、onEventTime、onProcessingTime方法均返回TriggerResult.CONTINUE。GlobalWindows默认使用NeverTrigger。

CountTrigger继承自Trigger，元素类型为Object。它定义了maxCount和ReducingStateDescriptor，用于窗口计数。在onElement方法中，当计数大于或等于maxCount时，清空计数并返回TriggerResult.FIRE，否则返回TriggerResult.CONTINUE。onEventTime和onProcessingTime方法均返回TriggerResult.CONTINUE。canMerge方法返回true。onMerge方法执行ctx.mergePartitionedState(stateDesc)。clear方法执行ctx.getPartitionedState(stateDesc).clear()。

PurgingTrigger是一种包装类型的Trigger，它包装了nestedTrigger。在onElement、onEventTime、onProcessingTime方法中，如果nestedTrigger的返回结果为fire，则包装返回TriggerResult.FIRE_AND_PURGE。canMerge、onMerge、clear等方法均委托给nestedTrigger处理。

总结：Trigger接收两个泛型参数，一个是元素类型，一个是窗口类型。它定义了onElement、onProcessingTime、onEventTime、canMerge、onMerge、clear等方法，其中onElement、onProcessingTime、onEventTime均需返回TriggerResult。TriggerResult用于表示触发器在onElement、onProcessingTime、onEventTime方法被调用时返回的动作，包含fire和purge两个属性，及CONTINUE、FIRE_AND_PURGE、FIRE、PURGE四个枚举值。SlidingEventTimeWindows、TumblingEventTimeWindows、EventTimeSessionWindows和DynamicEventTimeSessionWindows默认使用EventTimeTrigger；SlidingProcessingTimeWindows、TumblingProcessingTimeWindows、ProcessingTimeSessionWindows和DynamicProcessingTimeSessionWindows默认使用ProcessingTimeTrigger；GlobalWindows默认使用NeverTrigger。CountTrigger主要用于计数类型的窗口，使用ReducingStateDescriptor进行窗口计数。PurgingTrigger是一种包装类型的Trigger，根据nestedTrigger的返回结果在需要时包装返回TriggerResult.FIRE_AND_PURGE。

docTriggers

以上就是聊聊flink的Triggers的详细内容，更多请关注深圳推广【www.sztg.com.cn】。