spark和flink中计算topN的方法

一、SPARK

　　其中top算子调用的takeOrdered算子，takeOrdered算子底层使用的是优先队列（BoundedPriorityQueue），首先进入的是mapPatition，然后使用reduce将每个分区数据进行合并

• sortBy + take
  

  val url: URL = Launcher.getClass.getClassLoader.getResource("word.dat")
  val lines: RDD[String] = sc.textFile(url.getPath)
  lines
     .flatMap(_.split("\\s+"))
     .map(_ -> 1)
     .reduceByKey(_ + _)
     .sortBy(_._2, false)
     .take(3)
     .foreach(println)

• top

  val url: URL = Launcher.getClass.getClassLoader.getResource("word.dat")
  val lines: RDD[String] = sc.textFile(url.getPath)
  lines
     .flatMap(_.split("\\s+"))
     .map(_ -> 1)
     .reduceByKey(_ + _)
     .top(3)(Ordering.by(o => o._2))
     .foreach(println)
          

• takeOrdered

  val url: URL = Launcher.getClass.getClassLoader.getResource("word.dat")
  val lines: RDD[String] = sc.textFile(url.getPath)
  lines
     .flatMap(_.split("\\s+"))
     .map(_ -> 1)
     .reduceByKey(_ + _)
     .takeOrdered(3)(Ordering.by(o => -o._2))
     .foreach(println)

二、FLINK

• TreeMap

  public class TopNWithStateStat {
      public static void main(String[] args) throws Exception {
  
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
          env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
          env.setParallelism(1);
  
          DataStreamSource lines = env.socketTextStream("node", 9999);
  
          SingleOutputStreamOperator> wordcount = lines
                  .flatMap(new FlatMapFunction>() {
                      @Override
                      public void flatMap(String line, Collector> collector) throws Exception {
                          String[] arr = line.split("\\s+");
                          for (String word : arr) {
                              collector.collect(Tuple2.of(word, 1));
                          }
                      }
                  })
                  .keyBy(t -> t.f0)
                  .window(SlidingProcessingTimeWindows.of(Time.minutes(5), Time.seconds(30)))
                  .sum(1);
  
          wordcount
                  .windowAll(TumblingProcessingTimeWindows.of(Time.seconds(30)))
                  .process(new TopNWithTreeMap(3))
                  .printToErr();
  
  
          env.execute();
      }
  
      public static class TopNWithTreeMap extends ProcessAllWindowFunction, String, TimeWindow> {
  
          private Integer n = 3;
  
          public TopNWithTreeMap(Integer n) {
              this.n = n;
          }
  
          @Override
          public void process(Context context, Iterable> iterable, Collector collector) throws Exception {
              TreeMap> treeMap = new TreeMap>(
                      new Comparator() {
                          @Override
                          public int compare(Integer y, Integer x) {
                              return (x < y) ? -1 : 1;
                          }
                      });
              Iterator> iterator = iterable.iterator();
              while (iterator.hasNext()) {
                  Tuple2 t = iterator.next();
                  treeMap.put(t.f1, t);
                  if (treeMap.size() > n) {
                      treeMap.pollLastEntry();
                  }
              }
              
              for (Map.Entry> entry : treeMap.entrySet()) {
                  Tuple2 value = entry.getValue();
                  collector.collect(String.format("%s:%d", value.f0, value.f1));
              }
          }
      }
  }

• 小顶堆

  public static class TopNWithQueue extends ProcessAllWindowFunction, String, TimeWindow> {
  
          private Integer n = 3;
  
          public TopNWithQueue(Integer n) {
              this.n = n;
          }
  
          @Override
          public void process(Context context, Iterable> iterable, Collector collector) throws Exception {
              PriorityQueue> priorityQueue = new PriorityQueue<>(new Comparator>() {
                  @Override
                  public int compare(Tuple2 o1, Tuple2 o2) {
                      return (o1.f1 < o2.f1) ? -1 : 1;
                  }
              });
              Iterator> iterator = iterable.iterator();
              while (iterator.hasNext()) {
                  priorityQueue.add(iterator.next());
                  if (priorityQueue.size() > n) {
                      priorityQueue.poll();
                  }
              }
              
              for (Tuple2 t : priorityQueue) {
                  collector.collect(String.format("%s-%s:%d", t.f0, t.f1));
              }
          }
      }