Flink 零基础实战教程：如何计算实时热门商品

$ cd my-flink-project/src/main/resources     
$ curl https://raw.githubusercontent.com/wuchong/my-flink-project/master/src/main/resources/UserBehavior.csv > UserBehavior.csv     

package myflink;     
     
public class HotItems {     
     
  public static void main(String[] args) throws Exception {     
         
  }     
}     

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();     
// 为了打印到控制台的结果不乱序，我们配置全局的并发为1，这里改变并发对结果正确性没有影响     
env.setParallelism(1);     

/** 用户行为数据结构 **/     
public static class UserBehavior {     
  public long userId;         // 用户ID     
  public long itemId;         // 商品ID     
  public int categoryId;      // 商品类目ID     
  public String behavior;     // 用户行为, 包括("pv", "buy", "cart", "fav")     
  public long timestamp;      // 行为发生的时间戳，单位秒     
}     

// UserBehavior.csv 的本地文件路径     
URL fileUrl = HotItems2.class.getClassLoader().getResource("UserBehavior.csv");     
Path filePath = Path.fromLocalFile(new File(fileUrl.toURI()));     
// 抽取 UserBehavior 的 TypeInformation，是一个 PojoTypeInfo     
PojoTypeInfo<UserBehavior> pojoType = (PojoTypeInfo<UserBehavior>) TypeExtractor.createTypeInfo(UserBehavior.class);     
// 由于 Java 反射抽取出的字段顺序是不确定的，需要显式指定下文件中字段的顺序     
String[] fieldOrder = new String[]{"userId", "itemId", "categoryId", "behavior", "timestamp"};     
// 创建 PojoCsvInputFormat     
PojoCsvInputFormat<UserBehavior> csvInput = new PojoCsvInputFormat<>(filePath, pojoType, fieldOrder);     

DataStream<UserBehavior> dataSource = env.createInput(csvInput, pojoType);     

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);     

DataStream<UserBehavior> timedData = dataSource     
    .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<UserBehavior>() {     
      @Override     
      public long extractAscendingTimestamp(UserBehavior userBehavior) {     
        // 原始数据单位秒，将其转成毫秒     
        return userBehavior.timestamp * 1000;     
      }     
    });     

DataStream<UserBehavior> pvData = timedData     
    .filter(new FilterFunction<UserBehavior>() {     
      @Override     
      public boolean filter(UserBehavior userBehavior) throws Exception {     
        // 过滤出只有点击的数据     
        return userBehavior.behavior.equals("pv");     
      }     
    });     

DataStream<ItemViewCount> windowedData = pvData     
    .keyBy("itemId")     
    .timeWindow(Time.minutes(60), Time.minutes(5))     
    .aggregate(new CountAgg(), new WindowResultFunction());     

/** COUNT 统计的聚合函数实现，每出现一条记录加一 */     
public static class CountAgg implements AggregateFunction<UserBehavior, Long, Long> {     
     
  @Override     
  public Long createAccumulator() {     
    return 0L;     
  }     
     
  @Override     
  public Long add(UserBehavior userBehavior, Long acc) {     
    return acc + 1;     
  }     
     
  @Override     
  public Long getResult(Long acc) {     
    return acc;     
  }     
     
  @Override     
  public Long merge(Long acc1, Long acc2) {     
    return acc1 + acc2;     
  }     
}     

/** 用于输出窗口的结果 */     
public static class WindowResultFunction implements WindowFunction<Long, ItemViewCount, Tuple, TimeWindow> {     
     
  @Override     
  public void apply(     
      Tuple key,  // 窗口的主键，即 itemId     
      TimeWindow window,  // 窗口     
      Iterable<Long> aggregateResult, // 聚合函数的结果，即 count 值     
      Collector<ItemViewCount> collector  // 输出类型为 ItemViewCount     
  ) throws Exception {     
    Long itemId = ((Tuple1<Long>) key).f0;     
    Long count = aggregateResult.iterator().next();     
    collector.collect(ItemViewCount.of(itemId, window.getEnd(), count));     
  }     
}     
     
/** 商品点击量(窗口操作的输出类型) */     
public static class ItemViewCount {     
  public long itemId;     // 商品ID     
  public long windowEnd;  // 窗口结束时间戳     
  public long viewCount;  // 商品的点击量     
     
  public static ItemViewCount of(long itemId, long windowEnd, long viewCount) {     
    ItemViewCount result = new ItemViewCount();     
    result.itemId = itemId;     
    result.windowEnd = windowEnd;     
    result.viewCount = viewCount;     
    return result;     
  }     
}     

DataStream<String> topItems = windowedData     
    .keyBy("windowEnd")     
    .process(new TopNHotItems(3));  // 求点击量前3名的商品     

/** 求某个窗口中前 N 名的热门点击商品，key 为窗口时间戳，输出为 TopN 的结果字符串 */     
public static class TopNHotItems extends KeyedProcessFunction<Tuple, ItemViewCount, String> {     
     
  private final int topSize;     
     
  public TopNHotItems(int topSize) {     
    this.topSize = topSize;     
  }     
     
  // 用于存储商品与点击数的状态，待收齐同一个窗口的数据后，再触发 TopN 计算     
  private ListState<ItemViewCount> itemState;     
     
  @Override     
  public void open(Configuration parameters) throws Exception {     
    super.open(parameters);     
    // 状态的注册     
    ListStateDescriptor<ItemViewCount> itemsStateDesc = new ListStateDescriptor<>(     
        "itemState-state",     
        ItemViewCount.class);     
    itemState = getRuntimeContext().getListState(itemsStateDesc);     
  }     
     
  @Override     
  public void processElement(     
      ItemViewCount input,     
      Context context,     
      Collector<String> collector) throws Exception {     
     
    // 每条数据都保存到状态中     
    itemState.add(input);     
    // 注册 windowEnd+1 的 EventTime Timer, 当触发时，说明收齐了属于windowEnd窗口的所有商品数据     
    context.timerService().registerEventTimeTimer(input.windowEnd + 1);     
  }     
     
  @Override     
  public void onTimer(     
      long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {     
    // 获取收到的所有商品点击量     
    List<ItemViewCount> allItems = new ArrayList<>();     
    for (ItemViewCount item : itemState.get()) {     
      allItems.add(item);     
    }     
    // 提前清除状态中的数据，释放空间     
    itemState.clear();     
    // 按照点击量从大到小排序     
    allItems.sort(new Comparator<ItemViewCount>() {     
      @Override     
      public int compare(ItemViewCount o1, ItemViewCount o2) {     
        return (int) (o2.viewCount - o1.viewCount);     
      }     
    });     
    // 将排名信息格式化成 String, 便于打印     
    StringBuilder result = new StringBuilder();     
    result.append("====================================\n");     
    result.append("时间: ").append(new Timestamp(timestamp-1)).append("\n");     
    for (int i=0;i<topSize;i++) {     
      ItemViewCount currentItem = allItems.get(i);     
      // No1:  商品ID=12224  浏览量=2413     
      result.append("No").append(i).append(":")     
            .append("  商品ID=").append(currentItem.itemId)     
            .append("  浏览量=").append(currentItem.viewCount)     
            .append("\n");     
    }     
    result.append("====================================\n\n");     
     
    out.collect(result.toString());     
  }     
}     

topItems.print();     
env.execute("Hot Items Job");