Spark SQL案例：计算平均分

• 有多科成绩表，比如python.txt、spark.txt、django.txt，计算每个学生三科平均分

• Python成绩表 - python.txt

1 白小飞 90
2 洛银河 80
3 洛九天 85
4 李破晓 70
5 杨紫川 76
6 洛佩奇 77
7 花渔楼 79

• Spark成绩表 - spark.txt

1 白小飞 86
2 洛银河 96
3 洛九天 76
4 李破晓 82
5 杨紫川 91
6 洛佩奇 62
7 花渔楼 69

• Django成绩表 - django.txt

1 白小飞 80
2 洛银河 60
3 洛九天 76
4 李破晓 95
5 杨紫川 76
6 洛佩奇 82
7 花渔楼 86

• 期望输出结果如下所示：

• 启动HDFS与Spark
  

• 在HDFS上新建/score目录
  

• 在虚拟机上创建三个文件
  

• 我们的三个成绩文件上传到hdfs
  

• 查看三个文件内容
  

• 设置项目信息（项目名、保存位置、组编号、项目编号）
  
• 单击【Finish】按钮，将java目录改成scala目录
  

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

    4.0.0

    net.wfx.sql
    GradeTopNBySQL
    1.0-SNAPSHOT

    
        
            org.scala-lang
            scala-library
            2.11.8
        
        
            org.apache.spark
            spark-core_2.11
            2.1.1
        
        
            org.apache.spark
            spark-sql_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.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.sql包里创建CalculateAverageBySQL单例对象
  

package net.wfx.sql

import org.apache.spark.sql.{Dataset, SparkSession}


object CalculateAverageBySQL {
  def main(args: Array[String]): Unit = {
    // 创建或得到Spark会话对象
    val spark = SparkSession.builder()
      .appName("CalculateAverageBySQL")
      .master("local[*]")
      .getOrCreate()
    // 读取HDFS上“/score”目录里的成绩文件
    val lines: Dataset[String] = spark.read.textFile("hdfs://master:9000/score")
    // 导入隐式转换
    import spark.implicits._
    // 创建成绩数据集
    val gradeDS: Dataset[Grade] = lines.map(
      line => {
        val fields = line.split(" ")
        val id = fields(0).toInt
        val name = fields(1)
        val score = fields(2).toInt
        Grade(id, name, score)
      })
    // 将数据集转换成数据帧
    val df = gradeDS.toDF();
    // 基于数据帧创建临时表
    df.createOrReplaceTempView("t_grade")
    // 查询临时表，计算平均分
    val avg = spark.sql(
      """
        |select first(id) as id, name, cast(avg(score) as decimal(5, 2)) as average
        |   from t_grade
        |   group by name
        |   order by id
        |""".stripMargin
    )

    // 按照指定格式输出平均成绩
    println()
    avg.collect.foreach(row => println(row(0) + " " + row(1) + " " + row (2)))

    // 关闭Spark会话
    spark.close()
  }

  // 定义成绩样例类
  case class Grade(id: Int, name: String, score: Int)
}

• 在控制台查看输出结果