目的：统计近1小时内的热门商品，每5分钟更新一次
环境：jdk8 工具：idea

创建maven项目

引入jar包

  2.         <flink.version>1.10.1</flink.version>
  3.         <scala.binary.version>2.12</scala.binary.version>
  5.         <dependency>
  6.             <groupId>org.apache.flink</groupId>
  7.             <artifactId>flink-java</artifactId>
  8.             <version>${flink.version}</version>
  9.         </dependency>
  10.         <dependency>
  11.             <groupId>org.apache.flink</groupId>
  12.             <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
  13.             <version>${flink.version}</version>
  14.         </dependency>

代码演示

用文本数据作为实时数据：

Bean

  1. public class ItemViewCount {
  3.     private Long itemId; //商品Id
  4.     private Long windowEnd;//窗口号
  5.     private Long count; //统计
  7.    }
  8. public class UserBehavior {
  10.     private Long userId;//用户Id
  11.     private Long itemId; //商品Id
  12.     private Integer categoryId; //类型
  13.     private String behavior; //行为
  14.     private Long timestamp; //时间戳
  15. }

Main

  2.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //创建流处理执行环境
  4.         env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); //设置为事件时间
  5.         env.setParallelism(1);//并行度为1
  7.         DataStream<String> inputStream = env.readTextFile("文件的绝对路径");
  9.         //将文本数据转换为UserBehavior对象,并且分配时间戳和watermark
  10.         DataStream<UserBehavior> dataStream = inputStream
  11.                 .map(line -> {
  12.                     String[] fields = line.split(",");
  13.                     return new UserBehavior(new Long(fields[0]), new
  14.                             Long(fields[1]), new Integer(fields[2]), fields[3], new Long(fields[4]));
  15.                 })
  16.                 .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {
  17.                     @Override
  18.                     public long extractAscendingTimestamp(UserBehavior userBehavior) {
  19.                         return userBehavior.getTimestamp() * 1000L;
  20.                     }
  21.                 });
  23.        //过滤需要统计的类型,按照商品id分组,开窗长度为一小时,步长5分钟
  24.         DataStream<ItemViewCount> countDataStream = dataStream
  25.                 .filter(line -> "pv".equals(line.getBehavior()))
  26.                 .keyBy("itemId")
  27.                 .timeWindow(Time.hours(1), Time.minutes(5))
  28.                 .aggregate(new CountAgg(), new WindowResultFunction());  //增量聚合,全量聚合
  29.         // 根据窗口号分组,因为要用到Timer,所以用到底层API,在定时器执行时意味该窗口下的数据统计结束,进行输出
  30.         DataStream<String> result = countDataStream
  31.                 .keyBy("windowEnd")
  32.                 .process(new TopNHotItems(5));
  34.         result.print();
  36.         env.execute("hotList");
  38.     public static class CountAgg implements AggregateFunction<UserBehavior, Long, Long> {
  40.         @Override
  41.         public Long createAccumulator() {
  42.             return 0L;
  43.         }
  45.         @Override
  46.         public Long add(UserBehavior value, Long accumulator) {
  47.             return accumulator + 1;
  48.         }
  50.         @Override
  51.         public Long getResult(Long accumulator) {
  52.             return accumulator;
  53.         }
  55.         @Override
  56.         public Long merge(Long a, Long b) {
  57.             return a + b;
  58.         }
  59.     }
  61.     public static class WindowResultFunction implements WindowFunction<Long, ItemViewCount, Tuple, TimeWindow> {
  63.         @Override
  64.         public void apply(Tuple tuple, TimeWindow timeWindow, Iterable<Long> iterable, Collector<ItemViewCount> collector) throws Exception {
  65.             Long itemId = tuple.getField(0);
  66.             Long windowEnd = timeWindow.getEnd();
  67.             Long count = iterable.iterator().next();
  69.             collector.collect(new ItemViewCount(itemId, windowEnd, count));
  70.         }
  71.     }
  73.     public static class TopNHotItems extends KeyedProcessFunction<Tuple, ItemViewCount, String> {
  75.         private Integer size;
  78.         ListState<ItemViewCount> listState;
  80.         public TopNHotItems(Integer size) {
  81.             this.size = size;
  82.         }
  84.         @Override
  85.         public void processElement(ItemViewCount itemViewCount, Context context, Collector<String> collector) throws Exception {
  86.             listState.add(itemViewCount);
  87.             context.timerService().registerEventTimeTimer(itemViewCount.getWindowEnd() + 1);
  88.         }
  90.         @Override
  91.         public void open(Configuration parameters) throws Exception {
  92.             listState = getRuntimeContext().getListState(new ListStateDescriptor<ItemViewCount>("", ItemViewCount.class));
  93.         }
  95.         @Override
  96.         public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
  97.             ArrayList<ItemViewCount> itemViewCounts =
  98.                     Lists.newArrayList(listState.get().iterator());
  100.             itemViewCounts.sort(new Comparator<ItemViewCount>() {
  101.                 @Override
  102.                 public int compare(ItemViewCount o1, ItemViewCount o2) {
  103.                     return o2.getCount().intValue() - o1.getCount().intValue();
  104.                 }
  105.             });
  107.             StringBuilder result = new StringBuilder();
  108.             result.append("====================================\n");
  109.             result.append("窗口结束时间: ").append(new Timestamp(timestamp - 1)).append("\n");
  111.             for (int i = 0; i < size; i++) {
  112.                 ItemViewCount currentItemViewCount = itemViewCounts.get(i);
  113.                 result.append("No").append(i + 1).append(":")
  114.                         .append(" 商品 ID=")
  115.                         .append(currentItemViewCount.getItemId())
  116.                         .append(" 浏览量=")
  117.                         .append(currentItemViewCount.getCount())
  118.                         .append("\n");
  119.             }
  120.             result.append("====================================\n\n");
  122.             Thread.sleep(1000);
  123.             out.collect(result.toString());
  125.         }
  126.     }

热门商品统计-文本版.rar(4.8 MB)

结果

  1. SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
  2. SLF4J: Defaulting to no-operation (NOP) logger implementation
  3. SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
  4. ====================================
  5. 窗口结束时间: 2017-11-26 09:05:00.0
  6. No1: 商品 ID=5051027 浏览量=3
  7. No2: 商品 ID=3493253 浏览量=3
  8. No3: 商品 ID=4261030 浏览量=3
  9. No4: 商品 ID=4894670 浏览量=2
  10. No5: 商品 ID=3781391 浏览量=2
  11. ====================================
  14. ====================================
  15. 窗口结束时间: 2017-11-26 09:10:00.0
  16. No1: 商品 ID=812879 浏览量=5
  17. No2: 商品 ID=2600165 浏览量=4
  18. No3: 商品 ID=2828948 浏览量=4
  19. No4: 商品 ID=2338453 浏览量=4
  20. No5: 商品 ID=4261030 浏览量=4
  21. ====================================