拿来即用 | Flink 触发器实战！

2022-12-12 02:01:00  sum1
2022-12-12 02:02:00  sum2
2022-12-12 02:03:00  sum3
2022-12-12 02:04:00  sum4
......

public static class MyTrigger<W extends Window> extends Trigger<Object , W>{




        private final long interval;
        private final long maxCount;


        private ReducingStateDescriptor<Long> intervalDesc = new ReducingStateDescriptor<>("fire-time", new Min(), LongSerializer.INSTANCE);
        private ReducingStateDescriptor<Long> maxCountDesc = new ReducingStateDescriptor<>("sum", new Sum(), LongSerializer.INSTANCE);


        private MyTrigger(long interval, long maxCount){
            this.interval = interval;
            this.maxCount = maxCount;
        }


        public static <W extends Window>  MyTrigger<W> of(Time interval, long maxCount){
            return new MyTrigger<>(interval.toMilliseconds() -1, maxCount);
        }


        @Override
        public TriggerResult onElement(Object o, long l, W w, TriggerContext triggerContext) throws Exception {
            if (w.maxTimestamp() <= triggerContext.getCurrentWatermark()){
                return TriggerResult.FIRE_AND_PURGE;
            } else {
                triggerContext.registerProcessingTimeTimer(w.maxTimestamp());
            }


            ReducingState<Long> intervalState = triggerContext.getPartitionedState(intervalDesc);
            ReducingState<Long> maxCountState = triggerContext.getPartitionedState(maxCountDesc);


            maxCountState.add(1L);
            if (intervalState.get() == null){
                registerNextTimer(l, w, triggerContext,intervalState);
            }
            if (maxCountState.get() >= maxCount){
                maxCountState.clear();
                System.err.println("maxCount到达阈值，触发计算");
                return TriggerResult.FIRE;
            }
            return TriggerResult.CONTINUE;
        }


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


        @Override
        public TriggerResult onEventTime(long l, W w, TriggerContext triggerContext) throws Exception {
            if (l == w.maxTimestamp()){
                return TriggerResult.FIRE_AND_PURGE;
            }
            ReducingState<Long> maxCountState  = triggerContext.getPartitionedState(maxCountDesc);
            Long count = maxCountState .get();
            ReducingState<Long> intervalState = triggerContext.getPartitionedState(intervalDesc);
            Long interval = intervalState.get();
            if (interval != null && interval == l){
                intervalState.clear();
                registerNextTimer(l, w, triggerContext, intervalState);
                if (count != null){
                    System.err.println("触发定时器："+ new Timestamp(l) + "");
                    return TriggerResult.FIRE;
                }
            }
            return TriggerResult.CONTINUE;
        }


        @Override
        public void clear(W w, TriggerContext triggerContext) throws Exception {
            triggerContext.getPartitionedState(maxCountDesc).clear();


            ReducingState<Long> intervalState = triggerContext.getPartitionedState(intervalDesc);
            Long timer = intervalState.get();
            if (timer != null){
                triggerContext.deleteProcessingTimeTimer(timer);
                intervalState.clear();
            }


        }


        public boolean canMerge(){
            return true;
        }


        public void onMerge(W window, OnMergeContext ctx) throws Exception {
            // 合并数量间隔的状态
            ctx.mergePartitionedState(maxCountDesc);
            // 合并时间间隔的状态
            ctx.mergePartitionedState(intervalDesc);
            Long interval = ctx.getPartitionedState(intervalDesc).get();
            if (interval != null) {
                ctx.registerEventTimeTimer(interval);
            }
        }


        private static class Min implements ReduceFunction<Long> {


            @Override
            public Long reduce(Long aLong, Long t1) throws Exception {
                return Math.min(aLong, t1);
            }
        }


        private static class Sum implements ReduceFunction<Long>{


            @Override
            public Long reduce(Long aLong, Long t1) throws Exception {
                return aLong + t1;
            }
        }


        private void registerNextTimer(
                long time, W window, TriggerContext ctx,
                ReducingState<Long> intervalState) throws Exception {
            // 在下一个timer时间和窗口结束时间取最小值
            long nextTimer = Math.min(time + interval, window.maxTimestamp());
            intervalState.add(nextTimer);
            System.err.println("注册定时器："+ new Timestamp(nextTimer));
            ctx.registerEventTimeTimer(nextTimer);
        }
    }

public static class MyProcessWindowFunction extends ProcessWindowFunction<Tuple3<String, Integer, String>, Tuple3<String, Integer, String>, String , TimeWindow>{


        @Override
        public void process(String s, ProcessWindowFunction<Tuple3<String, Integer, String>, Tuple3<String, Integer, String>, String, TimeWindow>.Context context, Iterable<Tuple3<String, Integer, String>> iterable, Collector<Tuple3<String, Integer, String>> collector) throws Exception {
            Tuple3<String, Integer, String> tuple3 = new Tuple3<>();
            Integer sum = 0;
            String device_name = "";
            for (Tuple3<String, Integer, String> tuple : iterable){
                sum += tuple.f1;
                device_name = tuple.f0;
            }
            tuple3.f0 = device_name;
            tuple3.f1 = sum;
            tuple3.f2 = new Timestamp(context.currentProcessingTime()).toString();
            collector.collect(tuple3);
        }
}

WatermarkStrategy<Tuple3<String, Integer, String>> tuple3WatermarkStrategy = WatermarkStrategy
                .<Tuple3<String, Integer, String>>forMonotonousTimestamps().withTimestampAssigner(
                        new SerializableTimestampAssigner<Tuple3<String, Integer, String>>() {
                            @Override
                            public long extractTimestamp(Tuple3<String, Integer, String> stringIntegerStringTuple3, long l) {
                                return System.currentTimeMillis();
                            }
                        }
                );




DataStream<Tuple3<String, Integer, String>> tuple3DataStream2 = tuple3DataStream1
                .assignTimestampsAndWatermarks(tuple3WatermarkStrategy)
                .keyBy(event -> event.f0)
                //.process(new MyKeyedFunction());
                .window(TumblingProcessingTimeWindows.of(Time.days(1), Time.hours(-8)))
                .trigger(MyTrigger.of(Time.seconds(30) ,100000L))
                .process(new MyProcessWindowFunction());