Spark RDD案例：分组排行榜

• 分组求TopN是大数据领域常见的需求，主要是根据数据的某一列进行分组，然后将分组后的每一组数据按照指定的列进行排序。最后取前每一组前N行数据。
• 有一组学生的成绩

• 张三 98
  李四 87
  王五 70
  赵六 85
  胡八一 68
  张三 88
  李四 79
  王五 85
  赵六 77
  胡八一 70
  张三 89
  李四 90
  王五 83
  赵六 73
  胡八一 93
  张三 70
  李四 75
  王五 69
  赵六 80
  胡八一 79
  

• 同一个学生有多门成绩，现需要计算每个学生分数最高的前3个成绩，期望输出结果如下所示：
• 1
• 预备工作：启动集群的HDFS与Spark
• 二、完成任务 （一）新建Maven项目
• 设置项目信息

 增加目录scala

• 在pom.xml添加依赖

• 
  
      4.0.0
  
      net.cch.rdd
      GradeTopN
      1.0-SNAPSHOT
  
      
          8
          8
      
  
      
          
              org.scala-lang
              scala-library
              2.12.15
          
          
              org.apache.spark
              spark-core_2.12
              2.4.4
          
      
      
          src/main/scala
          
              
                  org.apache.maven.plugins
                  maven-assembly-plugin
                  3.3.0
                  
                      
                          jar-with-dependencies
                      
                  
                  
                      
                          make-assembly
                          package
                          
                              single
                          
                      
                  
              
              
                  net.alchim31.maven
                  scala-maven-plugin
                  3.3.2
                  
                      
                          scala-compile-first
                          process-resources
                          
                              add-source
                              compile
                          
                      
                      
                          scala-test-compile
                          process-test-resources
                          
                              testCompile
                          
                      
                  
              
          
      
  
  

   

  （三）创建日志属性文件
• 添加log4j.properties日志文件

• log4j.rootLogger=ERROR, stdout, logfile
  log4j.appender.stdout=org.apache.log4j.ConsoleAppender
  log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
  log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
  log4j.appender.logfile=org.apache.log4j.FileAppender
  log4j.appender.logfile.File=target/spark.log
  log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
  log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
  

   

  （四）创建分组排行榜单例对象
• 创建GradeTopN单例对象

package net.cch.rdd

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


object GradeTopN {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[*]")   //设置主节点位置
      .setAppName("GradeTopN")// 设置应用名称

    // 基于Spark配置对象创建Spark容器
    val sc = new SparkContext(conf)
    // 实现分组排行榜
    val topN = sc.textFile("hdfs://master:9000/input/grades.txt")
      .map(
        line => {
          val fields = line.split(" ")
          (fields(0),fields(1))
        }
      ) // 将每行成绩映射成二元组(name,grade)
      .groupByKey()
      .map(item =>{
        val name = item._1
        val top3 = item._2.toList.sortWith(_ > _).take(3)
        (name,top3)
      }) // 值排序，取前三

    //输出
    topN.collect.foreach(
      line => {
        val name = line._1
        val top2 = line._2
        var score = ""
        for(elem <- top2){
          score = score + " " + elem
        }
        println(name + ":" + score)
      }
    )

    //停止Spark容器，结束任务
    sc.stop()
  }
}

• 在控制台查看输出结果
  

查看结果

4、对这个二维组进行分组排序

对成绩进行排序取前三

5、输出结果