Spark基础学习笔记30：Spark SQL案例分析

• 零、本讲学习目标
• 一、使用Spark SQL实现词频统计
• • （一）数据源 - words.txt
  • （二）创建Maven项目
  • （三）添加依赖和构建插件
  • （四）修改源目录名称
  • （五）创建日志属性文件
  • （六）创建词频统计单例对象
  • （七）启动程序，查看结果
  • （八）词频统计数据转化流程图

1. 使用Spark SQL实现词频统计
2. 掌握Spark SQL与Hive整合
3. 掌握Spark SQL读写MySQL
4. 完成Spark热点搜索词统计
5. Spark SQL智慧交通数据分析

• 创建Maven项目 - SparkSQLWordCount
  

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

    4.0.0

    net.hw.wc
    SparkSQLWordCount
    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
        
    
            
        
            
                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
                        
                    
                
            
        
    

• 将源目录名由java改成scala
  
• 在pom.xml文件里，设置源目录
  

• 在resources目录里创建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.hw.wc包，在包里创建SparkSQLWordCount单例对象
  

package net.hw.wc

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


object SparkSQLWordCount {
  def main(args: Array[String]): Unit = {
    // 设置HADOOP用户名属性，否则本地运行访问会被拒绝
    System.setProperty("HADOOP_USER_NAME", "root")
    // 创建或得到SparkSession
    val spark = SparkSession.builder()
      .appName("SparkSQLWordCount")
      .master("local[*]")
      .getOrCreate()
    // 读取HDFS上的单词文件
    val lines: Dataset[String] = spark.read.textFile("hdfs://master:9000/input/words.txt")
    // 显示数据集lines内容
    lines.show()
    // 导入Spark会话对象的隐式转换
    import spark.implicits._
    // 将数据集中的数据按空格切分并合并
    val words: Dataset[String] = lines.flatMap(_.split(" "))
    // 显示数据集words内容
    words.show()
    // 将数据集默认列名由value改为word，并转换成数据帧
    val df = words.withColumnRenamed("value", "word").toDF()
    // 显示数据帧内容
    df.show()
    // 基于数据帧创建临时视图
    df.createTempView("v_words")
    // 执行SQL分组查询，实现词频统计
    val wc = spark.sql(
      """
        | select word, count(*) as count
        |    from v_words group by word
        |    order by count desc
        |""".stripMargin)
    // 显示词频统计结果
    wc.show()
    // 关闭会话
    spark.close()
  }
}

• 运行SparkSQLWordCount单例对象
  

• 文本文件，转化成数据集，再转化成数据帧，最后基于表查询得到结果数据帧