分组排行榜RDD

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

-设置项目信息（项目名、保存为之、组编号、项目编号）

-单机【Finish】按钮

-将java改成scala

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

    4.0.0

    net.wfx.rdd
    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.properties

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

-在net.wfx.rdd包里创建GradeTopN单例对象

package net.wfx.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)
        }) // 值排序，取前三

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

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

执行命令：val lines = sc.textFile("hdfs://master:9000/input/grades.txt")

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)
    })

-执行上述代码