devbox
从前慢现在也慢
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

java Flink（十四）Flink实战之电商用户行为分析之统计热门商品并用Flink table api实现 使用：map、aggregate、process、定时器、filter_java flink 实战

作者：从前慢现在也慢 | 2024-06-28 23:44:24

赞

踩

java flink 实战

继续学习了Fklink，今天在网上学习Flink的实战视频，跟着写了一下代码。

因为是学习，为了方便主要是用了读取文件的方式获取源数据，主要使用了滑动窗口、keyby、filter等功能。

题目：

统计某个时间段内，热门商品访问的前几名

首先是正常实现

1、目录结构：

1是存放bean对象类 2是放项目代码 3是文件路径

2、pom文件

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0"
  3.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  4.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  5.     <modelVersion>4.0.0</modelVersion>
  6.  
  7.     <groupId>org.example</groupId>
  8.     <artifactId>UserBehavior</artifactId>
  9.     <packaging>pom</packaging>
  10.     <version>1.0-SNAPSHOT</version>
  11.     <modules>
  12.         <module>HottltemsAnalysis</module>
  13.     </modules>
  14.  
  15.     <properties>
  16.             <flink.version>1.10.1</flink.version>
  17.             <scala.binary.version>2.12</scala.binary.version>
  18.             <kafka.version>2.2.0</kafka.version>
  19.     </properties>
  20.  
  21.     <dependencies>
  22.         <dependency>
  23.             <groupId>org.apache.flink</groupId>
  24.             <artifactId>flink-java</artifactId>
  25.             <version>${flink.version}</version>
  26.         </dependency>
  27.         <dependency>
  28.             <groupId>org.apache.flink</groupId>
  29.             <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
  30.             <version>${flink.version}</version>
  31.         </dependency>
  32.         <dependency>
  33.             <groupId>org.apache.kafka</groupId>
  34.             <artifactId>kafka_${scala.binary.version}</artifactId>
  35.             <version>${kafka.version}</version>
  36.         </dependency>
  37.         <dependency>
  38.             <groupId>org.apache.flink</groupId>
  39.             <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
  40.             <version>${flink.version}</version>
  41.         </dependency>
  42.     </dependencies>
  43.  
  44. </project>

3、数据源文件内容（部分）

  1. 464646,2512167,2733371,pv,1511658001
  2. 1007641,5046581,2355072,pv,1511658001
  3. 723938,4719377,1464116,pv,1511658001
  4. 513008,3472922,401357,pv,1511658001
  5. 769215,22738,2355072,pv,1511658002
  6. 652863,4967749,1320293,pv,1511658002
  7. 801610,900305,634390,pv,1511658002
  8. 411478,3259235,2667323,pv,1511658002
  9. 431664,764155,2520377,pv,1511658002
  10. 487768,4125503,2465336,pv,1511658002
  11. 223813,4104826,2042400,pv,1511658002
  12. 672849,1822977,4801426,fav,1511658002
  13. 550127,4602135,65362,pv,1511658002
  14. 205752,1467139,171529,pv,1511658002
  15. 64419,2029769,2729260,pv,1511658002
  16. 756093,2881426,2520377,pv,1511658002
  17. 48353,4362292,583014,pv,1511658002
  18. 355509,4712764,4082778,pv,1511658003
  19. 826492,4016552,2735466,pv,1511658003
  20. 624915,2243311,2520377,pv,1511658003
  21. 682317,655740,982926,fav,1511658003
  22. 677621,1051389,4801426,pv,1511658003
  23. 422974,4649255,4818107,pv,1511658003
  24. 86512,563566,4756105,pv,1511658003
  25. 565218,2331370,3607361,pv,1511658003
  26. 232313,4182588,1730376,pv,1511658003
  27. 436966,1329977,3607361,cart,1511658003
  28. 561158,269170,2342116,fav,1511658003

4、代码展示

首先是用于返回结果的包装类

  1. package Bean;
  2.  
  3. //包装类 用于返回展示的类型
  4. public class ItemViewCount {
  5.     private Long itemId;//分组Id
  6.     private Long windowEnd;//窗口时间戳
  7.     private Long count;//次数
  8.  
  9.     public ItemViewCount() {
  10.     }
  11.  
  12.     public ItemViewCount(Long itemId, Long windowEnd, Long count) {
  13.         this.itemId = itemId;
  14.         this.windowEnd = windowEnd;
  15.         this.count = count;
  16.     }
  17.  
  18.     public Long getItemId() {
  19.         return itemId;
  20.     }
  21.  
  22.     public void setItemId(Long itemId) {
  23.         this.itemId = itemId;
  24.     }
  25.  
  26.     public Long getWindowEnd() {
  27.         return windowEnd;
  28.     }
  29.  
  30.     public void setWindowEnd(Long windowEnd) {
  31.         this.windowEnd = windowEnd;
  32.     }
  33.  
  34.     public Long getCount() {
  35.         return count;
  36.     }
  37.  
  38.     public void setCount(Long count) {
  39.         this.count = count;
  40.     }
  41.  
  42.     @Override
  43.     public String toString() {
  44.         return "ItemViewCount{" +
  45.                 "itemId=" + itemId +
  46.                 ", windowEnd=" + windowEnd +
  47.                 ", count=" + count +
  48.                 '}';
  49.     }
  50. }

 

然后是用于接收数据的包装类

  1. package Bean;
  2.  
  3. public class UserBehavior {
  4.     //定义私有属性
  5.     private Long userId; //用户Id
  6.     private Long itemId; //分组Id
  7.     private Integer categoryId; //类别Id
  8.     private String behavior; //动作类型
  9.     private Long timestamp; //时间戳
  10.  
  11.     public UserBehavior() {
  12.     }
  13.  
  14.     public UserBehavior(Long userId, Long itemId, Integer categoryId, String behavior, Long timestamp) {
  15.         this.userId = userId;
  16.         this.itemId = itemId;
  17.         this.categoryId = categoryId;
  18.         this.behavior = behavior;
  19.         this.timestamp = timestamp;
  20.     }
  21.  
  22.     public Long getUserId() {
  23.         return userId;
  24.     }
  25.  
  26.     public void setUserId(Long userId) {
  27.         this.userId = userId;
  28.     }
  29.  
  30.     public Long getItemId() {
  31.         return itemId;
  32.     }
  33.  
  34.     public void setItemId(Long itemId) {
  35.         this.itemId = itemId;
  36.     }
  37.  
  38.     public Integer getCategoryId() {
  39.         return categoryId;
  40.     }
  41.  
  42.     public void setCategoryId(Integer categoryId) {
  43.         this.categoryId = categoryId;
  44.     }
  45.  
  46.     public String getBehavior() {
  47.         return behavior;
  48.     }
  49.  
  50.     public void setBehavior(String behavior) {
  51.         this.behavior = behavior;
  52.     }
  53.  
  54.     public Long getTimestamp() {
  55.         return timestamp;
  56.     }
  57.  
  58.     public void setTimestamp(Long timestamp) {
  59.         this.timestamp = timestamp;
  60.     }
  61.  
  62.     @Override
  63.     public String toString() {
  64.         return "UserBehavior{" +
  65.                 "userId=" + userId +
  66.                 ", itemId=" + itemId +
  67.                 ", categoryId=" + categoryId +
  68.                 ", behavior='" + behavior + '\'' +
  69.                 ", timestamp=" + timestamp +
  70.                 '}';
  71.     }
  72. }

项目代码

  1. package Project;
  2.  
  3. import Bean.ItemViewCount;
  4. import Bean.UserBehavior;
  5. import org.apache.commons.compress.utils.Lists;
  6. import org.apache.flink.api.common.functions.AggregateFunction;
  7. import org.apache.flink.api.common.functions.FilterFunction;
  8. import org.apache.flink.api.common.functions.MapFunction;
  9. import org.apache.flink.api.common.state.ListState;
  10. import org.apache.flink.api.common.state.ListStateDescriptor;
  11. import org.apache.flink.api.java.tuple.Tuple;
  12. import org.apache.flink.configuration.Configuration;
  13. import org.apache.flink.streaming.api.TimeCharacteristic;
  14. import org.apache.flink.streaming.api.datastream.DataStream;
  15. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  16. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  17. import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
  18. import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
  19. import org.apache.flink.streaming.api.windowing.time.Time;
  20. import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
  21. import org.apache.flink.util.Collector;
  22.  
  23. import java.sql.Timestamp;
  24. import java.util.ArrayList;
  25. import java.util.Comparator;
  26.  
  27. /**
  28.  * 热门商品统计
  29.  */
  30. public class HotItems {
  31.     public static void main(String[] args) throws Exception{
  32.         //创建执行环境
  33.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  34.         //并行度
  35.         env.setParallelism(1);
  36.         //设置时间语义
  37.         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  38.  
  39.         //读取数据，创建DataStream数据流(这里我们可以用kafka 但是为了方便我们使用读取文件)
  40.         DataStream<String> inputStream = env.readTextFile("D:\\idle\\UserBehavior\\HottltemsAnalysis\\src\\main\\resources\\UserBehavior.csv");
  41.         //转换为POJO，分配时间戳和watermark
  42.         DataStream<UserBehavior> dataStream = inputStream
  43.                 .map(new MapFunction<String, UserBehavior>() {
  44.                     public UserBehavior map(String s) throws Exception {
  45.                         String[] fields = s.split(",");
  46.                         return new UserBehavior(new Long(fields[0]), new Long(fields[1]), new Integer(fields[2]), fields[3], new Long(fields[4]));
  47.                     }
  48.                 }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
  49.                     @Override
  50.                     public long extractAscendingTimestamp(UserBehavior element) {
  51.                         return element.getTimestamp()*1000L;
  52.                     }
  53.                 });
  54.  
  55.         //分组开窗聚合，得到每个窗口内各个商品的count值
  56.         DataStream<ItemViewCount> windowAggStream = dataStream
  57.                 .filter(new FilterFunction<UserBehavior>() {
  58.                     public boolean filter(UserBehavior userBehavior) throws Exception {
  59.                         //过滤pv行为
  60.                         return "pv".equals(userBehavior.getBehavior());
  61.                     }
  62.                 })
  63.                 .keyBy("itemId")//按商品ID分组
  64.                 .timeWindow(Time.hours(1), Time.minutes(5))//开启滑动窗口,一小时一个窗口，每五分钟滑动一次
  65.                 .aggregate(new ItemCountAgg(), new WindowItemCountResult());//聚合操作，传入增量聚合函数以及全窗口函数用来包装信息
  66.         //收集同一窗口的所有商品count数据，排序输出top n
  67.         DataStream<String> resultStream = windowAggStream
  68.                 .keyBy("windowEnd")//按照窗口分组
  69.                 .process(new TopNHotItems(5)); //用自定义函数排序取前5
  70.         resultStream.print();
  71.         env.execute("hot items analysis");
  72.     }
  73.     //实现自定义增量聚合函数
  74.     public static class ItemCountAgg implements AggregateFunction<UserBehavior, Long, Long>{
  75.         public Long createAccumulator() {
  76.             return 0L;
  77.         }
  78.  
  79.         public Long add(UserBehavior userBehavior, Long aLong) {
  80.             return aLong+1;
  81.         }
  82.  
  83.         public Long getResult(Long aLong) {
  84.             return aLong;
  85.         }
  86.  
  87.         public Long merge(Long aLong, Long acc1) {
  88.             return aLong+acc1;
  89.         }
  90.     }
  91.     //自定义全窗口函数
  92.     public static class WindowItemCountResult implements WindowFunction<Long, ItemViewCount, Tuple, TimeWindow>{
  93.         public void apply(Tuple tuple, TimeWindow window, Iterable<Long> input, Collector<ItemViewCount> out) throws Exception {
  94.             Long itemId = tuple.getField(0);
  95.             Long windowEnd = window.getEnd();
  96.             Long count = input.iterator().next(); //这里是从ItemCountAgg聚合的结果中拿出数据
  97.  
  98.             out.collect(new ItemViewCount(itemId, windowEnd, count));
  99.         }
  100.     }
  101.     //自定义实现process function
  102.     public static class TopNHotItems extends KeyedProcessFunction<Tuple, ItemViewCount, String>{
  103.         //定义属性 top n 大小
  104.         private Integer topSize;
  105.  
  106.         public TopNHotItems(Integer topSize) {
  107.             this.topSize = topSize;
  108.         }
  109.         //定义列表状态，保存当前窗口内所有输出的ItemViewCount
  110.         ListState<ItemViewCount> itemViewCountListState;
  111.  
  112.         //通过上下文注册状态列表
  113.         @Override
  114.         public void open(Configuration parameters) throws Exception {
  115.             itemViewCountListState = getRuntimeContext().getListState(new ListStateDescriptor<ItemViewCount>("item-view-count-list", ItemViewCount.class));
  116.         }
  117.  
  118.         public void processElement(ItemViewCount value, Context ctx, Collector<String> out) throws Exception {
  119.             //每来一条数据，存入状态
  120.             itemViewCountListState.add(value);
  121.             //并注册定时器
  122.             ctx.timerService().registerEventTimeTimer(value.getWindowEnd() + 1);
  123.         }
  124.  
  125.         @Override
  126.         public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
  127.             //定时器出发，当前已收集到所有数据，排序输出
  128.             ArrayList<ItemViewCount> itemViewCounts = Lists.newArrayList(itemViewCountListState.get().iterator());
  129.             itemViewCounts.sort(new Comparator<ItemViewCount>() {
  130.                 public int compare(ItemViewCount o1, ItemViewCount o2) {
  131.                     return o2.getCount().intValue()-o1.getCount().intValue();
  132.                 }
  133.             });
  134.             //排名信息格式化成string，方便打印输出
  135.             StringBuilder resultBuilder = new StringBuilder();
  136.             resultBuilder.append("==========");
  137.             resultBuilder.append("窗口结束时间: ").append( new Timestamp(timestamp-1)).append("\n");
  138.             //遍历列表，取topN输出
  139.             for(int i = 0;i<Math.min(topSize,itemViewCounts.size());i++){
  140.                 ItemViewCount currentItemViewCount = itemViewCounts.get(i);
  141.                 resultBuilder.append("NO ").append(i+1).append(":")
  142.                         .append(" ID=").append(currentItemViewCount.getItemId())
  143.                         .append("热门度: ").append(currentItemViewCount.getCount())
  144.                         .append("\n");
  145.             }
  146.             resultBuilder.append("==========\n");
  147.             Thread.sleep(1000L);
  148.             out.collect(resultBuilder.toString());
  149.         }
  150.     }
  151.  
  152. }

5、结果展示

 

用Flink table api实现 

增加依赖：

代码：

  1. package Project;
  2.  
  3. import Bean.UserBehavior;
  4. import javafx.scene.control.Tab;
  5. import org.apache.flink.api.common.functions.MapFunction;
  6. import org.apache.flink.streaming.api.TimeCharacteristic;
  7. import org.apache.flink.streaming.api.datastream.DataStream;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
  10. import org.apache.flink.table.api.EnvironmentSettings;
  11. import org.apache.flink.table.api.Slide;
  12. import org.apache.flink.table.api.Table;
  13. import org.apache.flink.table.api.java.StreamTableEnvironment;
  14. import org.apache.flink.types.Row;
  15. import org.omg.CORBA.Environment;
  16.  
  17. public class HotitemsWithSql {
  18.     public static void main(String[] args) throws Exception{
  19.         //创建执行环境
  20.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  21.         //并行度
  22.         env.setParallelism(1);
  23.         //设置时间语义
  24.         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  25.  
  26.         //读取数据，创建DataStream数据流(这里我们可以用kafka 但是为了方便我们使用读取文件)
  27.         DataStream<String> inputStream = env.readTextFile("D:\\idle\\UserBehavior\\HottltemsAnalysis\\src\\main\\resources\\UserBehavior.csv");
  28.         //转换为POJO，分配时间戳和watermark
  29.         DataStream<UserBehavior> dataStream = inputStream
  30.                 .map(new MapFunction<String, UserBehavior>() {
  31.                     public UserBehavior map(String s) throws Exception {
  32.                         String[] fields = s.split(",");
  33.                         return new UserBehavior(new Long(fields[0]), new Long(fields[1]), new Integer(fields[2]), fields[3], new Long(fields[4]));
  34.                     }
  35.                 }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
  36.                     @Override
  37.                     public long extractAscendingTimestamp(UserBehavior element) {
  38.                         return element.getTimestamp()*1000L;
  39.                     }
  40.                 });
  41.         //创建表执行环境 blink版本
  42.         EnvironmentSettings settings = EnvironmentSettings.newInstance()
  43.                 .useBlinkPlanner()
  44.                 .inStreamingMode()
  45.                 .build();
  46.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
  47.         //将流转换成表 并筛选想要的数据
  48.         Table dataTable = tableEnv.fromDataStream(dataStream, "itemId, behavior, timestamp.rowtime as ts");
  49.         //分组开窗
  50.         //table api
  51.         Table windowAggTable = dataTable
  52.                 .filter("behavior='pv'")
  53.                 .window(Slide.over("1.hours").every("5.minutes").on("ts").as("w"))
  54.                 .groupBy("itemId, w")
  55.                 .select("itemId, w.end as windowEnd, itemId.count as cnt");
  56.         //利用开窗函数 对count值进行排序并获取row Number 得到TopN
  57.         //SQL
  58.         DataStream<Row> aggStream = tableEnv.toAppendStream(windowAggTable, Row.class);
  59.         tableEnv.createTemporaryView("agg", aggStream, "itemId, windowEnd, cnt");
  60.         //分组排序
  61.         Table resultTable = tableEnv.sqlQuery("select * from "+
  62.                 " ( select *, ROW_NUMBER() over (partition by windowEnd order by cnt desc) as row_num from agg)"+
  63.                 "where row_num <= 5");
  64.         tableEnv.toRetractStream(resultTable, Row.class).print();
  65.         env.execute("hot items with sql job");
  66.  
  67.  
  68.     }
  69. }

结果展示：

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/从前慢现在也慢/article/detail/767704
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。