Spark RDD:分组排行榜

@

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

唐敏 78
黄霞 75
黄霞 72
唐敏 65
萧平 85
王静 68
王静 78
萧平 86
王静 94
萧平 84
黄霞 62
唐敏 88

• 同一个同学有多门成绩，现在需要计算每个学生的分数最高的前三个成绩，期望输出结果如下所示：

王静: 94 78 68
萧平: 86 85 84
唐敏: 88 78 65

• 实现思路：使用Spark RDD的groupByKey()算子可以对(key, value)形式的RDD按照key进行分组，key相同的元素的value将聚合到一起，形成(key, value-list)，将value-list中的元素降序排列取前N个即可。

• 预备工作：启动集群的HDFS与Spark
  
• 将成绩文件 - grades.txt上传到HDFS上/input目录
  

• 设置项目信息（项目名、保存位置、组编号、项目编号）
  

• 单击【Finish】按钮
  

• 将java目录改成scala目录
  

• 在pom.xml文件里添加依赖与Maven构建插件
  

    4.0.0

    groupId
    GradeTopN
    1.0-SNAPSHOT

    
        
            org.scala-lang
            scala-library
            2.11.12
        
        
            org.apache.spark
            spark-core_2.11
            2.1.1
        
    
    
        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.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

• 在net.Lee.rdd包里创建GradeTopN单例对象
  

package net.Lee.rdd

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


object GradeTopN {
  def main(args: Array[String]): Unit = {
    // 创建Spark配置对象
    val conf = new SparkConf()
      .setAppName("GradeTopN") // 设置应用名称
      .setMaster("local[*]") // 设置主节点位置（目前本地调试）
    // 基于Spark配置对象创建Spark容器
    val sc = new SparkContext(conf);
    // 实现分组排行榜
    val top3 = 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)
      }) // 值排序，取前三

    // 输出分组排行榜结果
    top3.collect.foreach(line => {
      val name = line._1
      var scores = ""
      for (score <- line._2)
        scores = scores + " " + score
      println(name + ":" + scores)
    })

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

• 在控制台查看输出结果
  

• 输入spark-shell然后 执行命令：val lines = sc.textFile("hdfs://master:9000/input/grades.txt") lines.collect.foreach(println)

val grades = lines.map(line => {
          val fields = line.split(" ")
          (fields(0), fields(1))
        })

• 执行上述代码

• 执行命令：val groupGrades = grades.groupByKey()

val top3 = groupGrades.map(item => {
          val name = item._1
          val top3 = item._2.toList.sortWith(_ > _).take(3)
          (name, top3)
        })

• 执行上述代码
  

top3.collect.foreach(line => {
      val name = line._1
      var scores = ""
      for (score <- line._2)
        scores = scores + " " + score
      println(name + ":" + scores)
    })

• 执行上述代码