Flink 自定义触发器

缺乏、安全感 2021-11-18 01:48 574阅读 0赞

trigger接口有五个方法允许trigger对不同的事件做出反应：

onElement():进入窗口的每个元素都会调用该方法。

onEventTime():事件时间timer触发的时候被调用。

onProcessingTime():处理时间timer触发的时候会被调用。

onMerge():有状态的触发器相关，并在它们相应的窗口合并时合并两个触发器的状态，例如使用会话窗口。

clear():该方法主要是执行窗口的删除操作。

关于上述方法需要注意两点：

1).前三方法决定着如何通过返回一个TriggerResult来操作输入事件。

*  CONTINUE:什么都不做。
 *  FIRE:触发计算。
 *  PURE:清除窗口的元素。
 *  FIRE\_AND\_PURE:触发计算和清除窗口元素。

2).这些方法中的任何一个都可用于为将来的操作注册处理或事件时间计时器

内置和自定义触发器

Flink内部有一些内置的触发器:

*  EventTimeTrigger：基于事件时间和watermark机制来对窗口进行触发计算。
 *  ProcessingTimeTrigger:基于处理时间触发。
 *  CountTrigger:窗口元素数超过预先给定的限制值的话会触发计算。
 *  PurgingTrigger作为其它trigger的参数，将其转化为一个purging触发器。

针对实际开发中的业务场景我们需要根据不同的业务去触发计算，这个时候我们就需要自己手动去定义触发器：

下面是转载的简单的案例，已经够大家去理解了：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

private static Logger LOG = LoggerFactory.getLogger(CountTriggerWithTimeout.class);
    
        /**
         * 窗口最大数据量
         */
        private int maxCount;
        /**
         * event time / process time
         */
        private TimeCharacteristic timeType;
        /**
         * 用于储存窗口当前数据量的状态对象
         */
        private ReducingStateDescriptor<Long> countStateDescriptor =
                new ReducingStateDescriptor("counter", new Sum(), LongSerializer.INSTANCE);
    
    
        public CountTriggerWithTimeout(int maxCount, TimeCharacteristic timeType) {
    
            this.maxCount = maxCount;
            this.timeType = timeType;
        }
    
    
        private TriggerResult fireAndPurge(TimeWindow window, TriggerContext ctx) throws Exception {
            clear(window, ctx);
            return TriggerResult.FIRE_AND_PURGE;
        }
    
    
        @Override
        public TriggerResult onElement(T element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
            ReducingState<Long> countState = ctx.getPartitionedState(countStateDescriptor);
            countState.add(1L);
    
            if (countState.get() >= maxCount) {
                LOG.info("fire with count: " + countState.get());
                return fireAndPurge(window, ctx);
            }
            if (timestamp >= window.getEnd()) {
                LOG.info("fire with tiem: " + timestamp);
                return fireAndPurge(window, ctx);
            } else {
                return TriggerResult.CONTINUE;
            }
        }
    
        @Override
        public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            if (timeType != TimeCharacteristic.ProcessingTime) {
                return TriggerResult.CONTINUE;
            }
    
            if (time >= window.getEnd()) {
                return TriggerResult.CONTINUE;
            } else {
                LOG.info("fire with process tiem: " + time);
                return fireAndPurge(window, ctx);
            }
        }
    
        @Override
        public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            if (timeType != TimeCharacteristic.EventTime) {
                return TriggerResult.CONTINUE;
            }
    
            if (time >= window.getEnd()) {
                return TriggerResult.CONTINUE;
            } else {
                LOG.info("fire with event tiem: " + time);
                return fireAndPurge(window, ctx);
            }
        }
    
        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
            ReducingState<Long> countState = ctx.getPartitionedState(countStateDescriptor);
            countState.clear();
        }
    
        /**
         * 计数方法
         */
        class Sum implements ReduceFunction<Long> {
    
            @Override
            public Long reduce(Long value1, Long value2) throws Exception {
                return value1 + value2;
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20211116/891fa59330dd4a849272898f1d98e3e7.png