flink-触发器Trigger和移除器Evictor

Trigger 触发器

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

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

源码样例

  
  @PublicEvolving
    public <W extends Window> WindowedStream<T, KEY, W> window(
            WindowAssigner<? super T, W> assigner) {
        return new WindowedStream<>(this, assigner);
    }
  

  @PublicEvolving
    public 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个方法

 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提供触发器

ProcessingTimeoutTrigger

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() {}

    @Override
    public 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 immediately
            return TriggerResult.FIRE;
        } else {
            ctx.registerEventTimeTimer(window.maxTimestamp());
            return TriggerResult.CONTINUE;
        }
    }

    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) {
       // 限定触发条件为窗口关闭时间，否则就继续窗口 
       return time == window.maxTimestamp() ? TriggerResult.FIRE : TriggerResult.CONTINUE;
    }

    @Override
    public 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> {
        @Override
        public 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) {
                    //注册触发器  间隔10s
                    triggerContext.registerEventTimeTimer(i);
                }
                isFirstEvent.update(true);
            }
            return TriggerResult.CONTINUE;
        }

        @Override
        public TriggerResult onEventTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
           //使用的事件时间，因此触发窗口的计算
            return TriggerResult.FIRE;
        }

        @Override
        public TriggerResult onProcessingTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
            return TriggerResult.CONTINUE;
        }

        @Override
        public 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():定义执行窗口函数后移除数据的操作；

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

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

 .keyBy(r -> r.user)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)));
        window.evictor(new Evictor<Event, TimeWindow>() {
                    @Override
                    public 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();
                            }
                        }
                    }

                    @Override
                    public void evictAfter(Iterable<TimestampedValue<Event>> elements, int size, TimeWindow window, EvictorContext evictorContext) {
                        return;
                    }
                })
                .process(new ProcessWindowFunction<Event, UrlViewCount, String, TimeWindow>() {
                    @Override
                    public 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(),否侧获取数据一一直为-1
                                        i.longValue(),
                                        context.window().getStart(),
                                        context.window().getEnd()
                                ));
                    } })
                .print();