flink 全局窗口

末蓝、 2022-09-11 14:25 190阅读 0赞

import org.apache.flink.streaming.api.datastream.AllWindowedStream;
    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.windowing.AllWindowFunction;
    import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
    import org.apache.flink.util.Collector;
    
    import java.util.ArrayList;
    import java.util.Comparator;
    import java.util.Iterator;
    import java.util.List;
    import java.util.stream.Stream;
    
    /**
     * @Author: wpp
     * @Date: 2021/9/17 11:03
     */
    public class CountWindowGlobal2 {
        public static void main(String[] args) throws Exception {
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    //                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      env.setRestartStrategy();
            DataStreamSource<String> socketStream = env.socketTextStream("localhost", 9001);
            SingleOutputStreamOperator<Integer> mapStream = socketStream.map(Integer::parseInt);
    
            //划分Non-Keyed countWindowAll，并行度为1
            AllWindowedStream<Integer, GlobalWindow> globalWindowStream = mapStream.countWindowAll(5);
    //只有一个窗口触发了，才会执行下面一次
            SingleOutputStreamOperator<Integer> apply = globalWindowStream.apply(new AllWindowFunction<Integer, Integer, GlobalWindow>() {
    
                @Override
                public void apply(GlobalWindow window, Iterable<Integer> values, Collector<Integer> out) throws Exception {
    
                    List<Integer> list = new ArrayList<>();
    
                    for(Integer val : values){
                       list.add(val);
                    }
    
                     list.sort(new Comparator<Integer>() {
                        @Override
                        public int compare(Integer o1, Integer o2) {
                            return Integer.compare(o1, o2);
                        }
                    });
                    for(Integer row: list){
                        out.collect(row);
                    }
                }
    
            });
    
            apply.print();
    
            env.execute("");
        }
    
    }

flink 状态设置时间

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd3Bwd3BwMQ_size_20_color_FFFFFF_t_70_g_se_x_16][]

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAd3Bwd3BwMQ_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/2a656670bcf040f984e2f5413a87a4e3.png