Flink应用简单案例-统计TopN

爱被打了一巴掌 2024-04-22 07:01 187阅读 0赞

#### 文章目录 ####

*   *  pom文件
     *  前置实体类
     *  方式一:使用 ProcessAllWindowFunction
     *  方式一:使用 KeyedProcessFunction

?诉求：网站中一个非常经典的例子，就是实时统计一段时间内的热门 url。例如，需要统计最近10 秒钟内最热门的两个 url 链接，并且每 5 秒钟更新一次。我们知道，这可以用一个滑动窗口来实现，而“热门度”一般可以直接用访问量来表示。于是就需要开滑动窗口收集 url 的访问数据，按照不同的 url 进行统计，而后汇总排序并最终输出前两名。这其实就是著名的“Top N”问题。  
?很显然，简单的增量聚合可以得到 url 链接的访问量，但是后续的排序输出 Top N 就很难实现了。所以接下来我们用窗口处理函数进行实现。

### pom文件 ###

<properties>
            <maven.compiler.source>8</maven.compiler.source>
            <maven.compiler.target>8</maven.compiler.target>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
            <flink.version>1.13.0</flink.version>
            <java.version>1.8</java.version>
            <scala.binary.version>2.12</scala.binary.version>
            <slf4j.version>1.7.30</slf4j.version>
        </properties>
        <dependencies>
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-java</artifactId>
                <version>${flink.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-clients_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
            </dependency>
            
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-api</artifactId>
                <version>${slf4j.version}</version>
            </dependency>
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
                <version>${slf4j.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.logging.log4j</groupId>
                <artifactId>log4j-to-slf4j</artifactId>
                <version>2.14.0</version>
            </dependency>
    
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.18.22</version>
                <scope>compile</scope>
            </dependency>
    
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.58</version>
            </dependency>
    
        </dependencies>

### 前置实体类 ###

@Data
    @NoArgsConstructor
    @AllArgsConstructor
    @Builder
    public class UserViewCount {
        
        public String url;
        public Long count;
        public Long windowStart;
        public Long windowEnd;
    
        @Override
        public String toString() {
        
            return "UserViewCount{" +
                    "url='" + url + '\'' +
                    ", count=" + count +
                    ", windowStart=" + new Timestamp(windowStart) +
                    ", windowEnd=" + new Timestamp(windowEnd) +
                    '}';
        }
    }

import com.entity.UserEvent;
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
    
    import java.util.Calendar;
    import java.util.Random;
    
    public class CustomSouce implements SourceFunction<UserEvent> {
        
        // 声明一个布尔变量，作为控制数据生成的标识位
        private Boolean running = true;
    
        @Override
        public void run(SourceContext<UserEvent> ctx) throws Exception {
        
            Random random = new Random(); // 在指定的数据集中随机选取数据
            String[] users = {
        "Mary", "Alice", "Bob", "Cary"};
            String[] urls = {
        "./home", "./cart", "./fav", "./prod?id=1",
                    "./prod?id=2"};
            while (running) {
        
                ctx.collect(new UserEvent(
                        users[random.nextInt(users.length)],
                        urls[random.nextInt(urls.length)],
                        Calendar .getInstance().getTimeInMillis()
                ));
                // 隔 200 ms生成一个点击事件，方便观测
                Thread.sleep(200);
            }
        }
        @Override
        public void cancel() {
        
            running = false;
        }
    
    }

import lombok.AllArgsConstructor;
    import lombok.Data;
    import lombok.NoArgsConstructor;
    import lombok.ToString;
    
    @Data
    @NoArgsConstructor
    @ToString
    @AllArgsConstructor
    public class UserEvent {
        
        private String userName;
        private String url;
        private Long timestemp;
    }

### 方式一:使用 ProcessAllWindowFunction ###

public class TopN_TestProcessAllWindowFuntion {
        
    
        //取top 3的數據
        private static  Integer needTopCount = 3;
    
        public static void main(String[] args) throws Exception {
        
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            env.getConfig().setAutoWatermarkInterval(100);
    
            SingleOutputStreamOperator<UserEvent> stream = env.addSource(new CustomSouce())
                    .assignTimestampsAndWatermarks(WatermarkStrategy.<UserEvent>forBoundedOutOfOrderness(Duration.ZERO)
                            .withTimestampAssigner(new SerializableTimestampAssigner<UserEvent>() {
        
                                @Override
                                public long extractTimestamp(UserEvent element, long recordTimestamp) {
        
                                    return element.getTimestemp();
                                }
                            }));
    
            //按照用户姓名进行排名
            stream.map(data -> data.getUserName())
                    .windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                    .aggregate(new UserHashMapCountAgg(), new UserCountWindowResult())
                    .print();
    
    
            env.execute();
        }
    
    
    
        public static class UserHashMapCountAgg implements AggregateFunction<String, HashMap<String, Long>, List<Tuple2<String, Long>>>{
        
            @Override
            public HashMap<String, Long> createAccumulator() {
        
                return new HashMap<>();
            }
    
            @Override
            public HashMap<String, Long> add(String value, HashMap<String, Long> accumulator) {
        
                if (accumulator.containsKey(value)){
        
                    accumulator.put(value, accumulator.get(value) + 1L);
                }else{
        
                    accumulator.put(value, 1L);
                }
                return accumulator;
            }
    
            @Override
            public List<Tuple2<String, Long>> getResult(HashMap<String, Long> accumulator) {
        
                List<Tuple2<String, Long>> resultList = new ArrayList<>();
                accumulator.forEach((key, count) -> {
        
                    resultList.add(Tuple2.of(key, count));
                });
    
                //排序
                resultList.sort(new Comparator<Tuple2<String, Long>>() {
        
                    @Override
                    public int compare(Tuple2<String, Long> o1, Tuple2<String, Long> o2) {
        
                        return o2.f1.intValue() - o1.f1.intValue();
                    }
                });
    //            resultList.sort(Comparator.comparing((key_1, key_2) -> key_1.f1.compareTo(key_1.f1)));
    //            Collections.sort(resultList, Comparator.comparing(key -> key.f1));
                return resultList;
            }
    
            @Override
            public HashMap<String, Long> merge(HashMap<String, Long> a, HashMap<String, Long> b) {
        
                //do nothing
                return null;
            }
        }
    
    
        //实现自定义窗口，包装用户输出信息
        public static class UserCountWindowResult extends ProcessAllWindowFunction<List<Tuple2<String, Long>>, String, TimeWindow>{
        
    
            @Override
            public void process(ProcessAllWindowFunction<List<Tuple2<String, Long>>, String, TimeWindow>.Context context, Iterable<List<Tuple2<String, Long>>> elements, Collector<String> out) throws Exception {
        
                StringBuilder stringBuilder = new StringBuilder();
                stringBuilder.append("-------------------------\n");
                stringBuilder.append("窗口结束时间:" + new Timestamp(context.window().getEnd()) + "\n");
    
                //取出top n数据
    
                List<Tuple2<String, Long>> needData = elements.iterator().next();
                if (CollectionUtils.isNotEmpty(needData)) {
        
                    for (Integer i = 0; i < needTopCount; i++) {
        
                        Tuple2<String, Long> data = needData.get(i) == null ? null : needData.get(i);
                        if (null != data){
        
                            String resultStr = "No." + (i + 1) + " "
                                    + "userName:" + data.f0 + " "
                                    + "count:" + data.f1 + "\n";
                            stringBuilder.append(resultStr);
                        }
                    }
                    stringBuilder.append("-------------------------\n");
                }
                out.collect(stringBuilder.toString());
            }
        }
    
    
    }

**截取部分运行结果**  
![在这里插入图片描述][3fcb549ffb644f45be01209199df5b36.png]

### 方式一:使用 KeyedProcessFunction ###

import com.entity.UserEvent;
    import com.entity.UserViewCount;
    import my.test.source.CustomSouce;
    import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
    import org.apache.flink.api.common.eventtime.WatermarkStrategy;
    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.typeinfo.Types;
    import org.apache.flink.configuration.Configuration;
    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.windowing.ProcessWindowFunction;
    import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    import org.apache.flink.util.Collector;
    
    import java.sql.Timestamp;
    import java.util.ArrayList;
    import java.util.Comparator;
    
    public class TopN_KeyedProcessFunction {
        
        public static void main(String[] args) throws Exception {
        
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            env.getConfig().setAutoWatermarkInterval(100);
    
            SingleOutputStreamOperator<UserEvent> eventStream = env.addSource(new CustomSouce())
                    .assignTimestampsAndWatermarks(WatermarkStrategy.<UserEvent>forMonotonousTimestamps()
                    .withTimestampAssigner(new SerializableTimestampAssigner<UserEvent>() {
        
                       @Override
                       public long extractTimestamp(UserEvent element, long
                               recordTimestamp) {
        
                           return element.getTimestemp();
                       }
                   }));
    
            //按照访问url分区，求出每个 url 的访问量
            SingleOutputStreamOperator<UserViewCount> urlCountStream = eventStream.keyBy(data -> data.getUrl())
                            //滑动事件窗口:开窗(10s大小的窗口)、滑动距离5s
                            .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                            .aggregate(new NewUrlViewCountAgg(), new NewUrlViewCountResult());
    
            // 对结果中同一个窗口的统计数据，进行排序处理
            SingleOutputStreamOperator<String> result = urlCountStream.keyBy(data ->
                            data.windowEnd)
                    .process(new TopN(3));
            result.print("result");
    
    
            env.execute();
        }
    
        //自定义增量聚合
        public static class NewUrlViewCountAgg implements AggregateFunction<UserEvent, Long, Long>{
        
            @Override
            public Long createAccumulator() {
        
                return 0L;
            }
    
            @Override
            public Long add(UserEvent value, Long accumulator) {
        
                return accumulator + 1L;
            }
    
            @Override
            public Long getResult(Long accumulator) {
        
                return accumulator;
            }
    
            @Override
            public Long merge(Long a, Long b) {
        
                return null;
            }
        }
    
        //自定义全窗口函数，只需要包装窗口信息
        public static class NewUrlViewCountResult extends ProcessWindowFunction<Long, UserViewCount, String, TimeWindow>{
        
    
            @Override
            public void process(String url, ProcessWindowFunction<Long, UserViewCount, String, TimeWindow>.Context context, Iterable<Long> elements, Collector<UserViewCount> out) throws Exception {
        
                // 结合窗口信息，包装输出内容
                long currentWindowStartTimeStemp = context.window().getStart();
                long currentWindowEndTimeStemp = context.window().getEnd();
                out.collect(new UserViewCount(url, elements.iterator().next(), currentWindowStartTimeStemp, currentWindowEndTimeStemp));
            }
        }
    
        // 自定义处理函数，排序取 top n
        public static class TopN extends KeyedProcessFunction<Long, UserViewCount, String>{
        
            // 将 n 作为属性
            private Integer n;
            // 定义一个列表状态
            private ListState<UserViewCount> urlViewCountListState;
    
            public TopN(Integer n) {
        
                this.n = n;
            }
    
            public TopN() {
        
                super();
            }
    
            @Override
            public void open(Configuration parameters) throws Exception {
        
                // 从环境中获取列表状态句柄
                urlViewCountListState = getRuntimeContext().getListState(
                        new ListStateDescriptor<UserViewCount>("url-view-count-list",
                                Types.POJO(UserViewCount.class)));
            }
    
            @Override
            public void processElement(UserViewCount value, KeyedProcessFunction<Long, UserViewCount, String>.Context ctx, Collector<String> out) throws Exception {
        
                // 将 count 数据添加到列表状态中，保存起来
                urlViewCountListState.add(value);
    
                // 注册 window end + 1ms 后的定时器，等待所有数据到齐开始排序
                ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey() + 1);
            }
    
            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<Long, UserViewCount, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
        
                // 将数据从列表状态变量中取出，放入 ArrayList，方便排序
                ArrayList<UserViewCount> urlViewCountArrayList = new ArrayList<>();
                for (UserViewCount urlViewCount : urlViewCountListState.get()) {
        
                    urlViewCountArrayList.add(urlViewCount);
                }
    
                // 清空状态，释放资源
                urlViewCountListState.clear();
    
    
                // 排序
                urlViewCountArrayList.sort(new Comparator<UserViewCount>() {
        
                    @Override
                    public int compare(UserViewCount o1, UserViewCount o2) {
        
                        return o2.count.intValue() - o1.count.intValue();
                    }
                });
    
    
                // 提取前两名，构建输出结果
                StringBuilder result = new StringBuilder();
                result.append("========================================\n");
                result.append("窗口结束时间：" + new Timestamp(timestamp - 1) + "\n");
                for (int i = 0; i < this.n; i++) {
        
                    UserViewCount userViewCount = urlViewCountArrayList.get(i);
                    if (null == userViewCount){
        
                        break;
                    }
                    UserViewCount needUserView = userViewCount;
                    String info = "No." + (i + 1) + " "
                            + "url：" + userViewCount.url + " "
                            + "浏览量：" + userViewCount.count + "\n";
                    result.append(info);
                }
                result.append("========================================\n");
                out.collect(result.toString());
    
            }
    
        }
    
    }

**部分结果截图**  
![在这里插入图片描述][23d32d1e5cc54820b8fef0c958caebc7.png]

[3fcb549ffb644f45be01209199df5b36.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/22/879ef9bbfd4843e8aa4b831996fd936c.png
[23d32d1e5cc54820b8fef0c958caebc7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/22/7cb086846291462eac618c8ae7394784.png