java flink（二十）实战之电商页面统计热门二(apache log)访问地址使用：详细理解flink延迟处理机制、使用flink侧输出流、flink双计时器使用。数据排序

迷南。 2023-01-21 03:28 4阅读 0赞

基于上文我们本文来具体理解flink的延迟处理机制以及侧输出流使用，如何注册使用两个计时器等。

**1、首先理解flink窗口的范围，flink如何设置每个窗口的范围**

根据flink底层代码，我们可以看到flink的初始窗口的范围选择是根据窗口大小选择最近的整数倍的窗口。

举个例子：

比如我们的窗口大小是10s

第一条进来的数据事件时间是 16000000009

那么flink设置的第一个窗口范围就是\[16000000000, 16000000010 )  前闭后开

**2、理解watermark的使用以及触发**

watermark的主要用途是做延迟处理。比如我们watermark设置的是3s

那么我们在触发16000000010 的窗口结束该进行窗口计算时，我们不会立即触发计算，而是根据事件时间再等待3s

等待16000000011，16000000012， 16000000013 的数据都接收到了 我们发现没有属于\[16000000000, 16000000010 )的数据，

如果在16000000011，16000000012， 16000000013 的数据之间接收到了16000000008的数据，也会算入到\[16000000000, 16000000010 )的窗口中

然后在接收到窗口end+watermark的数据 即16000000013的数据后 才进行\[16000000000, 16000000010 )的窗口计算。

**3、理解计时器的触发**

如果我们计时器设置的是床口结束时间+1ms

那么计时器里边的计算会等到 窗口end+1ms的数据接收到才会执行。即接收到16000000011或者更大的数据（优先级低于watermark，所以会等到3s后触发）。

**4、什么时候数据进入侧输出流**

根据时间语义

.allowedLateness(Time.minutes(1)) //除了watermark等待时间外 先计算一遍，然后1分钟内 每来一条重新计算一遍,更新之前的结果

## **根据上篇内容进行修改 我们增加侧输出流、侧输出流打印、双定时器使用** ##

package Project;
    
    import Bean.ApachLogEvent;
    import Bean.PageViewCount;
    import com.sun.org.apache.xerces.internal.dom.PSVIElementNSImpl;
    import org.apache.commons.compress.utils.Lists;
    import org.apache.flink.api.common.functions.AggregateFunction;
    import org.apache.flink.api.common.state.ListState;
    import org.apache.flink.api.common.state.ListStateDescriptor;
    import org.apache.flink.api.common.state.MapState;
    import org.apache.flink.api.common.state.MapStateDescriptor;
    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.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
    import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    import org.apache.flink.table.planner.expressions.In;
    import org.apache.flink.util.Collector;
    import org.apache.flink.util.OutputTag;
    
    import java.net.URL;
    import java.sql.Timestamp;
    import java.text.SimpleDateFormat;
    import java.util.ArrayList;
    import java.util.Comparator;
    import java.util.Map;
    
    public class HotPages {
        public static void main(String[] args) throws Exception{
            //创建环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //设置时间语义
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
            //设置并行度
            env.setParallelism(1);
    
            //读取文件 转换成POJO类型
            //URL resource = HotPages.class.getResource("/apache.log");
            //DataStream<String> inputStream = env.readTextFile(resource.getPath());
            DataStream<String> inputStream = env.socketTextStream("192.168.6.23", 9999);
            DataStream<ApachLogEvent> dataStream = inputStream
                    .map(line -> {
                        //分隔文件内容
                        String[] fields = line.split(" ");
                        //格式化时间格式
                        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");
                        Long timestamp = simpleDateFormat.parse(fields[3]).getTime();
                        return new ApachLogEvent(fields[0], fields[1],timestamp,fields[5],fields[6]);
                    })
                    //Watermark 设置延迟时间
                    .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<ApachLogEvent>(Time.seconds(3)) {
                        @Override
                        public long extractTimestamp(ApachLogEvent apachLogEvent) {
                            return apachLogEvent.getTimestamp();
                        }
                    });
            //分组开窗聚合
            //定义一个测输出流标签
            //OutputTag<ApachLogEvent> lateTag = new OutputTag<ApachLogEvent>("late");
            //首先过滤 只要get请求即可
            SingleOutputStreamOperator<PageViewCount> windowAggStream = dataStream.filter(data -> "GET".equals(data.getMethod()))
                    //按照url分组
                    //.keyBy(data ->data.getUrl())
                    .keyBy(ApachLogEvent::getUrl)
                    //.keyBy("url")
                    //开窗 10分钟统计一次 5秒钟滑动一次
                    .timeWindow(Time.minutes(10), Time.seconds(5))
                    .allowedLateness(Time.minutes(1)) //除了watermark等待时间外 先计算一遍，然后1分钟内 每来一条重新计算一遍,更新之前的结果
                    .sideOutputLateData(new OutputTag<ApachLogEvent>("late"){}) //侧输出流 超过了一分钟的数据 进入侧输出流
                    .aggregate(new PageCountAgg(), new PageCountResult());
    
            windowAggStream.print("agg");
            //打印侧输出流的数据
            windowAggStream.getSideOutput(new OutputTag<ApachLogEvent>("late"){}).print("late");
    
            //收集同一窗口count数据，排序输出
            DataStream<String> resultStream = windowAggStream.keyBy(PageViewCount::getWindowEnd)
                    .process(new TopNHotPages(3));
            resultStream.print();
            env.execute("hot pages job");
        }
        //自定义预聚合函数
        public static class PageCountAgg implements AggregateFunction<ApachLogEvent, Long, Long>{
            @Override
            public Long createAccumulator() {
                return 0L;
            }
    
            @Override
            public Long add(ApachLogEvent apachLogEvent, Long aLong) {
                return aLong+1;
            }
    
            @Override
            public Long getResult(Long aLong) {
                return aLong;
            }
    
            @Override
            public Long merge(Long aLong, Long acc1) {
                return aLong+acc1;
            }
        }
        //实现自定义窗口函数
         public static class PageCountResult implements WindowFunction<Long, PageViewCount, String, TimeWindow>{
            @Override
            public void apply(String s, TimeWindow timeWindow, Iterable<Long> iterable, Collector<PageViewCount> collector) throws Exception {
                collector.collect( new PageViewCount(s, timeWindow.getEnd(), iterable.iterator().next()));;
            }
        }
        //实现自定义的窗口函数
        public static class TopNHotPages extends KeyedProcessFunction<Long, PageViewCount,String>{
            private static Integer topSize;
    
            public TopNHotPages(int i) {
                topSize=i;
            }
    
            //定义状态 保存当前所有PageViewCount到List中
            //ListState<PageViewCount> pageViewCountListState;
            MapState<String, Long> pageViewCountMapState;
            @Override
            public void processElement(PageViewCount pageViewCount, Context context, Collector<String> collector) throws Exception {
                //pageViewCountListState.add(pageViewCount);
                pageViewCountMapState.put(pageViewCount.getUrl(),pageViewCount.getCount());
                //定时器+1ms 执行统计
                context.timerService().registerProcessingTimeTimer(pageViewCount.getWindowEnd()+1);
                //再定义一个定时器 用于清空状态
                context.timerService().registerProcessingTimeTimer(pageViewCount.getWindowEnd()+60*1000L);
            }
    
            @Override
            public void open(Configuration parameters) throws Exception {
                pageViewCountMapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Long>("page-count-list",String.class,Long.class));
    
            }
    
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                //先判断是否到了窗口关闭清理时间 如果是 直接清空状态返回
                if(timestamp==ctx.getCurrentKey()+60*1000){
                    pageViewCountMapState.clear();
                    return;
                }
                System.out.println("执行任务"+ctx.getCurrentKey());
                ArrayList<Map.Entry<String, Long>>  pageViewCounts = Lists.newArrayList(pageViewCountMapState.entries().iterator());
    
                //排序
                pageViewCounts.sort(new Comparator<Map.Entry<String, Long>>() {
                    @Override
                    public int compare(Map.Entry<String, Long> o1, Map.Entry<String, Long> o2) {
                        if(o1.getValue()>o2.getValue()){
                            return -1;
                        }else if(o1.getValue()<o2.getValue()){
                            return 1;
                        }else{
                            return 0;
                        }
                    }
                });
                //格式化成字符串输出
                //排名信息格式化成string，方便打印输出
                StringBuilder resultBuilder = new StringBuilder();
                resultBuilder.append("==========");
                resultBuilder.append("窗口结束时间: ").append( new Timestamp(timestamp-1)).append("\n");
                //遍历列表，取topN输出
                for(int i = 0;i<Math.min(topSize,pageViewCounts.size());i++){
                    Map.Entry<String, Long> currentItemViewCount = pageViewCounts.get(i);
                    resultBuilder.append("NO ").append(i+1).append(":")
                            .append(" url=").append(currentItemViewCount.getKey())
                            .append("热门度: ").append(currentItemViewCount.getValue())
                            .append("\n");
                }
                resultBuilder.append("==========\n");
                Thread.sleep(1000L);
                out.collect(resultBuilder.toString());
    //            pageViewCounts.clear();
            }
        }
    }