Spark 实现分组topn排序（scala版本）

四种方法实现分组排序

数据集格式：

http://bigdata.edu360.cn/laoduan
http://bigdata.edu360.cn/laoduan
http://javaee.edu360.cn/xiaoxu
http://javaee.edu360.cn/xiaoxu
http://javaee.edu360.cn/laoyang
http://javaee.edu360.cn/laoyang
http://javaee.edu360.cn/laoyang

按照每个学科求老师访问量排序结果

方案1：先reduceByKey进行聚合，再按照学科进行分组，在每个分组内进行排序

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * 求每个学科的最受欢迎的老师方法1
 *
 * 先reduceByKey进行聚合，再按照学科进行分组，在分组内进行排序
 *
 * 缺点：分组内数据用迭代器调用，当数据量较大时，迭代器的数据放到内存中会内存溢出
 *    （分组内数据进行排序，调用的是集合上的排序方法，不是RDD的，集合上的数据需要一次性加载到内存中）
 */
object GroupFavoriteTeacher1 {

  def main(args: Array[String]): Unit = {
    val favTeacher: SparkConf = new SparkConf().setMaster("local[2]").setAppName("FavTeacher")

    val context: SparkContext = new SparkContext(favTeacher)

    // 读取一行
    // 数据格式： http://bigdata.edu360.cn/laozhang
    val lines: RDD[String] = context.textFile("./teacher.log")

    // 切分形成（（学科，老师），1）
    val sbjectTeacher: RDD[((String, String), Int)] = lines.map(line => {
      val strings: Array[String] = line.split("/")
      val sbject: String = strings(2).split("\\.")(0) // 学科
      val teacher = strings(3) // 老师

      ((sbject, teacher), 1)
    })

    // 聚合相成 （（学科，老师）, n）， 相同（学科，老师）仅保留一条记录
    val groupTeahcerReduce: RDD[((String, String), Int)] = sbjectTeacher.reduceByKey((a, b) => {
      a + b
    })

    // 按照学科进行分组,【并指定分组后RDD的分区数量】  分组条件相同的数据会在一个分组中，一个分区内可以有多个分组
    // ------经过分组后，相同学科的数据（一个分组内的数据）可以用一个迭代器调用
    val group: RDD[(String, Iterable[((String, String), Int)])] = groupTeahcerReduce.groupBy(
      (t:((String, String), Int))=> t._1._1,2)

   // 将每个组内的数据的迭代器拿出（一个迭代器对应一个分组），按照n进行排序
   val value: RDD[(String, List[((String, String), Int)])] = group.mapValues((it) => {
     it.toList.sortBy(u => {
       u._2
     }).reverse
   })

    // action算子进行驱动
   // value.foreach(u=>println(u._1 +":"+u._2))
     value.foreach(u=>{
      print(u._1+":")
       u._2.foreach(it=>{print(it._1._2+"?"+it._2)})
       println()
     })

    context.stop()

  }

}

方案2：按照学科进行过滤，每次对一个学科的数据进行排序

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

/**
 * 求每个学科的最受欢迎的老师方法2
 *
 * 按照学科进行过滤，每次过滤出一个学科的老师放入一个RDD中，再对过滤出的RDD进行排序，
 * RDD排序方法是内存+磁盘的
 *
 * 缺点：需要多次过滤，效率低
 */
object GroupFavoriteTeacher2 {

  def main(args: Array[String]): Unit = {
    val favTeacher: SparkConf = new SparkConf().setMaster("local[2]").setAppName("FavTeacher")

    val context: SparkContext = new SparkContext(favTeacher)

    // 读取一行
    val lines: RDD[String] = context.textFile("./teacher.log")  //  http://bigdata.edu360.cn/laozhang

    // 切分形成（（学科，老师），1）
    val sbjectTeacher: RDD[((String, String), Int)] = lines.map(line => {
      val strings: Array[String] = line.split("/")
      val sbject: String = strings(2).split("\\.")(0) // 学科
      val teacher = strings(3) // 老师

      ((sbject, teacher), 1)
    })

    // 分组聚合 （（学科，老师），n）
    val reduced: RDD[((String, String), Int)] = sbjectTeacher.reduceByKey((a, b) => {
      a + b
    })

    // 获取所有的学科集合
    val keys: RDD[(String, String)] = reduced.keys
    val subjects: RDD[String] = keys.map(u => {
      u._1
    })

    // 这里需要用一个action算子驱动subjects，并放到集合中，遍历集合中数据
    // 因为不能在一个RDD上的transformation里面调用其他的transformation
    val strings: Array[String] = subjects.collect()

    // 遍历每个学科，一次过滤出一个学科的所有老师进行排序
    for(sb<-strings){

      // 过滤出的数据放到一个RDD中，该RDD中的数据学科相同
      val filtered: RDD[((String, String), Int)] = reduced.filter(it => {
        it._1._1==sb
      })

      // 将每次过滤出的相同学科的数据进行排序
      // RDD排序，内存加磁盘进行排序
      val sorted: RDD[((String, String), Int)] = filtered.sortBy(it => {
        it._2
      }, false)

      // action算子进行驱动
      sorted.foreach(u=>{println(u)})

    }


  context.stop()



  }
}

方案3：自定义分区器，将相同学科的数据放在同一个分区内，再对每个分区进行排序

import org.apache.spark.{Partitioner, SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

import scala.collection.mutable

/**
 * 求每个学科的最受欢迎的老师方法3
 *
 * 自定义分区器实现分组聚合，先按照（key，vale）：（（学科，老师），1）进行聚合
 * 然后按照学科进行分区，将相同学科的数据放在同一分区。然后针对每个分区中的数据按照n进行排序
 *
 *缺点：集合上进行排序，会内存溢出；
 *     shuffle次数较多，速度慢，  reduceByKey、partitionBy需要shuffle
 */
object GroupFavoriteTeacher3 {

  def main(args: Array[String]): Unit = {
    val favTeacher: SparkConf = new SparkConf().setMaster("local[3]").setAppName("FavTeacher")

    val context: SparkContext = new SparkContext(favTeacher)

    // 读取一行
    val lines: RDD[String] = context.textFile("./teacher.log")  //  http://bigdata.edu360.cn/laozhang

    // 切分形成（（学科，老师），1）
    val sbjectTeacher: RDD[((String, String), Int)] = lines.map(line => {
      val strings: Array[String] = line.split("/")
      val sbject: String = strings(2).split("\\.")(0) // 学科
      val teacher = strings(3) // 老师

      ((sbject, teacher), 1)
    })

    // 分组聚合 （（学科，老师），n）
    val reduced: RDD[((String, String), Int)] = sbjectTeacher.reduceByKey((a, b) => {
      a + b
    })

    // 获取所有的学科集合
    val keys: RDD[(String, String)] = reduced.keys
    val subjects: RDD[String] = keys.map(u => {
      u._1
    })

    // 这里需要用一个action算子驱动subjects，，并放到集合中，遍历集合中数据
    // 因为不能在一个RDD上的transformation里面调用其他的transformation
    val strings: Array[String] = subjects.collect()

    // 自定义一个分区器，按照学科进行分区
    val sbPartitioner: SubjectPartitioner = new SubjectPartitioner(strings)
    val paritioner: RDD[((String, String), Int)] = reduced.partitionBy(sbPartitioner)

    // 在每个分区内排序，调用的是集合上的排序，
    // mapPartitions算子，每次拿一个分区中数据，（用迭代器获取）
    val sorted: RDD[((String, String), Int)] = paritioner.mapPartitions((it) => {
      it.toList.sortBy(u => {
        u._2
      }).reverse.iterator
    })

    // action算子驱动
    sorted.foreach(u=>{println(u)})


    context.stop()
  }

  /**
   * 自定义分区器，按照学科进行分区
   */
  class SubjectPartitioner( subjects :Array[String]) extends  Partitioner{
    
    // 设置分区规则，设置每个学科对应的分区ID，用map存。
    // 不能用subject.hashMap % numPartitions，这样不能保证每个分区内只有一个学科的数据
    val rules: mutable.HashMap[String, Int] = new mutable.HashMap[String, Int]()
    var i = 0;
    for(sb <- subjects){
      rules.put(sb, i)  // rules（sb） = i
      i = i+1
    }

    // 返回分区的数量（下一个RDD有多少个分区）
    override def numPartitions: Int = subjects.length

    // 根据传入的key计算该key所在的分区编号
    override def getPartition(key: Any): Int ={
      val subject: String = key.asInstanceOf[(String, String)]._1  //强转格式
       rules.getOrElse(subject,0)
    }
  }



}

方案4：在ReduceByKey的同时调用自定义分区器，减少shuffle次数，提高效率

import org.apache.spark.{Partitioner, SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

import scala.collection.mutable

/**
 *求每个学科的最受欢迎的老师方法4
 *
 * 在reduceByKey的同时按照学科进行分区，减少shuffle的次数
 *
 * 缺点：集合上进行排序，会内存溢出
 */
object GroupFavoriteTeacher4 {

  def main(args: Array[String]): Unit = {
    val favTeacher: SparkConf = new SparkConf().setMaster("local[3]").setAppName("FavTeacher")

    val context: SparkContext = new SparkContext(favTeacher)

    // 读取一行
    val lines: RDD[String] = context.textFile("./teacher.log")  //  http://bigdata.edu360.cn/laozhang

    // 切分形成（（学科，老师），1）
    val sbjectTeacher: RDD[((String, String), Int)] = lines.map(line => {
      val strings: Array[String] = line.split("/")
      val sbject: String = strings(2).split("\\.")(0) // 学科
      val teacher = strings(3) // 老师

      ((sbject, teacher), 1)
    })

    // 获取所有学科的集合
    val subjects: Array[String] = sbjectTeacher.map(u => {
      u._1._1
    }).distinct().collect()

    // reduceByKey的同时按照指定分区器进行分区
    // 该RDD的一个分区内仅有一个学科的数据
    val partitioner: SubjectPartitioner = new SubjectPartitioner(subjects)
    val reduced: RDD[((String, String), Int)] = sbjectTeacher.reduceByKey(partitioner, (a, b) => {
      a + b
    })

    // 在每个分区内排序，调用的是集合上的排序，
    // mapPartitions算子，每次拿一个分区中数据，（用迭代器获取）
    val sorted: RDD[((String, String), Int)] = reduced.mapPartitions(it => {
      it.toList.sortBy(u => {
        u._2
      }).reverse.iterator
    })

    // action算子
    sorted.foreach(u=>{println(u)})


    context.stop()
  }


  /**
   * 自定义分区器，按照学科进行分区
   */
  class SubjectPartitioner( subjects :Array[String]) extends  Partitioner{

    // 设置分区规则，设置每个学科对应的分区ID，用map存。
    // 不能用subject.hashMap % numPartitions，这样不能保证每个分区内只有一个学科的数据
    val rules: mutable.HashMap[String, Int] = new mutable.HashMap[String, Int]()
    var i = 0;
    for(sb <- subjects){
      rules.put(sb, i)  // rules（sb） = i
      i = i+1
    }

    // 返回分区的数量（下一个RDD有多少个分区）
    override def numPartitions: Int = subjects.length

    // 根据传入的key计算该key所在的分区编号
    override def getPartition(key: Any): Int ={
      val subject: String = key.asInstanceOf[(String, String)]._1  //强转格式
      rules.getOrElse(subject,0)
    }
  }



}

原文链接：https://blog.csdn.net/qq_29672495/article/details/107048512

你可能也喜欢