Spark 求TopN的优化思路

本文以求Top3为示范

首先想到的思路是整体排序后取出前三个，具体代码如下：

object TopN {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("TopN")
    val sc = new SparkContext(conf)
    val data: RDD[Int] = sc.makeRDD((10000 to 1 by -1).toArray)
    val sorted: Array[Int] = data.sortBy(item => item).take(3)
    print("[" + sorted.mkString(",") + "]")
    sc.stop()
  }
}

job运行时间：

这样在在数据量比较大的情况下，所有数据都集中到一个executor中，可能会导致该工作节点内存溢出。解决方案是先对数据进行分区，取出每个分区的前三个，然后再对所有分区的前三名进行整体排序，具体代码如下：

object TopN {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("TopN")
    val sc = new SparkContext(conf)
    val data: RDD[Int] = sc.makeRDD((10000 to 1 by -1).toArray)
    val sortedInPartition: RDD[Int] = data.mapPartitions(_.toArray.sorted.take(3).iterator)
    val sorted: Array[Int] = sortedInPartition.sortBy(item => item).take(3)
    print("[" + sorted.mkString(",") + "]")
    sc.stop()
  }
}

job运行时间：

由此可见，优化后的运行时间明显缩短了！！！