Flink trigger 触发器demo

╰+哭是因爲堅強的太久メ 2023-01-01 01:56 157阅读 0赞

1,如果输出是数据的话，是不满足我们业务需求的 ，因为每次触发都是从头到尾的数据，我想要的是每次触发都是新的数据，可是触发器只能控制触发效果，不能做到控制输出结果，输出结果是由process函数决定的。

没太多意义。

2，完整代码

package com.sjb.flink.trigger;
    
    import org.apache.flink.api.common.functions.ReduceFunction;
    import org.apache.flink.api.common.functions.RichFlatMapFunction;
    import org.apache.flink.api.common.state.MapState;
    import org.apache.flink.api.common.state.MapStateDescriptor;
    import org.apache.flink.api.common.state.ReducingState;
    import org.apache.flink.api.common.state.ReducingStateDescriptor;
    import org.apache.flink.api.common.typeutils.base.LongSerializer;
    import org.apache.flink.api.java.functions.KeySelector;
    
    import org.apache.flink.streaming.api.TimeCharacteristic;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
    import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
    import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.streaming.api.windowing.triggers.Trigger;
    import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    import org.apache.flink.util.Collector;
    
    import java.util.Iterator;
    
    
    /**
     * @program: flink
     * @description: Flink 自定义触发器
     * @author: Mr.Wang
     * @create: 2020-12-30 15:28
     **/
    public class WindowTriggerDemo {
        public static void main(String[] args) {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
    
            DataStreamSource<String> source = env.addSource(new SourceFunction<String>() {
                @Override
                public void run(SourceContext<String> ctx) throws Exception {
                    int count = 0;
                    while (true) {
                        count++;
                        Thread.sleep(1000);
                        long timeMillis = System.currentTimeMillis();
                        ctx.collect("数据写入:" + String.valueOf(timeMillis) + "_________count=" + count);
                    }
                }
    
                @Override
                public void cancel() {
    
                }
            });
    
            SingleOutputStreamOperator<String> process = source.keyBy(new KeySelector<String, String>() {
                @Override
                public String getKey(String s) throws Exception {
                    return "bbb";
                }
            }).window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
                    .trigger(new CustomTrigger(5, 20 * 1000L))
                    .process(new ProcessWindowFunction<String, String, String, TimeWindow>() {
                        @Override
                        public void process(String input, Context context, Iterable<String> iterable, Collector<String> collector) throws Exception {
                            Iterator<String> iterator = iterable.iterator();
                            while (iterator.hasNext()) {
                                String next = iterator.next();
                                collector.collect(next);
                            }
    
                        }
                    });
            process.print("===>");
            try {
                env.execute("开始任务.....");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        public static class CustomTrigger extends Trigger<String, TimeWindow> {
            private int triggerCount = 10;
            private long triggerTime = 5 * 1000L;
    
            private static int flag = 0;
    
            private final ReducingStateDescriptor<Long> stateDesc =
                    new ReducingStateDescriptor<>("count", new Sum(), LongSerializer.INSTANCE);
    
            private final ReducingStateDescriptor<Long> timeStateDesc =
                    new ReducingStateDescriptor<>("fire-time", new Min(), LongSerializer.INSTANCE);
    
            public CustomTrigger(int count, long time) {
                this.triggerCount = count;
                this.triggerTime = time;
            }
    
            @Override
            public TriggerResult onElement(String element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
    
                ReducingState<Long> fireTimestamp = ctx.getPartitionedState(timeStateDesc);
    
                ReducingState<Long> count = ctx.getPartitionedState(stateDesc);
                count.add(1L);
    
                if (count.get() >= triggerCount) {
                    System.out.println("====数据触发........");
                    // 满足条数时也需要清除时间的触发器，如果不是创建结束的触发器
                    if (fireTimestamp.get() != window.maxTimestamp()) {
    //                logger.info("delete trigger : {}, {}", sdf.format(fireTimestamp.get()), fireTimestamp.get());
                        ctx.deleteProcessingTimeTimer(fireTimestamp.get());
                    }
                    count.clear();
                    return TriggerResult.FIRE;
                }
                //todo  触发之后，下一条数据进来才设置时间计数器注册下一次触发的时间
                timestamp = ctx.getCurrentProcessingTime();
                if (fireTimestamp.get() == null) {
                    long nextFireTimestamp = timestamp + triggerTime;
                    ctx.registerProcessingTimeTimer(nextFireTimestamp);
                    fireTimestamp.add(nextFireTimestamp);
                }
    
                return TriggerResult.CONTINUE;
            }
    
            @Override
            public TriggerResult onProcessingTime(long time, TimeWindow timeWindow, TriggerContext ctx) throws Exception {
                // count
                ReducingState<Long> count = ctx.getPartitionedState(stateDesc);
                //interval
                ReducingState<Long> fireTimestamp = ctx.getPartitionedState(timeStateDesc);
    
                // time trigger and window end
                if (time == timeWindow.maxTimestamp()) {
    
                    // 窗口结束，清0条数和时间的计数器
                    count.clear();
                    ctx.deleteProcessingTimeTimer(fireTimestamp.get());
                    fireTimestamp.clear();
                    return TriggerResult.FIRE_AND_PURGE;
                } else if (fireTimestamp.get() != null && fireTimestamp.get().equals(time)) {
    
                    // 时间计数器触发，清0条数和时间计数器
                    count.clear();
                    fireTimestamp.clear();
                    return TriggerResult.FIRE;
                }
                return TriggerResult.CONTINUE;
            }
    
            @Override
            public TriggerResult onEventTime(long l, TimeWindow timeWindow, TriggerContext ctx) throws Exception {
                return TriggerResult.CONTINUE;
            }
    
            @Override
            public void clear(TimeWindow timeWindow, TriggerContext ctx) throws Exception {
                System.out.println("clear ....");
                ctx.getPartitionedState(stateDesc).clear();
                ReducingState<Long> fireTimestamp = ctx.getPartitionedState(timeStateDesc);
                long timestamp = fireTimestamp.get();
                ctx.deleteProcessingTimeTimer(timestamp);
                fireTimestamp.clear();
            }
    
    
            private static class Sum implements ReduceFunction<Long> {
                private static final long serialVersionUID = 1L;
    
                @Override
                public Long reduce(Long value1, Long value2) throws Exception {
                    return value1 + value2;
                }
    
            }
    
            private static class Min implements ReduceFunction<Long> {
                private static final long serialVersionUID = 1L;
    
                @Override
                public Long reduce(Long value1, Long value2) throws Exception {
                    return Math.min(value1, value2);
                }
            }
    
        }
    }

3，打印输出

====数据触发........  
===>> 数据写入:1609316959863\_\_\_\_\_\_\_\_\_count=1  
===>> 数据写入:1609316960874\_\_\_\_\_\_\_\_\_count=2  
===>> 数据写入:1609316961875\_\_\_\_\_\_\_\_\_count=3  
===>> 数据写入:1609316962876\_\_\_\_\_\_\_\_\_count=4  
===>> 数据写入:1609316963876\_\_\_\_\_\_\_\_\_count=5  
===>> 数据写入:1609316964877\_\_\_\_\_\_\_\_\_count=6  
===>> 数据写入:1609316965877\_\_\_\_\_\_\_\_\_count=7  
===>> 数据写入:1609316966878\_\_\_\_\_\_\_\_\_count=8  
===>> 数据写入:1609316967878\_\_\_\_\_\_\_\_\_count=9  
===>> 数据写入:1609316968879\_\_\_\_\_\_\_\_\_count=10  
===>> 数据写入:1609316969879\_\_\_\_\_\_\_\_\_count=11  
===>> 数据写入:1609316970880\_\_\_\_\_\_\_\_\_count=12  
===>> 数据写入:1609316971881\_\_\_\_\_\_\_\_\_count=13  
===>> 数据写入:1609316972881\_\_\_\_\_\_\_\_\_count=14  
===>> 数据写入:1609316973882\_\_\_\_\_\_\_\_\_count=15  
===>> 数据写入:1609316974882\_\_\_\_\_\_\_\_\_count=16  
===>> 数据写入:1609316975883\_\_\_\_\_\_\_\_\_count=17  
===>> 数据写入:1609316976883\_\_\_\_\_\_\_\_\_count=18  
===>> 数据写入:1609316977884\_\_\_\_\_\_\_\_\_count=19  
===>> 数据写入:1609316978885\_\_\_\_\_\_\_\_\_count=20  
====数据触发........  
===>> 数据写入:1609316959863\_\_\_\_\_\_\_\_\_count=1  
===>> 数据写入:1609316960874\_\_\_\_\_\_\_\_\_count=2  
===>> 数据写入:1609316961875\_\_\_\_\_\_\_\_\_count=3  
===>> 数据写入:1609316962876\_\_\_\_\_\_\_\_\_count=4  
===>> 数据写入:1609316963876\_\_\_\_\_\_\_\_\_count=5  
===>> 数据写入:1609316964877\_\_\_\_\_\_\_\_\_count=6  
===>> 数据写入:1609316965877\_\_\_\_\_\_\_\_\_count=7  
===>> 数据写入:1609316966878\_\_\_\_\_\_\_\_\_count=8  
===>> 数据写入:1609316967878\_\_\_\_\_\_\_\_\_count=9  
===>> 数据写入:1609316968879\_\_\_\_\_\_\_\_\_count=10  
===>> 数据写入:1609316969879\_\_\_\_\_\_\_\_\_count=11  
===>> 数据写入:1609316970880\_\_\_\_\_\_\_\_\_count=12  
===>> 数据写入:1609316971881\_\_\_\_\_\_\_\_\_count=13  
===>> 数据写入:1609316972881\_\_\_\_\_\_\_\_\_count=14  
===>> 数据写入:1609316973882\_\_\_\_\_\_\_\_\_count=15  
===>> 数据写入:1609316974882\_\_\_\_\_\_\_\_\_count=16  
===>> 数据写入:1609316975883\_\_\_\_\_\_\_\_\_count=17  
===>> 数据写入:1609316976883\_\_\_\_\_\_\_\_\_count=18  
===>> 数据写入:1609316977884\_\_\_\_\_\_\_\_\_count=19  
===>> 数据写入:1609316978885\_\_\_\_\_\_\_\_\_count=20  
===>> 数据写入:1609316979885\_\_\_\_\_\_\_\_\_count=21  
===>> 数据写入:1609316980885\_\_\_\_\_\_\_\_\_count=22  
===>> 数据写入:1609316981885\_\_\_\_\_\_\_\_\_count=23  
===>> 数据写入:1609316982886\_\_\_\_\_\_\_\_\_count=24  
===>> 数据写入:1609316983887\_\_\_\_\_\_\_\_\_count=25