flink-触发器Trigger和移除器Evictor

窗口原理与机制

图片链接：https://blog.csdn.net/qq_35590459/article/details/132177154

1. 数据流进入算子前，被提交给WindowAssigner，决定元素被放到哪个或哪些窗口，同时可能会创建新窗口或者合并旧的窗口。
2. 每一个窗口都拥有一个属于自己的触发器Trigger，每当有元素被分配到该窗口，或者之前注册的定时器超时时，Trigger都会被调用。
3. Trigger被触发后，窗口中的元素集合就会交给Evictor（如果指定了），遍历窗口中的元素列表，并决定最先进入窗口的多少个元素需要被移除。
4. 窗口函数计算结果值，发送给下游；

Trigger 触发器

触发器作用：控制窗口什么时候除法计算。即执行窗口函数；基于WindowStream调用trigger（）方法，传入自定义触发器（trigger）；

每一个窗口分配器（windowAssigner） 都会对应一个默认的触发器；

 源码样例

  @PublicEvolvingpublic <W extends Window> WindowedStream<T, KEY, W> window(WindowAssigner<? super T, W> assigner) {return new WindowedStream<>(this, assigner);}@PublicEvolvingpublic WindowedStream(KeyedStream<T, K> input, WindowAssigner<? super T, W> windowAssigner) {this.input = input;this.builder =new WindowOperatorBuilder<>(windowAssigner,windowAssigner.getDefaultTrigger(input.getExecutionEnvironment()),input.getExecutionConfig(),input.getType(),input.getKeySelector(),input.getKeyType());}==============默认触发器===
public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {return EventTimeTrigger.create();}

Triger类有4个方法

1.  onElement:窗口中每来一个元素调用该方法。
   onProcessingTime：当注册的处理时间定时器触发时，将调用这个方法。onEventTime：当注时的事件时间定时器触发时，将调用这个方法。clear：窗口关闭冰销毁时调用这个方法，一般用来清除自定义状态。onElement() ，onProcessingTime()，onEventTime()方法的返回类型都是 TriggerResult；TriggerResult中包含四个枚举值：
   CONTINUE：表示对窗口不执行任何操作。
   FIRE：触发计算并输出结果。注意计算完成后，窗口中的数据并不会被清除，将会被保留。
   PURGE：表示将窗口中的数据和窗口清除。
   FIRE_AND_PURGE：表示先将数据进行计算，输出结果，然后将窗口中的数据和窗口进行清除。
   

源码

/** No action is taken on the window. */
CONTINUE(false, false),
/** {@code FIRE_AND_PURGE} evaluates the window function and emits the window result. */
FIRE_AND_PURGE(true, true),
/*** On {@code FIRE}, the window is evaluated and results are emitted. The window is not purged,* though, all elements are retained.*/
FIRE(true, false),
/*** All elements in the window are cleared and the window is discarded, without evaluating the* window function or emitting any elements.*/
PURGE(false, true);

flink提供的触发器 

flink提供触发器

• EventTimeTrigger：通过对比EventTime和窗口的Endtime确定是否触发窗口计算，如果EventTime大于Window EndTime则触发，否则不触发，窗口将继续等待。
• ProcessingTimeoutTrigger：当内置触发器满足设置的超时时间时，触发窗口的计算。
• ProcessTimeTrigger：通过对比ProcessTime和窗口EndTme确定是否触发窗口，如果ProcessTime大于EndTime则触发计算，否则窗口继续等待。
• ContinuousEventTimeTrigger：根据间隔时间周期性触发窗口或者Window的结束时间小于当前EndTime触发窗口计算。
• ContinuousProcessingTimeTrigger：根据间隔时间周期性触发窗口或者Window的结束时间小于当前ProcessTime触发窗口计算。
• CountTrigger：根据接入数据量是否超过设定的阙值判断是否触发窗口计算。
• DeltaTrigger：根据接入数据计算出来的Delta指标是否超过指定的Threshold去判断是否触发窗口计算。
• PurgingTrigger：可以将任意触发器作为参数转换为Purge类型的触发器，计算完成后数据将被清理。
• NeverTrigger：任何时候都不触发窗口计算，全局窗口触发器，

原文链接：https://blog.csdn.net/qq_37555071/article/details/122514061

水印触发一般是窗口关闭时间

flink提供的触发器是与窗口对应，当有水印时，如果水印时间大于等于窗口结束时间会触发计算；window.maxTimestamp()获取的是窗口end-1; EventTimeTrigger 的源码可以很明确可以看到注册时注册了触发时间为window.maxTimestamp(),这也是窗口关闭的触发机制。

如果在窗口关闭前触发计算设置多个触发计算时间，这样实现一些特定的需求。例如，每10s输出一次当天的累计数据；

public class EventTimeTrigger extends Trigger<Object, TimeWindow> {private static final long serialVersionUID = 1L;private EventTimeTrigger() {}@Overridepublic TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx)throws Exception {if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {// if the watermark is already past the window fire immediatelyreturn TriggerResult.FIRE;} else {ctx.registerEventTimeTimer(window.maxTimestamp());return TriggerResult.CONTINUE;}}@Overridepublic TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) {// 限定触发条件为窗口关闭时间，否则就继续窗口 return time == window.maxTimestamp() ? TriggerResult.FIRE : TriggerResult.CONTINUE;}@Overridepublic TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx)throws Exception {return TriggerResult.CONTINUE;}
.....

自定义触发器

继承Triger，重写抽象方法，案例

.window(TumblingEventTimeWindows.of(Time.hours(24))).trigger(new MyTrigger()).process(new WindowResult()).print();窗口长24小时，每十秒触发一次计算
===================public static class MyTrigger extends Trigger<Event, TimeWindow> {@Overridepublic TriggerResult onElement(Event event, long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {//定义状态，记录该状态 触发器第一个元素进来时注册全部的触发器ValueState<Boolean> isFirstEvent = triggerContext.getPartitionedState(new ValueStateDescriptor<Boolean>("first-event", Types.BOOLEAN));//第一次注册，右面全部跳过if (isFirstEvent.value() == null) {for (long i = timeWindow.getStart(); i < timeWindow.getEnd(); i = i + 10000L) {//注册触发器  间隔10striggerContext.registerEventTimeTimer(i);}isFirstEvent.update(true);}return TriggerResult.CONTINUE;}@Overridepublic TriggerResult onEventTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {//使用的事件时间，因此触发窗口的计算return TriggerResult.FIRE;}@Overridepublic TriggerResult onProcessingTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {return TriggerResult.CONTINUE;}@Overridepublic void clear(TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {ValueState<Boolean> isFirstEvent = triggerContext.getPartitionedState(new ValueStateDescriptor<Boolean>("first-event", Types.BOOLEAN));isFirstEvent.clear();}}

移除器Evictor

作用：主要用来定义移除某些数据的逻辑。基于windowedStream调用evictor()方法，就可以传入一个自定义得移除器(Evictor)。不同窗口类型都有各自预测实现的移除器。

stream.keyby().window().evictor(new MyEvictor)

evictBefore():定义窗口执行函数之前移除的数据操作，移除后的数据不参与窗口计算；

evictAfter():定义执行窗口函数后移除数据的操作；

默认情况下预实现的移出弃都是在执行窗口函数之前移除数据

flink 提供的移除器

CountEvictor: 仅记录用户指定数量的元素，一旦窗口中的元素超过这个数量，多余的元素会从窗口缓存的开头移除； CountEvictor在countwindow中有明确定义引用。
DeltaEvictor: 接收 DeltaFunction 和 threshold 参数，计算最后一个元素与窗口缓存中所有元素的差值， 并移除差值大于或等于 threshold 的元素。（暂时不清楚作用）
TimeEvictor:  接受窗口inteval时间，它会找到窗口中元素的最大 timestamp max_ts 并移除比 max_ts - inteval小的所有元素。TimeEvictor.of() 方法来构建； inteval 不是窗口时间，如果为0，窗口没有数据输出

//TimeEvictor  部分源码 @Overridepublic void evictBefore(Iterable<TimestampedValue<Object>> elements, int size, W window, EvictorContext ctx) {if (!doEvictAfter) {evict(elements, size, ctx);}}@Overridepublic void evictAfter(Iterable<TimestampedValue<Object>> elements, int size, W window, EvictorContext ctx) {if (doEvictAfter) {evict(elements, size, ctx);}}private void evict(Iterable<TimestampedValue<Object>> elements, int size, EvictorContext ctx) {if (!hasTimestamp(elements)) {return;}long currentTime = getMaxTimestamp(elements);long evictCutoff = currentTime - windowSize;//移除时间窗口时间之前的数据，注意：获取的并不是窗口end时间for (Iterator<TimestampedValue<Object>> iterator = elements.iterator();iterator.hasNext(); ) {TimestampedValue<Object> record = iterator.next();if (record.getTimestamp() <= evictCutoff) {iterator.remove();}}}
// 获取当前元素中最大的时间private long getMaxTimestamp(Iterable<TimestampedValue<Object>> elements) {long currentTime = Long.MIN_VALUE;for (Iterator<TimestampedValue<Object>> iterator = elements.iterator();iterator.hasNext(); ) {TimestampedValue<Object> record = iterator.next();currentTime = Math.max(currentTime, record.getTimestamp());}return currentTime;}// 保留多长时间的数据public static <W extends Window> TimeEvictor<W> of(Time windowSize) {return new TimeEvictor<>(windowSize.toMilliseconds());}/*** Creates a {@code TimeEvictor} that keeps the given number of elements. Eviction is done* before/after the window function based on the value of doEvictAfter.** @param windowSize The amount of time for which to keep elements.* @param doEvictAfter Whether eviction is done after window function.*/public static <W extends Window> TimeEvictor<W> of(Time windowSize, boolean doEvictAfter) {return new TimeEvictor<>(windowSize.toMilliseconds(), doEvictAfter);}

例如
stream.keyBy(r -> r.user)
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.evictor(TimeEvictor.of(Time.seconds(3)))  // 只输出窗口关闭前3s的数据
.process( new WindowResult())
.print();

注意：如果在evict中使用了iterable.iterator(),后面再次使用时不能直接使用

 .keyBy(r -> r.user).window(TumblingEventTimeWindows.of(Time.seconds(10)));window.evictor(new Evictor<Event, TimeWindow>() {@Overridepublic void evictBefore(Iterable<TimestampedValue<Event>> elements, int size, TimeWindow window, EvictorContext evictorContext) {Iterator<TimestampedValue<Event>> iterator = elements.iterator();while (iterator.hasNext()){TimestampedValue<Event> next = iterator.next();if(next.getValue().url.equals("./prod?id=1")){iterator.remove();}}}@Overridepublic void evictAfter(Iterable<TimestampedValue<Event>> elements, int size, TimeWindow window, EvictorContext evictorContext) {return;}}).process(new ProcessWindowFunction<Event, UrlViewCount, String, TimeWindow>() {@Overridepublic void process(String s, ProcessWindowFunction<Event, UrlViewCount, String, TimeWindow>.Context context, Iterable<Event> elements, Collector<UrlViewCount> out) throws Exception {AtomicInteger i= new AtomicInteger();elements.forEach(v-> i.getAndIncrement());out.collect(new UrlViewCount(s+"====",// 获取迭代器中的元素个数  不能再使用iterable.spliterator().getExactSizeIfKnown(),否侧获取数据一一直为-1i.longValue(),context.window().getStart(),context.window().getEnd()));} }).print();