Flink windowAll 函数案例代码实操

太过爱你忘了你带给我的痛 2024-04-17 16:02 56阅读 0赞

1，先创建一个生产者 模拟数据

>     package com.coder.flink.core.a_bilibili;
>     
>     import com.coder.flink.core.streaming2.custormSource.MyNoParalleSource;
>     import org.apache.flink.api.common.serialization.SimpleStringSchema;
>     import org.apache.flink.api.java.utils.ParameterTool;
>     import org.apache.flink.streaming.api.datastream.DataStreamSource;
>     import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
>     import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;
>     
>     import java.util.Properties;
>     
>     public class KafkaProducer {
>         public static void main(String[] args) throws Exception {
>             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
>             DataStreamSource<String> text = env.addSource(new kafkaSourceDemo()).setParallelism(1);
>             args = new String[]{"--input-topic", "dianyou_wxgz_test2", "--output-topic", "dianyou_wxgz_test3", "--bootstrap.servers", "172.10.4.141:9092,172.10.4.142:9092,172.10.4.143:9092",
>                     "--zookeeper.connect", "172.10.4.63:2181,172.10.4.64:2181,172.10.4.65:2181", "--group.id", "cc"};
>     
>             ParameterTool parameterTool = ParameterTool.fromArgs(args);
>             FlinkKafkaProducer010<String> producer = new FlinkKafkaProducer010<String>("topn", new SimpleStringSchema(),  parameterTool.getProperties());
>     
>             text.addSink(producer);
>             env.execute();
>         }
>     }

2，操作代码：

>     package com.coder.flink.core.a_bilibili;
>     
>     import com.coder.flink.core.streaming.KafkaEvent;
>     import com.coder.flink.core.streaming.KafkaEventSchema;
>     import org.apache.flink.api.common.functions.FlatMapFunction;
>     import org.apache.flink.api.common.restartstrategy.RestartStrategies;
>     import org.apache.flink.api.common.serialization.SimpleStringSchema;
>     import org.apache.flink.api.java.tuple.Tuple2;
>     import org.apache.flink.api.java.utils.ParameterTool;
>     import org.apache.flink.configuration.Configuration;
>     import org.apache.flink.streaming.api.TimeCharacteristic;
>     import org.apache.flink.streaming.api.datastream.DataStream;
>     import org.apache.flink.streaming.api.datastream.DataStreamSource;
>     import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
>     import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
>     import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
>     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.windows.TimeWindow;
>     import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
>     import org.apache.flink.util.Collector;
>     
>     import java.sql.Timestamp;
>     import java.util.Comparator;
>     import java.util.Map;
>     import java.util.Properties;
>     import java.util.TreeMap;
>     
>     public class TopN {
>     
>         public static void main(String[] args) throws Exception {
>     
>     
>             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
>     
>             //todo 获取kafka的配置属性
>             args = new String[]{"--input-topic", "dianyou_wxgz_test2", "--output-topic", "dianyou_wxgz_test3", "--bootstrap.servers", "172.10.4.141:9092,172.10.4.142:9092,172.10.4.143:9092",
>                     "--zookeeper.connect", "172.10.4.63:2181,172.10.4.64:2181,172.10.4.65:2181", "--group.id", "cc"};
>     
>             ParameterTool parameterTool = ParameterTool.fromArgs(args);
>             if (parameterTool.getNumberOfParameters() < 5) {
>                 System.out.println("Missing parameters!" +
>                         "Usage: Kafka --input-topic <topic> --output-topic <topic> " +
>                         "--bootstrap.servers <kafka brokers> " +
>                         "--zookeeper.connect <zk quorum> --group.id <some id>");
>                 return;
>             }
>              env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
>     
>     
>     //        //todo 指定输入数据为kafka topic
>             DataStreamSource<String> inputStream = env.addSource(new FlinkKafkaConsumer010<String>(
>                             "topn",
>                             new SimpleStringSchema(),
>                             parameterTool.getProperties())
>     //                        .assignTimestampsAndWatermarks(new CustomWatermarkExtractor())
>                     //todo 对数据按字段分组，然后使用map算子累加
>             );
>     
>             DataStream<Tuple2<String, Integer>> ds = inputStream
>                     .flatMap(new LineSplitter()); //将输入语句split成一个一个单词并初始化count值为1的Tuple2<String, Integer>类型
>     
>     
>             DataStream<Tuple2<String, Integer>> wcount = ds
>                     .keyBy(0)
>                     .window(SlidingProcessingTimeWindows.of(Time.seconds(6),Time.seconds(3)))
>                     //key之后的元素进入一个总时间长度为600s,每5s向后滑动一次的滑动窗口
>                     .sum(1);// 将相同的key的元素第二个count值相加
>     
>     
>             wcount.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(1)))
>                     //所有key元素进入一个5s长的窗口（选5秒是因为上游窗口每5s计算一轮数据，topN窗口一次计算只统计一个窗口时间内的变化）
>                     .process(new TopNAllFunction(3))
>                     .print();
>             //redis sink redis -> 接口
>     
>             env.execute();
>         }//
>     
>     
>         private static final class LineSplitter implements
>                 FlatMapFunction<String, Tuple2<String, Integer>> {
>             @Override
>             public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
>     
>                 out.collect(new Tuple2<String, Integer>(value, 1));
>             }
>         }
>     
>         private static class TopNAllFunction extends ProcessAllWindowFunction<Tuple2<String, Integer>, String, TimeWindow> {
>     
>             private int topSize = 3;
>     
>             public TopNAllFunction(int topSize) {
>     
>                 this.topSize = topSize;
>             }
>     
>             @Override
>             public void open(Configuration parameters) throws Exception {
>                 super.open(parameters);
>             }
>     
>             @Override
>             public void process(
>                     ProcessAllWindowFunction<Tuple2<String, Integer>, String, TimeWindow>.Context arg0,
>                     Iterable<Tuple2<String, Integer>> input,
>                     Collector<String> out) throws Exception {
>     
>                 //treemap按照key降序排列，相同count值不覆盖
>                 TreeMap<Integer, Tuple2<String, Integer>> treemap = new TreeMap<Integer, Tuple2<String, Integer>>(
>                         new Comparator<Integer>() {
>     
>                             @Override
>                             public int compare(Integer y, Integer x) {
>                                 return (x < y) ? -1 : 1;
>                             }
>     
>                         });
>     
>                 //只保留前面TopN个元素
>     
>                 for (Tuple2<String, Integer> element : input) {
>                     treemap.put(element.f1, element);
>                     if (treemap.size() > topSize) {
>                         treemap.pollLastEntry();
>                     }
>                 }
>     
>     
>                 for (Map.Entry<Integer, Tuple2<String, Integer>> entry : treemap
>                         .entrySet()) {
>                     out.collect("=================\n热销图书列表:\n" + new Timestamp(System.currentTimeMillis()) + treemap.toString() + "\n===============\n");
>                 }
>     
>             }
>     
>     
>         }
>     }

3，输出结果为：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

4，也可以用list排序

public void process(
            ProcessAllWindowFunction<Tuple2<String, Integer>, String, TimeWindow>.Context arg0,
            Iterable<Tuple2<String, Integer>> input,
            Collector<String> out) throws Exception {
        List<Tuple2<String, Integer>>  list = new ArrayList<>();
        System.out.println("list = " + list);
        input.forEach(single ->{
            list.add(single);
        });
        Collections.sort(list,new Comparator<Tuple2<String, Integer>>() {
    
            @Override
            public int compare(Tuple2<String, Integer> o1, Tuple2<String, Integer> o2) {
    
                int i = o1.f1 - o2.f1;
                return i;
            }
        });
        System.out.println("list = " + list);

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/06d11f089edf40358c714f5d976b1b16.png