flink 窗口和水位线

一、窗口

    **1、什么是窗口？**

    我们的flink主要是用来处理无界数据流，一种方式就是将我们的无界数据流切割成有限的“数据块”进行处理，这就是我们的窗口(window)。

    **2、窗口分类**

滚动窗口、滑动窗口、会话窗口

滚动窗口：timeWindow（Time.seconds(3)） count-tumbling-window

滑动窗口：timeWindow（Time.seconds(5),Time.seconds(3)) count-sliding-window

** 3、窗口能解决什么问题（为什么使用窗口）**

    首先flink是一个实现了流批一体的计算框架，当我们使用批处理时我们引入了窗口计算，实现我们的批处理。

** 4、滚动窗口（每个区消费总额Top3的公司）**

public class CityShopNameTopN {

    public static void main(String[] args) throws Exception{

        // TODO: 2022/9/2创建Flink流式处理环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        // TODO: 2022/9/2 设置并行度
        environment.setParallelism(1);

        String uu = UUID.randomUUID().toString().substring(0, 6).replace("-", "");

        String groupId = "ware_goods_group"+uu;

        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource("dwd_foo_order_detail",groupId);
        DataStreamSource<String> order_detail = environment.addSource(kafkaSource);
        SingleOutputStreamOperator<JSONObject> map1 = order_detail.map(d -> JSON.parseObject(d));
//水位线

        SingleOutputStreamOperator<JSONObject> watermarks = map1.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {

                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        long time = 0;
                        try {
                            time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(element.getString("createTime")).getTime();
                        } catch (ParseException e) {
                            e.printStackTrace();
                        }
                        return time;
                    }
                }));

        SingleOutputStreamOperator<Tuple3<String, String, Double>> map = watermarks.map(new MapFunction<JSONObject, Tuple3<String, String, Double>>() {
            @Override
            public Tuple3<String, String, Double> map(JSONObject value) throws Exception {
                String goodsNum = value.getString("goodsNum");
                String goodsPrice = value.getString("goodsPrice");
                return new Tuple3<>(value.getString("regionName"), value.getString("cityName"), Integer.valueOf(goodsNum) * Double.valueOf(goodsPrice));
            }
        });
        SingleOutputStreamOperator<Tuple3<String, String, Double>> process = map.keyBy(data -> data.f0 + "," + data.f1).sum(2).keyBy(data -> data.f0 + "," + data.f1)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(1))).process(new ProcessWindowFunction<Tuple3<String, String, Double>, Tuple3<String, String, Double>, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<Tuple3<String, String, Double>> iterable, Collector<Tuple3<String, String, Double>> collector) throws Exception {
                        ArrayList<Tuple3<String, String, Double>> list = new ArrayList<>();
                        for (Tuple3<String, String, Double> value : iterable) {
                            list.add(value);
                        }
                        list.sort(new Comparator<Tuple3<String, String, Double>>() {
                            @Override
                            public int compare(Tuple3<String, String, Double> o1, Tuple3<String, String, Double> o2) {
                                return (int) (o2.f2 - o1.f2);
                            }
                        });
                        for (int i = 0; i < list.size() && i < 3; i++) {
                            collector.collect(list.get(i));
                        }
                    }
                });

        process.print();
        //落地
         process.addSink(new SinkPG());

     这个指标中我们使用的是滚动窗口，（由于数据过少）采用了每一秒把进来的数据进行一次计算，当然这样是不合乎常规的，比如我们需要统计一天的销售额，我们可以使用TumblingProcessingTimeWindows.of(Time.days(1))来给他一个一天的窗口。 

二、watermark（水位线）

** 1、什么是watermark(水位线的机制)？**

    watermark本质就是一个时间戳。实际上就是在原有的结束时间上再多等一个最大允许的数据延迟时间或者乱序时间，一旦有事件时间在这个多等时间刻度线后的消息事件达到就立刻触发窗口计算。

** 2、 水位线能解决什么问题（为什么使用水位线）**

    水位线是事件时间的进展，它是整个应用的全局逻辑时钟。水位线生成之后，会随着数据在任务间流动，从而给每个任务指明当前的事件时间；当然水位线的时间也不是越大越好，如果过大也会降低、影响我们的实时性；如果在我们的一个最大延迟时间数据还未到，我们可以定义一个测输出流标签，把迟到的数据放到我们的标签中

** 3、统计url的访问量**

public class Test{
    public static void main(String[] args) throws Exception {
        //流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置全局并行度
        env.setParallelism(1);
        //设置水位线生成间隔
        env.getConfig().setAutoWatermarkInterval(100);
        SingleOutputStreamOperator<Event> eventStream = env.socketTextStream("hadoop103", 9999).map(
                new MapFunction<String, Event>() {
                    @Override
                    public Event map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new Event(split[0].trim(), split[1].trim(), Long.valueOf(split[2].trim()));
                    }
                }
        ).returns(new TypeHint<Event>() {
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event element, long recordTimestamp) {
                        return element.timestamp;
                    }
                }));

        eventStream.print("  input  ");

        //定义输出标签
        OutputTag<Event> later = new OutputTag<Event>("later") {
        };

        //统计 url 访问量

        SingleOutputStreamOperator<UrlViewCount> result = eventStream.keyBy(data -> data.url)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))

                .allowedLateness(Time.minutes(1)) //1 min 延迟
                //迟到数据输出到册数出列
                .sideOutputLateData(later)
                .aggregate(new UrlCountViewExample.UrlViewCountAgg(), new UrlCountViewExample.UrlViewCountResult());

        result.print("  result  ");
        //侧输出流
        result.getSideOutput(later).print("later datas");

        env.execute();
    }

     统计访问量我们给了一分钟等待迟到数据如果还未到达我们就定义一个标签，把迟到数据放到测输出流以便后边的计算。

    **三、总结 **

** 在我们的计算中watermark和window是结合在一起使用的，首先我们给它定义一个水位线，保证数据的乱序程度，然后根据业务需求定义我们的窗口（更适用那种窗口）。**