SparkSQL手册.md

IT-风油精 517 2023-08-27

第一章 SparkSQL概述

1.1 SparkSQL是什么

​ Spark SQL 是 Spark 用于结构化数据(structured data)处理的 Spark 模块。

1.2 Hive and SparkSQL

​ SparkSQL的前身是Shark，给熟悉EDBMS但又不理解MapReduce的技术人员提供快速上手的工具。

​ Hive是早期唯一运行在Hadoop上的SQL-ON-Hadoop工具。但是MapReduce计算过程中大量的中间磁盘落地过程消耗了大量的I/O，降低的运行效率，为了提高SQL-ON-Hadoop的效率，大量的SQL-ON-Hadoop工具开始产生，其中表现较为突出的是：

  • Drill
  • Impala
  • Shark

​ 其中Shark是伯克利实验室Spark生态环境的组件之一，是基于Hive所开发的工具，它修改了下图所示的右下角的内存管理，物理计划，执行三个模块，并使之能运行在Spark引擎上。Shark 的出现，使得 SQL-on-Hadoop 的性能比 Hive 有了 10-100 倍的提高。

​ 2014 年 6 月 1 日 Shark 项目和 SparkSQL 项目的主持人 Reynold Xin 宣布：停止对 Shark 的开发，团队将所有资源放 SparkSQL 项目上，至此，Shark 的发展画上了句话，但也因此发 展出两个支线：SparkSQL 和 Hive on Spark。

​ 其中 SparkSQL 作为 Spark 生态的一员继续发展，而不再受限于 Hive，只是兼容 Hive；而Hive on Spark 是一个 Hive 的发展计划，该计划将 Spark 作为Hive 的底层引擎之一，也就是说，Hive 将不再受限于一个引擎，可以采用 Map-Reduce、Tez、Spark 等引擎。

​ 对于开发人员来讲，SparkSQL 可以简化 RDD 的开发，提高开发效率，且执行效率非常快，所以实际工作中，基本上采用的就是SparkSQL。Spark SQL 为了简化 RDD 的开发， 提高开发效率，提供了 2 个编程抽象，类似 Spark Core 中的 RDD

  • DataFrame
  • DataSet

1.3 SparkSQL特点

  • 易整合
  • 统一的数据访问
  • 兼容Hive
  • 标准数据连接

1.4 DataFrame介绍

​ 在 Spark 中，DataFrame 是一种以 RDD 为基础的分布式数据集，类似于传统数据库中的二维表格。DataFrame 与 RDD 的主要区别在于，前者带有 schema 元信息，即 DataFrame所表示的二维表数据集的每一列都带有名称和类型。这使得 Spark SQL 得以洞察更多的结构信息，从而对藏于 DataFrame 背后的数据源以及作用于 DataFrame 之上的变换进行了针对性的优化，最终达到大幅提升运行时效率的目标。反观 RDD，由于无从得知所存数据元素的具体内部结构，Spark Core 只能在 stage 层面进行简单、通用的流水线优化。

​ 同时，与 Hive 类似，DataFrame 也支持嵌套数据类型（struct、array 和 map）。从 API 易用性的角度上看，DataFrame API 提供的是一套高层的关系操作，比函数式的 RDD API 要更加友好，门槛更低。

image-20230812161546257

​ 左侧的 RDD[Person]虽然以 Person 为类型参数，但 Spark 框架本身不了解 Person 类的内部结构。而右侧的 DataFrame 却提供了详细的结构信息，使得 Spark SQL 可以清楚地知道该数据集中包含哪些列，每列的名称和类型各是什么。

​ DataFrame 是为数据提供了 Schema 的视图。可以把它当做数据库中的一张表来对待DataFrame 也是懒执行的，但性能上比 RDD 要高，主要原因：优化的执行计划，即查询计划通过 Spark catalyst optimiser 进行优化。

1.5 DataSet介绍

​ DataSet是分布式数据集合。DataSet是Spark1.6中的新抽象，是DataFrame的一个拓展。它提供了RDD的优势（强类型，使用强大的lambda函数的能力）以及SparkSQL优化执行引擎的优点。DataSet也可以使用此功能性的转换（如map，flatMap，filter等等）。

  • DataSet是DataFrame API的一个拓展，是SparkSQL最新的数据抽象；
  • 友好的 API 风格，既具有类型安全检查也具有 DataFrame 的查询优化特性；
  • 用样例类来对 DataSet 中定义数据的结构信息，样例类中每个属性的名称直接映射到DataSet 中的字段名称；
  • DataSet 是强类型的。比如可以有 DataSet[Car]，DataSet[Person]；
  • DataFrame 是 DataSet 的特列，DataFrame=DataSet[Row] ，所以可以通过 as 方法将DataFrame 转换为 DataSet。Row 是一个类型，跟 Car、Person 这些的类型一样，所有的表结构信息都用 Row 来表示。获取数据时需要指定顺序。

第二章 SparkSQL编程

2.1 不同之处

​ SparkCore编程中，需要先构建SparkContext上下文，SparkSQL其实是对SparkCore的一次封装。

  • 旧版SparkSQL，提供两种SQL查询起始点：SQLContext，用于Spark自己提供的SQL查询；HiveContext，用于连接Hive查询。
  • 新版SparkSQL，提供 SparkSession，实质上是 SQLContext 和 HiveContext 的组合，所以在 SQLContex 和 HiveContext 上可用的 API 在 SparkSession 上同样是可以使用的。

image-20230812194833167

2.2 DataFrame

​ Spark SQL 的 DataFrame API 允许我们使用 DataFrame 而不用必须去注册临时表或者生成 SQL 表达式。DataFrame API 既有 transformation 操作也有 action 操作。

2.2.1 创建DataFrame

（1）从Spark数据源进行创建

查看Spark支持创建文件的数据源格式（按tab提示可以看到支持的文件类型）

scala> spark.read.
csv      jdbc   load     options   parquet   table   textFile
format   json   option   orc       schema    text

在Spark的bin/data目录下创建测试数据user.json文件（存放数据的路径可以随便，只要路径写对即可）

{"username": "zhangsan","age": 20}
{"username": "lisi","age": 30}
{"username": "wangwu","age": 40}
{"username": "laoliu","age": 60}
{"username": "aqi","age": 17}

读取json文件创建DataFrame

scala> val df = spark.read.json("datas/user.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, username: string]

scala> df.show()
+---+--------+
|age|username|
+---+--------+
| 20|zhangsan|
| 30|    lisi|
| 40|  wangwu|
| 60|  laoliu|
| 17|     aqi|
+---+--------+

注意：如果从内存中获取数据，spark 可以知道具体的数据类型。如果是数字，默认作为 Int 处理；但是从文件中读取的数字，不能确定是什么类型，所以用 bigint 接收，可以和Long类型转换，但是和 Int 不能进行转换。

（2）从RDD进行转换

（3）从Hive Table 进行查询返回

2.2.2 SQL语法

​ SQL 语法风格是指我们查询数据的时候使用 SQL 语句来查询，这种风格的查询必须要有临时视图或者全局视图来辅助。

（1）读取 JSON 文件创建一个临时表

scala> val df = spark.read.json("..//data//user.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, username: string]

（2）DataFrame创建一个临时表（仅Session范围内有效）

scala> df.createOrReplaceTempView("user")

（3）通过SQL语句实现查询全表并展示

scala> val sql_df = spark.sql("select * from user")
sql_df: org.apache.spark.sql.DataFrame = [age: bigint, username: string]

scala> sql_df.show
+---+--------+
|age|username|
+---+--------+
| 20|zhangsan|
| 30|    lisi|
| 40|  wangwu|
| 60|  laoliu|
| 17|     aqi|
+---+--------+

（4）DataFrame创建一个全局表

scala> df.createGlobalTempView("globalUser")

（5）通过SQL语句实现查询全表

scala> spark.sql("SELECT * FROM global_temp.globalUser").show()
scala> spark.newSession().sql("SELECT * FROM global_temp.globalUser").show()

2.2.3 DSL语法

​ DataFrame 提供一个特定领域语言(domain-specific language, DSL)去管理结构化的数据。可以在 Scala, Java, Python 和 R 中使用 DSL，使用 DSL 语法风格不必去创建临时视图了。

（1）创建一个 DataFrame

scala> val df = spark.read.json("..//data//user.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, username: string]

（2）查看dataFrame的Scheme信息

scala> df.printSchema
root
 |-- age: long (nullable = true)
 |-- username: string (nullable = true)

（3）只查看“username”列数据

scala> df.select("username").show
+--------+
|username|
+--------+
|zhangsan|
|    lisi|
|  wangwu|
|  laoliu|
|     aqi|
+--------+

（4）查看“username”列数据以及“age+1”数据

注意:涉及到运算的时候, 每列都必须使用$, 或者采用引号表达式：单引号+字段名

scala> df.select($"username",$"age"+1).show
+--------+---------+
|username|(age + 1)|
+--------+---------+
|zhangsan|       21|
|    lisi|       31|
|  wangwu|       41|
|  laoliu|       61|
|     aqi|       18|
+--------+---------+

scala> df.select('username,'age+1).show()
+--------+---------+
|username|(age + 1)|
+--------+---------+
|zhangsan|       21|
|    lisi|       31|
|  wangwu|       41|
|  laoliu|       61|
|     aqi|       18|
+--------+---------+

scala> df.select('username,'age+1 as "newage").show
+--------+------+
|username|newage|
+--------+------+
|zhangsan|    21|
|    lisi|    31|
|  wangwu|    41|
|  laoliu|    61|
|     aqi|    18|
+--------+------+

（5）查看“age”大于30的数据

scala> df.filter('age>30).show
+---+--------+
|age|username|
+---+--------+
| 40|  wangwu|
| 60|  laoliu|
+---+--------+

（6）按照“age”分组，查看数据条数

scala> df.groupBy('age).count.show
+---+-----+
|age|count|
+---+-----+
| 17|    1|
| 30|    1|
| 20|    1|
| 60|    1|
| 40|    1|
+---+-----+

2.2.4 RDD转换为DataFrame

注意：RDD 和 DataFrame 或 DataSet 之间转换，需要引入

import  sparkSession.implicits._

创建的 sparkSession 对象不能使用 var 声明，因为 Scala 只支持 val 修饰的对象的引入。在spark-shell 中无需导入，自动完成此操作。在实际开发中，一般通过样例类将 RDD 转换为 DataFrame。

scala> val RDD1 = sc.textFile("..//data/test.txt").toDF.show
+-----------+
|      value|
+-----------+
|   name age|
|zhangsan 16|
|    lisi 20|
|  wangwu 24|
+-----------+

scala> case class User(name:String, age:Int)
defined class User

scala> sc.makeRDD(List(("zhangsan",30),("lisi",40))).map(t=>User(t._1,t._2)).toDF.show
+--------+---+
|    name|age|
+--------+---+
|zhangsan| 30|
|    lisi| 40|
+--------+---+

2.2.5 DataFrame 转换为 RDD

DataFrame 其实就是对 RDD 的封装，所以可以直接获取内部的 RDD

scala> case class User(name:String,age:Int)
defined class User

scala> val df = sc.makeRDD(List(("zhangsan",30),("lisi",40))).map(t => User(t._1,t._2)).toDF
df: org.apache.spark.sql.DataFrame = [name: string, age: int]

scala> val rdd = df.rdd
rdd: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[6] at rdd at <console>:25

scala> val array = rdd.collect
array: Array[org.apache.spark.sql.Row] = Array([zhangsan,30], [lisi,40])

scala> array(0)(1)
res0: Any = 30

scala> array(0)(0)
res1: Any = zhangsan

scala> array(0).getAs[String]("name")
res3: String = zhangsan

2.3 DataSet

DataSet 是具有强类型的数据集合，需要提供对应的类型信息。

2.3.1 创建DataSet

注意：在实际使用的时候，很少用到把序列转换成DataSet，更多的是通过RDD来得到DataSet

（1）使用样例类序列创建DataSet

scala> case class Person(name:String,age:Long)
defined class Person

scala> val caseClassDS = Seq(Person("zhangsan",25)).toDS
caseClassDS: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]

scala> caseClassDS.show
+--------+---+
|    name|age|
+--------+---+
|zhangsan| 25|
+--------+---+

（2）使用基本类型的序列创建DataSet

scala> val ds = Seq("1","2","3","4","5","zhangsan","lisi").toDS
ds: org.apache.spark.sql.Dataset[String] = [value: string]

scala> ds.show
+--------+
|   value|
+--------+
|       1|
|       2|
|       3|
|       4|
|       5|
|zhangsan|
|    lisi|
+--------+

2.3.2 RDD 转换为 DataSet（重点，常用）

​ SparkSQL能自动将包含case类的RDD转换成DataSet，case类定义了table的结构，case类属性通过反射变成了表的列名。case类可以包含诸如Seq或者Array等复杂的结构。

scala> sc.makeRDD(List(("zhangsan",33),("luoxiang",50))).map
map   mapPartitions   mapPartitionsWithIndex   mapValues

scala> sc.makeRDD(List(("zhangsan",33),("luoxiang",50))).map(t=>User(t._1,t._2)).toDS
res2: org.apache.spark.sql.Dataset[User] = [name: string, age: int]

2.3.3 DataSet 转化为 RDD

​ DataSet 其实也是对 RDD 的封装，所以可以直接获取内部的 RDD。

scala> case class User(name:String,age:Int)
defined class User

scala> val ds1 = sc.makeRDD(List(("zhangsan",30),("luoxiang",50))).map(t=>User(t._1,t._2)).toDS
ds1: org.apache.spark.sql.Dataset[User] = [name: string, age: int]

scala> val rdd1 = ds1.rdd
rdd1: org.apache.spark.rdd.RDD[User] = MapPartitionsRDD[6] at rdd at <console>:25

scala> rdd1.collect
res3: Array[User] = Array(User(zhangsan,30), User(luoxiang,50))

2.3.4 DataFrame 和 DataSet转换

DataFrame 其实是 DataSet 的特例，所以它们之间是可以互相转换的。

（1）DataFrame 转换为 DataSet

scala> case class User(name:String,age:Int)
defined class User

scala> val df = sc.makeRDD(List(("zhansgan",30),("luoxiang",50))).toDF("name","age")
df: org.apache.spark.sql.DataFrame = [name: string, age: int]

scala> val ds = df.as[User]
ds: org.apache.spark.sql.Dataset[User] = [name: string, age: int]

scala> ds.show
+--------+---+
|    name|age|
+--------+---+
|zhansgan| 30|
|luoxiang| 50|
+--------+---+

（2）DataSet 转换为 DataFrame

scala> val df1 = ds1.toDF
df1: org.apache.spark.sql.DataFrame = [name: string, age: int]

scala> df1.show
+--------+---+
|    name|age|
+--------+---+
|zhangsan| 30|
|luoxiang| 50|
+--------+---+

2.4 RDD-DataFrame-DataSet的关系

在 SparkSQL 中 Spark 为我们提供了两个新的抽象，分别是 DataFrame 和 DataSet。他们和 RDD 有什么区别呢？首先从版本的产生上来看：

  • Spark1.0 => RDD

  • Spark1.3 => DataFrame

  • Spark1.6 => Dataset

如果同样的数据都给到这三个数据结构，他们分别计算之后，都会给出相同的结果。不同是的他们的执行效率和执行方式。在后期的 Spark 版本中，DataSet 有可能会逐步取代 RDD和 DataFrame 成为唯一的 API 接口。

2.4.1 三者的共性

  • RDD、DataFrame、DataSet 全都是 spark 平台下的分布式弹性数据集，为处理超大型数据提供便利;
  • 三者都有惰性机制，在进行创建、转换，如 map 方法时，不会立即执行，只有在遇到Action 如 foreach 时，三者才会开始遍历运算;
  • 三者有许多共同的函数，如 filter，排序等;
  • 在对 DataFrame 和 Dataset 进行操作许多操作都需要这个包:import spark.implicits._（在创建好 SparkSession 对象后尽量直接导入）
  • 三者都会根据 Spark 的内存情况自动缓存运算，这样即使数据量很大，也不用担心会内存溢出
  • 三者都有 partition 的概念
  • DataFrame 和 DataSet 均可使用模式匹配获取各个字段的值和类型

2.4.2 三者的区别

（1）RDD

  • RDD 一般和 spark mllib 同时使用
  • RDD 不支持 sparksql 操作

（2）DataFrame

  • 与 RDD 和 Dataset 不同，DataFrame 每一行的类型固定为 Row，每一列的值没法直接访问，只有通过解析才能获取各个字段的值
  • DataFrame 与 DataSet 一般不与 spark mllib 同时使用
  • DataFrame 与 DataSet 均支持 SparkSQL 的操作，比如 select，groupby 之类，还能注册临时表/视窗，进行 sql 语句操作
  • DataFrame 与 DataSet 支持一些特别方便的保存方式，比如保存成 csv，可以带上表头，这样每一列的字段名一目了然

（3）DataSet

  • Dataset 和 DataFrame 拥有完全相同的成员函数，区别只是每一行的数据类型不同。DataFrame 其实就是 DataSet 的一个特例 type DataFrame = Dataset[Row]
  • DataFrame 也可以叫 Dataset[Row],每一行的类型是 Row，不解析，每一行究竟有哪些字段，各个字段又是什么类型都无从得知，只能用上面提到的 getAS 方法或者共性中的第七条提到的模式匹配拿出特定字段。而 Dataset 中，每一行是什么类型是不一定的，在自定义了 case class 之后可以很自由的获得每一行的信息

2.4.3 三者之间的互相转换

image-20230820001407718

2.5 开发SparkSQL

2.5.1 添加maven依赖

主要是spark-sql_2.12，其他依赖可自行选择，生产环境最好还是使用新版本。

<dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-yarn_2.12</artifactId>
            <version>3.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.0.0</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.27</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.12</artifactId>
            <version>3.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hive</groupId>
            <artifactId>hive-exec</artifactId>
            <version>1.2.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.12</artifactId>
            <version>3.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
            <version>3.0.0</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-core</artifactId>
            <version>2.10.1</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/com.alibaba/druid -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid</artifactId>
            <version>1.1.10</version>
        </dependency>
    </dependencies>

2.5.2 简单实现（学习）

package com.fyj.bigdata.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object Spark01_SparkSQL_Basic {

  def main(args: Array[String]): Unit = {
    // TODO 创建SparkSQL的运行环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark01_SparkSQL_Basic")
    val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    // 隐式转换，在使用DataFrame时，如果涉及到转换操作，需要引入转换规则
    import sparkSession.implicits._
    // TODO 执行逻辑操作
    val dataFrame: DataFrame = sparkSession.read.json("datas/user.json")
    // 1 可直接查看dataFrame的内容
    //    dataFrame.show()
    // 2 创建视图
    dataFrame.createOrReplaceTempView("user")
    // 2.1 DataFrame => SQL
        sparkSession.sql("select * from user").show()
    //    sparkSession.sql("select age,username from user").show()
    //    sparkSession.sql("select avg(age) from user").show()
    // 2.2 DataFrame => DSL
    // 在使用DataFrame时，如果涉及到转换操作，需要引入转换规则
    //    dataFrame.select("age","username").show()
    //    dataFrame.select($"age"+1).show()
    //    dataFrame.select('age+1).show() // ' 作用等同于 $""
    // TODO DataSet(DataFrame其实是特定泛型的DataSet)
    //    val seq = Seq(1,2,3,4)
    //    val ds: Dataset[Int] = seq.toDS()
    //    ds.show()
    // RDD <==转换==> DataFrame
//    val rdd = sparkSession.sparkContext.makeRDD(List((1, "zhangsan", 30), (2, "lisi", 18)))
//    val dF: DataFrame = rdd.toDF("id", "name", "age")
//    val rowRDD: RDD[Row] = dF.rdd
//
//    // DataFrame <==转换==> DataSet
//    val ds: Dataset[User] = dF.as[User]
//    val dataFrame1 = ds.toDF()
//
//    // RDD <==转换==> DataSet
//    val ds1: Dataset[User] = rdd.map {
//      case (id, name, age) => {
//        User(id, name, age)
//      }
//    }.toDS()
//    val rdd1: RDD[User] = ds1.rdd
   // TODO 关闭环境
    sparkSession.close()
  }

  // 创建类
  case class User(id: Int, name: String, age: Int)

}

2.6 自定义函数

用户可以通过 spark.udf 功能添加自定义函数，实现自定义功能。

2.6.1 UDF函数

目标：给username加一个自定义前缀。

package com.fyj.bigdata.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

object Spark02_SparkSQL_UDF {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark02_SparkSQL_UDF")
    val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    import sparkSession.implicits._
    val dataFrame = sparkSession.read.json("datas/user.json")
    dataFrame.createOrReplaceTempView("user")
    //sparkSession.sql("select age from user where username='lisi'").show()
    // 注册UDF函数
    sparkSession.udf.register("perfixname", (name: String) => {
      "Name:" + name
    })
    // 在SparkSQL中引用UDF函数
    sparkSession.sql("select age,perfixname(username) from user").show()
    sparkSession.close()
  }

}

2.6.2 UDAF

​ SparkSQL中强类型的DataSet和弱类型的DataFrame都提供了相关的聚合函数，如count()，countDistinct()，avg()，max()，min()。

​ 用户也可以自定义聚合函数UDAF：老版本的方式：通过继承 UserDefinedAggregateFunction 来实现用户自定义弱类型聚合函数；从 Spark3.0 版本后的方式：可以统一采用强类型聚合函数 Aggregator。

目标：计算出年龄的平均值

（1）实现方式-弱类型-UDAF

package com.fyj.bigdata.spark.sql

import io.netty.buffer.Unpooled.buffer
import org.apache.calcite.adapter.enumerable.RexImpTable.UserDefinedAggReflectiveImplementor
import org.apache.spark.SparkConf
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, LongType, StructField, StructType}

object Spark03_SparkSQL_UDAF {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark03_SparkSQL_UDAF")
    val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    val dataFrame = sparkSession.read.json("datas/user.json")
    dataFrame.createOrReplaceTempView("user")
    //sparkSession.sql("select age from user where username='lisi'").show()
    //sparkSession.sql("select * from user").show()
    // 注册UDF函数-弱类型
    sparkSession.udf.register("avgAge", new MyAvgUDAF())
    // 在SparkSQL中引用UDF函数
    sparkSession.sql("select avgAge(age) from user").show()
    sparkSession.close()
  }

  /**
   * 自定义聚合函数类：计算年龄的平均值
   * 1.继承UserDefinedAggregateFunction
   * 2.重写方法
   */
  class MyAvgUDAF extends UserDefinedAggregateFunction {
    // 1.输入数据的结构 intput
    override def inputSchema: StructType = {
      StructType(
        Array(StructField("age", LongType))
      )
    }

    // 2.缓冲区数据的结构 buffer
    override def bufferSchema: StructType = {
      StructType(
        Array(
          StructField("total", LongType),
          StructField("count", LongType),
        )
      )
    }

    // 3.输出数据的结构 output
    override def dataType: DataType = LongType

    // 4.函数的稳定性
    override def deterministic: Boolean = true

    // 5.缓冲区的初始化
    override def initialize(buffer: MutableAggregationBuffer): Unit = {
      // 0和1表示位置
      buffer.update(0, 0L)
      buffer.update(1, 0L)
    }

    // 6.根据输入的值更新缓冲区的数据 按照业务逻辑处理数据
    override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
      buffer.update(0, buffer.getLong(0) + input.getLong(0))
      buffer.update(1, buffer.getLong(1) + 1)
    }

    // 7.缓冲区的数据合并（分布式运算后需要合并）
    override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
      buffer1.update(0, buffer1.getLong(0) + buffer2.getLong(0))
      buffer1.update(1, buffer1.getLong(1) + buffer2.getLong(1))
    }

    // 8.计算平均值
    override def evaluate(buffer: Row): Any = {
      buffer.getLong(0) / buffer.getLong(1)
    }
  }
}

（2）实现方式-强类型-UDAF-老版本

package com.fyj.bigdata.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.{Dataset, Encoder, Encoders, SparkSession, TypedColumn, functions}

object Spark05_SparkSQL_UDAF {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark05_SparkSQL_UDAF")
    val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    import sparkSession.implicits._
    val dataFrame = sparkSession.read.json("datas/user.json")

    val ds: Dataset[User] = dataFrame.as[User]

    // 注册UDF函数-早期强类型方式，将UDAF函数转换为查询的列对象
    val column: TypedColumn[User, Long] = new MyAvgUDAF().toColumn
    // 在SparkSQL中引用UDF函数
    ds.select(column).show()
    sparkSession.close()
  }

  /**
   * 自定义聚合函数类：计算年龄的平均值
   * 1.继承org.apache.spark.sql.expressions.Aggregator 定义泛型
   * IN:输入的数据类型User
   * BUF:缓冲区的数据类型Buff
   * OUT:输出的数据类型Long
   *
   * 2.重写方法（6个）
   */
  // 定义用于dataset的类
  case class User(var age: Long, var username: String)

  // 定义缓冲区数据类型
  case class Buff(var total: Long, var count: Long)

  class MyAvgUDAF extends Aggregator[User, Buff, Long] {
    // 1. z & zero 通常表示初始化或零值的意思，这里是做缓冲区初始化
    override def zero: Buff = {
      Buff(0L, 0L)
    }

    // 2. 根据输入的数据更新缓冲区的数据
    override def reduce(b: Buff, user: User): Buff = {
      b.total = b.total + user.age
      b.count = b.count + 1
      b
    }

    // 3. 合并缓冲区
    override def merge(b1: Buff, b2: Buff): Buff = {
      b1.total = b1.total + b2.total
      b1.count = b1.count + b2.count
      b1
    }

    // 4. 计算结果
    override def finish(reduction: Buff): Long = {
      reduction.total / reduction.count
    }

    // 5. 缓冲区的编码操作
    override def bufferEncoder: Encoder[Buff] = Encoders.product

    // 6. 输出的编码操作
    override def outputEncoder: Encoder[Long] = Encoders.scalaLong
  }
}

（3）实现方式-强类型-UDAF-新版本

package com.fyj.bigdata.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.sql.expressions.{Aggregator, MutableAggregationBuffer}
import org.apache.spark.sql.types.{DataType, LongType, StructField, StructType}
import org.apache.spark.sql.{Encoder, Encoders, Row, SparkSession, functions}

object Spark04_SparkSQL_UDAF {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark04_SparkSQL_UDAF")
    val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    val dataFrame = sparkSession.read.json("datas/user.json")
    dataFrame.createOrReplaceTempView("user")
    // 注册UDF函数-强类型
    sparkSession.udf.register("avgAge", functions.udaf(new MyAvgUDAF()))
    // 在SparkSQL中引用UDF函数
    sparkSession.sql("select avgAge(age) from user").show()
    sparkSession.close()
  }

  /**
   * 自定义聚合函数类：计算年龄的平均值
   * 1.继承org.apache.spark.sql.expressions.Aggregator 定义泛型
   * IN:输入的数据类型Long
   * BUF:缓冲区的数据类型Buff
   * OUT:输出的数据类型Long
   *
   * 2.重写方法（6个）
   */
  // 定义缓冲区数据类型
  case class Buff(var total: Long, var count: Long)

  class MyAvgUDAF extends Aggregator[Long, Buff, Long] {
    // 1. z & zero 通常表示初始化或零值的意思，这里是做缓冲区初始化
    override def zero: Buff = {
      Buff(0L, 0L)
    }

    // 2. 根据输入的数据更新缓冲区的数据
    override def reduce(b: Buff, a: Long): Buff = {
      b.total = b.total + a
      b.count = b.count + 1
      b
    }

    // 3. 合并缓冲区
    override def merge(b1: Buff, b2: Buff): Buff = {
      b1.total = b1.total + b2.total
      b1.count = b1.count + b2.count
      b1
    }

    // 4. 计算结果
    override def finish(reduction: Buff): Long = {
      reduction.total / reduction.count
    }

    // 5. 缓冲区的编码操作
    override def bufferEncoder: Encoder[Buff] = Encoders.product

    // 6. 输出的编码操作
    override def outputEncoder: Encoder[Long] = Encoders.scalaLong
  }
}

2.7 数据的加载和保存

2.7.1 通用的加载和保存方式

​ SparkSQL 提供了通用的保存数据和数据加载的方式。这里的通用指的是使用相同的API，根据不同的参数读取和保存不同格式的数据，SparkSQL 默认读取和保存的文件格式为 parquet。

2.7.1.1 加载数据

spark.read.load 是加载数据的通用方法，如果读取不同格式的数据，可以对不同的数据格式进行设定。

scala> spark.read.
csv format jdbc json load option options orc parquet schema table text textFile

格式如下：
scala> spark.read.format("…")[.option("…")].load("…")
  • format(“…”)：指定加载的数据类型，包括"csv"、“jdbc”、“json”、“orc”、“parquet"和"textFile”。
  • load(“…”)：在"csv"、“jdbc”、“json”、“orc”、"parquet"和"textFile"格式下需要传入加载数据的路径。
  • option(“…”)：在"jdbc"格式下需要传入 JDBC 相应参数，url、user、password 和 dbtable。

可支持直接文件上操作：

scala>spark.sql("select * from json.`/opt/module/data/user.json`").show

2.7.1.2 保存数据

df.write.save 是保存数据的通用方法，如果保存不同格式的数据，可以对不同的数据格式进行设定.

scala> val df = spark.read.json("../data/user.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, username: string]

scala> df.write.
bucketBy   format       jdbc   mode     options   parquet       save          sortBy
csv        insertInto   json   option   orc       partitionBy   saveAsTable   text

格式如下：
scala>df.write.format("…")[.option("…")].save("…")
  • format(“…”)：指定保存的数据类型，包括"csv"、“jdbc”、“json”、“orc”、“parquet"和"textFile”。
  • save (“…”)：在"csv"、“orc”、"parquet"和"textFile"格式下需要传入保存数据的路径。
  • option(“…”)：在"jdbc"格式下需要传入 JDBC 相应参数，url、user、password 和 dbtable.

保存操作可以使用 SaveMode, 用来指明如何处理数据，使用 mode()方法来设置。有一点很重要: 这些 SaveMode 都是没有加锁的, 也不是原子操作。

SaveMode 是一个枚举类，其中的常量包括：

Scala/Java Any Language Meaning
SaveMode.ErrorIfExists(default) “error”(default) 如果文件已经存在则抛出异常
SaveMode.Append “append” 如果文件已经存在则追加
SaveMode.Overwrite “overwrite” 如果文件已经存在则覆盖
SaveMode.Ignore “ignore” 如果文件已经存在则忽略 
df.write.mode("append").json("/opt/module/data/output")

2.7.2 Parquet

​ Spark SQL 的默认数据源为 Parquet 格式。Parquet 是一种能够有效存储嵌套数据的列式存储格式。数据源为 Parquet 文件时，Spark SQL 可以方便的执行所有的操作，不需要使用 format。修改配置项 spark.sql.sources.default，可修改默认数据源格式。

（1）加载数据

scala> val df = spark.read.load("../examples/src/main/resources/users.parquet")
df: org.apache.spark.sql.DataFrame = [name: string, favorite_color: string ... 1 more field]

scala> df.show
+------+--------------+----------------+
|  name|favorite_color|favorite_numbers|
+------+--------------+----------------+
|Alyssa|          null|  [3, 9, 15, 20]|
|   Ben|           red|              []|
+------+--------------+----------------+

（2）保存数据

dataframe保存数据：

scala> val df1 = spark.read.json("../data/user.json")
scala> df1.write.mode("append").save("../data/user")

dataSet保存数据：

scala> case class User(username:String,age:Long)
scala> df1.write.mode("append").save("../data/user")
scala> val ds1 = df1.as[User]
scala> ds1.write.mode("append").format("json").save("../data/dsUser")

2.7.3 JSON

​ Spark SQL 能够自动推测 JSON 数据集的结构，并将它加载为一个 Dataset[Row]. 可以通过 SparkSession.read.json()去加载 JSON 文件。

注意：Spark 读取的 JSON 文件不是传统的 JSON 文件，每一行都应该是一个 JSON 串。

读取JSON文件的步骤：

（1）导入隐式转换

scala> import spark.implicits._

（2）加载JSON文件

scala> val path = "/opt/module/spark-local/people.json"
scala> val peopleDF = spark.read.json(path)

（3）创建临时表

scala> peopleDF.createOrReplaceTempView("people")

（4）数据查询

scala> val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19")
scala> teenagerNamesDF.show()
+------+
| name|
+------+
|Justin|
+------+

2.7.4 CSV

​ Spark SQL 可以配置 CSV 文件的列表信息，读取 CSV 文件,CSV 文件的第一行设置为数据列

scala> val df = spark.read.format("csv").option("sep",";").option("inferSchema","true").option("header",true).load("..\\
examples\\src\\main\\resources\\people.csv")
df: org.apache.spark.sql.DataFrame = [name: string, age: int ... 1 more field]

scala> df.show
+-----+---+---------+
| name|age|      job|
+-----+---+---------+
|Jorge| 30|Developer|
|  Bob| 32|Developer|
+-----+---+---------+

2.7.5 MySQL的加载和保存

​ Spark SQL 可以通过 JDBC 从关系型数据库中读取数据的方式创建 DataFrame，通过对DataFrame 一系列的计算后，还可以将数据再写回关系型数据库中。如果使用 spark-shell 操作，可在启动 shell 时指定相关的数据库驱动路径或者将相关的数据库驱动放到 spark 的类路径下。

（1）命令行方式

bin/spark-shell --jars mysql-connector-java-5.1.27-bin.jar

（2）代码实现

导入mysql依赖到pom.xml

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.27</version>
        </dependency>

代码：

package com.fyj.bigdata.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql._
import org.datanucleus.store.types.wrappers.backed.Properties

import scala.collection.JavaConverters._

object Spark06_SparkSQL_Mysql {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark06_SparkSQL_Mysql")
    val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    import sparkSession.implicits._
    /**
     * 读取并加载mysql数据库中的数据的方式一共有三种：
     *
     */
    //第一种，读取数据
    val dataFrame = sparkSession.read.format("jdbc").
      option("url", "jdbc:mysql://linux1:3306/spark-sql").
      option("driver", "com.mysql.jdbc.Driver").
      option("user", "root").
      option("password", 1234).
      option("dbtable", "user")
      .load()
    dataFrame.show()
    //第二种，读取数据
    val dataFrame1 = sparkSession.read.format("jdbc").options(Map(
      "url" -> "jdbc:mysql://linux1:3306/spark-sql?user=root&password=123123",
      "driver" -> "com.mysql.jdbc.Driver",
      "dbtable" -> "user"
    )).load()
    dataFrame1.show()
    //第三种，读取数据
//    val properties: Properties = new Properties()
//    properties.setProperty("user","root")
//    properties.setProperty("password","123456")
//    val dataFrame2 = sparkSession.read.jdbc("jdbc:mysql://linux1:3306/spark-sql", "user", properties)
//    dataFrame2.show()

    // 将数据保存至mysql数据库中
    dataFrame.write.format("jdbc")
      .option("url", "jdbc:mysql://linux1:3306/spark-sql")
      .option("driver", "com.mysql.jdbc.Driver")
      .option("user", "root")
      .option("password", 1234)
      .option("dbtable", "user1")
      .mode(SaveMode.Append)
      .save()

	
    sparkSession.close()
  }
}

2.7.6 内部Hive（不常用）

​ Apache Hive 是 Hadoop 上的 SQL 引擎，Spark SQL 编译时可以包含 Hive 支持，也可以不包含。包含 Hive 支持的 Spark SQL 可以支持 Hive 表访问、UDF (用户自定义函数)以及 Hive 查询语言(HiveQL/HQL)等。需要强调的一点是，如果要在 Spark SQL 中包含Hive 的库，并不需要事先安装 Hive。一般来说，最好还是在编译 Spark SQL 时引入 Hive支持，这样就可以使用这些特性了。如果你下载的是二进制版本的 Spark，它应该已经在编译时添加了 Hive 支持。

​ 若要把 Spark SQL 连接到一个部署好的 Hive 上，你必须把 hive-site.xml 复制到Spark 的配置文件目录中($SPARK_HOME/conf)。即使没有部署好 Hive，Spark SQL 也可以运行。 需要注意的是，如果你没有部署好 Hive，Spark SQL 会在当前的工作目录中创建出自己的 Hive 元数据仓库，叫作 metastore_db。此外，如果你尝试使用 HiveQL 中的CREATE TABLE (并非 CREATE EXTERNAL TABLE)语句来创建表，这些表会被放在你默认的文件系统中的 /user/hive/warehouse 目录中(如果你的 classpath 中有配好的hdfs-site.xml，默认的文件系统就是 HDFS，否则就是本地文件系统)。

​ spark-shell 默认是 Hive 支持的；代码中是默认不支持的，需要手动指定（加一个参数即可）。

​ 如果使用 Spark 内嵌的 Hive, 则什么都不用做, 直接使用即可.Hive 的元数据存储在 derby 中, 默认仓库地址:$SPARK_HOME/spark-warehouse

scala> spark.sql("show databases;").show
23/08/25 00:05:55 WARN HiveConf: HiveConf of name hive.stats.jdbc.timeout does not exist
23/08/25 00:05:55 WARN HiveConf: HiveConf of name hive.stats.retries.wait does not exist
23/08/25 00:05:57 WARN ObjectStore: Version information not found in metastore. hive.metastore.schema.verification is not enabled so recording the schema version 2.3.0
23/08/25 00:05:57 WARN ObjectStore: setMetaStoreSchemaVersion called but recording version is disabled: version = 2.3.0, comment = Set by MetaStore fyj@192.168.1.102
+---------+
|namespace|
+---------+
|  default|
+---------+


scala> spark.sql("show tables;").show
23/08/25 00:06:12 WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException
+--------+---------+-----------+
|database|tableName|isTemporary|
+--------+---------+-----------+
+--------+---------+-----------+

向表加载本地数据

数据样例：

[fyj@hadoop102 graphx]$ cat followers.txt 
2 1
4 1
1 2
6 3
7 3
7 6
6 7
3 7

用内置hive建表并加载数据

scala> spark.sql("create table followers(aaa String,bbb String) row format delimited fields terminated by ' ';")
23/08/25 00:16:41 WARN HiveMetaStore: Location: file:/opt/module/spark/bin/spark-warehouse/followers specified for non-external table:followers
res19: org.apache.spark.sql.DataFrame = []

scala> spark.sql("load data local inpath '/opt/module/spark/data/graphx/followers.txt' into table followers")
res20: org.apache.spark.sql.DataFrame = []

scala> spark.sql("select * from followers").show
+---+---+
|aaa|bbb|
+---+---+
|  2|  1|
|  4|  1|
|  1|  2|
|  6|  3|
|  7|  3|
|  7|  6|
|  6|  7|
|  3|  7|
+---+---+

2.7.7 外部HIVE（重点）

如果想连接外部已经部署好的 Hive，需要通过以下几个步骤：

  • Spark 要接管 Hive 需要把 hive-site.xml 拷贝到 conf/目录下
  • 把 Mysql 的驱动 copy 到 jars/目录下
  • 如果访问不到 hdfs，则需要把 core-site.xml 和 hdfs-site.xml 拷贝到 conf/目录下
  • 重启 spark-shell

（1）用 spark-shell 运行连接外部 hive

scala> spark.sql("show databases;").show
23/08/25 00:25:06 WARN HiveConf: HiveConf of name hive.metastore.event.db.notification.api.auth does not exist
+---------+
|namespace|
+---------+
|  default|
|    gmall|
+---------+

（2）运行 SparkSQL CLI

Spark SQL CLI 可以很方便的在本地运行 Hive 元数据服务以及从命令行执行查询任务。在Spark 目录下执行如下命令启动 Spark SQL CLI，直接执行 SQL 语句，类似一 Hive 窗口

运行脚本：./spark-sql 会出来一个sql窗口，在窗口内输入sql即可

spark-sql (default)> show databases;
23/08/25 00:28:14 INFO CodeGenerator: Code generated in 163.042892 ms
23/08/25 00:28:14 INFO CodeGenerator: Code generated in 4.845857 ms
namespace
default
gmall
Time taken: 1.238 seconds, Fetched 2 row(s)
23/08/25 00:28:14 INFO SparkSQLCLIDriver: Time taken: 1.238 seconds, Fetched 2 row(s)

（3）运行 Spark beeline

​ Spark Thrift Server 是 Spark 社区基于 HiveServer2 实现的一个 Thrift 服务。旨在无缝兼容HiveServer2。因为 Spark Thrift Server 的接口和协议都和 HiveServer2 完全一致，因此我们部署好 Spark Thrift Server 后，可以直接使用 hive 的 beeline 访问 Spark Thrift Server 执行相关语句。Spark Thrift Server 的目的也只是取代 HiveServer2，因此它依旧可以和 Hive Metastore进行交互，获取到 hive 的元数据。

​ 如果想连接 Thrift Server，需要通过以下几个步骤：

  • Spark 要接管 Hive 需要把 hive-site.xml 拷贝到 conf/目录下；
  • 把 Mysql 的驱动 copy 到 jars/目录下；
  • 如果访问不到 hdfs，则需要把 core-site.xml 和 hdfs-site.xml 拷贝到 conf/目录下；
  • 启动 Thrift Server。

启动：

sbin/start-thriftserver.sh

连接：

bin/beeline -u jdbc:hive2://linux1:10000 -n root -p 123456

报错信息：

Error: Could not open client transport with JDBC Uri: jdbc:hive2://hadoop102:10000: Failed to open new session: java.lang.RuntimeException: org.apache.hadoop.security.AccessControlException: Permission denied: user=root, access=EXECUTE, inode="/tmp":fyj:supergroup:drwxrwx---xxxxxxxxxx 

解决办法，将hdfs的/tmp目录设置为777，重连即可

hdfs dfs -chmod 777 /tmp

[fyj@hadoop102 bin]$ ./beeline -u jdbc:hive2://hadoop102:10000 -n root -p 123456
log4j:WARN No appenders could be found for logger (org.apache.hadoop.util.Shell).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Connecting to jdbc:hive2://hadoop102:10000
Connected to: Apache Hive (version 3.1.2)
Driver: Hive JDBC (version 2.3.7)
Transaction isolation: TRANSACTION_REPEATABLE_READ
Beeline version 2.3.7 by Apache Hive
0: jdbc:hive2://hadoop102:10000> show databases;
INFO  : Compiling command(queryId=fyj_20230825005402_9a3a7ea4-feeb-4e50-b389-b82f065b1428): show databases
INFO  : Concurrency mode is disabled, not creating a lock manager
INFO  : Semantic Analysis Completed (retrial = false)
INFO  : Returning Hive schema: Schema(fieldSchemas:[FieldSchema(name:database_name, type:string, comment:from deserializer)], properties:null)
INFO  : Completed compiling command(queryId=fyj_20230825005402_9a3a7ea4-feeb-4e50-b389-b82f065b1428); Time taken: 1.082 seconds
INFO  : Concurrency mode is disabled, not creating a lock manager
INFO  : Executing command(queryId=fyj_20230825005402_9a3a7ea4-feeb-4e50-b389-b82f065b1428): show databases
INFO  : Starting task [Stage-0:DDL] in serial mode
INFO  : Completed executing command(queryId=fyj_20230825005402_9a3a7ea4-feeb-4e50-b389-b82f065b1428); Time taken: 0.016 seconds
INFO  : OK
INFO  : Concurrency mode is disabled, not creating a lock manager
+----------------+
| database_name  |
+----------------+
| default        |
| gmall          |
+----------------+
2 rows selected (1.345 seconds)
0: jdbc:hive2://hadoop102:10000>

（4）代码操作Hive

导入依赖maven pom.xml：

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>3.0.0</version>
        </dependency>        
		<dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.12</artifactId>
            <version>3.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hive</groupId>
            <artifactId>hive-exec</artifactId>
            <version>1.2.1</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.33</version>
        </dependency>

将 hive-site.xml 文件拷贝到项目的 src/main/resources 目录下和 target/classes 目录下，hdfs的配置文件也可以导入，如果配置文件中没写IP，只写了hostname需要改成IP（如果本地hosts文件有hostname解析就不用改）

代码实现：

package com.fyj.bigdata.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.sql._

object Spark07_SparkSQL_Hive {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark07_SparkSQL_Hive")
    /**
     * 连接外置Hive
     * （1）创建可支持Hive的sparkSession，enableHiveSupport()
     * （2）拷贝Hive-size.xml文件classpath下
     * （3）启用Hive的支持
     * （4）增加对应的依赖关系（包含MySQL驱动）
     */
    val sparkSession = SparkSession.builder().enableHiveSupport().config(sparkConf).getOrCreate()

    sparkSession.sql("show databases").show()
    sparkSession.sql("use gmall;").show()
    sparkSession.sql("show tables;").show()

    sparkSession.close()
  }
}

注意：在IDEA开发工具中创建数据库默认是在本地仓库，通过参数修改数据库仓库的地址，修改为自己的 hadoop 用户名称

config("spark.sql.warehouse.dir", "hdfs://IP:8020/user/hive/warehouse")
System.setProperty("HADOOP_USER_NAME", "root")

第三章 实战演练

3.1 数据样例：

user_visit_action：

2019-07-17  95 26070e87-1ad7-49a3-8fb3-cc741facaddf   37 2019-07-17 00:00:02    手机 -1 -1 \N \N \N \N 3
2019-07-17 95 26070e87-1ad7-49a3-8fb3-cc741facaddf   48 2019-07-17 00:00:10    \N 16 98 \N \N \N \N 19

product_info：

1  商品_1   自营
2  商品_2   自营
3  商品_3   自营

city_info：

1  北京 华北
2  上海 华东
3  深圳 华南

3.2 需求：

  • 查询出来所有的点击记录，并与 city_info 表连接，得到每个城市所在的地区，与Product_info 表连接得到产品名称

  • 按照地区和商品 id 分组，统计出每个商品在每个地区的总点击次数

  • 每个地区内按照点击次数降序排列

  • 只取前三名

  • 城市备注需要自定义 UDAF 函数

image-20230827002641183

3.3 创建表结构和导入数据

package com.fyj.bigdata.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.sql._

object Spark08_SparkSQL_SalesProject {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "fyj")
    /**
     * 销售项目实战演练
     * （1）提前创建数据库sales
     * （2）建表
     * （3）导入数据
     * （4）检查数据
     */
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark08_SparkSQL_SalesProject")
    val sparkSession = SparkSession.builder().enableHiveSupport().config(sparkConf).getOrCreate()
    // 1. 选择数据库
    sparkSession.sql("use sales")
    // 2. 建表user_visit_action并从本地导入数据
    sparkSession.sql(
      """
        |CREATE TABLE `user_visit_action`(
        | `date` string,
        | `user_id` bigint,
        | `session_id` string,
        | `page_id` bigint,
        | `action_time` string,
        | `search_keyword` string,
        | `click_category_id` bigint,
        | `click_product_id` bigint,
        | `order_category_ids` string,
        | `order_product_ids` string,
        | `pay_category_ids` string,
        | `pay_product_ids` string,
        | `city_id` bigint)
        |row format delimited fields terminated by '\t'
        |""".stripMargin)
    // 注意导入的文件路径
    sparkSession.sql("load data local inpath 'datas/user_visit_action.txt' into table sales.user_visit_action")


    // 3. 建表product_info并从本地导入数据
    sparkSession.sql(
      """
        |CREATE TABLE `product_info`(
        | `product_id` bigint,
        | `product_name` string,
        | `extend_info` string)
        |row format delimited fields terminated by '\t'
        |""".stripMargin)
    sparkSession.sql("load data local inpath 'datas/product_info.txt' into table sales.product_info")

    // 4. 建表city_info并从本地导入数据
    sparkSession.sql(
      """
        |CREATE TABLE `city_info`(
        | `city_id` bigint,
        | `city_name` string,
        | `area` string)
        |row format delimited fields terminated by '\t'
        |""".stripMargin)
    sparkSession.sql("load data local inpath 'datas/city_info.txt' into table sales.city_info")

    // 5. 抽样检查数据
    sparkSession.sql("select * from sales.user_visit_action limit 10").show()
    sparkSession.sql("select * from sales.product_info limit 10").show()
    sparkSession.sql("select * from sales.city_info limit 10").show()

    sparkSession.close()
  }
}

3.4 实现需求

package com.fyj.bigdata.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.sql._
import org.apache.spark.sql.expressions.Aggregator

import scala.collection.mutable
import scala.collection.mutable.ListBuffer

object Spark10_SparkSQL_SalesProject {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "fyj")
    /**
     * 销售项目实战演练
     * （1）提前创建数据库sales
     * （2）建表
     * （3）导入数据
     * （4）检查数据
     */
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark09_SparkSQL_SalesProject")
    val sparkSession = SparkSession.builder().enableHiveSupport().config(sparkConf).getOrCreate()
    // 1. 选择数据库
    sparkSession.sql("use sales")

    sparkSession.sql(
      """
        |			select
        |			u.*
        |			,p.product_name
        |			,c.area
        |			,c.city_name
        |			from user_visit_action u
        |			join product_info p on u.click_product_id = p.product_id
        |			join city_info c on u.city_id = c.city_id
        |			where u.click_product_id > -1
        |""".stripMargin).createOrReplaceTempView("t1")

    // 注册自定义UDAF函数
    sparkSession.udf.register("cityRemark",functions.udaf(new CityRemarkUDAF()))
    sparkSession.sql(
      """
        |		select
        |		area
        |		,product_name
        |		,count(*) as clickcnt
        |  ,cityRemark(city_name) as city_remark
        |   from t1 group by t1.area,t1.product_name
        |""".stripMargin).createOrReplaceTempView("t2")

    sparkSession.sql(
      """
        |	  select
        |	  *
        |	  ,rank() over (partition by area order by clickcnt desc ) as rank
        |	  from t2
        |""".stripMargin).createOrReplaceTempView("t3")

    sparkSession.sql(
      """
        |   select
        |   *
        |   from t3  where rank <= 3
        |""".stripMargin).show(false)

    sparkSession.close()
  }

  case class Buffer(var total: Long, var cityMap: mutable.Map[String, Long])

  /**
   * 自定义聚合函数：实现城市备注功能
   * 1.继承Aggregator
   * IN ： 城市名称 【输入】
   * BUF：Buffer -> 总点击量，map[(city,cnt),(city,cnt)]
   * OUT：所需要的备注信息【输出】
   * 2.重写方法6个
   */
  class CityRemarkUDAF extends Aggregator[String, Buffer, String] {
    override def zero: Buffer = {
      Buffer(0, mutable.Map[String, Long]())
    }

    override def reduce(buff: Buffer, city: String): Buffer = {
      buff.total += 1
      val newCount = buff.cityMap.getOrElse(city, 0L) + 1
      buff.cityMap.update(city, newCount)
      buff
    }

    override def merge(buff1: Buffer, buff2: Buffer): Buffer = {
      buff1.total += buff2.total
      val map1 = buff1.cityMap
      val map2 = buff2.cityMap
      buff1.cityMap = map1.foldLeft(map2) {
        case (map, (city, cnt)) => {
          val newCount = map.getOrElse(city, 0L) + cnt
          map.update(city, newCount)
          map
        }
      }
      buff1
    }

    override def finish(buff: Buffer): String = {
      val remarkList = ListBuffer[String]()
      val cityMap = buff.cityMap
      val totalcnt = buff.total
      // 降序排列
      val cityCntList = cityMap.toList.sortWith(
        (left,right) => {
          left._2 > right._2
        }
      ).take(2)
      val hasMore = cityMap.size >2
      var rsum = 0L
      cityCntList.foreach{
        case (city,cnt) => {
          val r = cnt * 100 / totalcnt
          remarkList.append(s"${city} ${r}%")
          rsum += r
        }
      }
      if (hasMore){
        remarkList.append(s"其他 ${100 - rsum}%")
      }
      remarkList.mkString(", ")
    }

    override def bufferEncoder: Encoder[Buffer] = Encoders.product

    override def outputEncoder: Encoder[String] = Encoders.STRING
  }
}

结果：

+----+------------+--------+------------------------------+----+
|area|product_name|clickcnt|city_remark                   |rank|
+----+------------+--------+------------------------------+----+
|华东|商品_86     |371     |上海 16%, 杭州 15%, 其他 69%  |1   |
|华东|商品_47     |366     |杭州 15%, 青岛 15%, 其他 70%  |2   |
|华东|商品_75     |366     |上海 17%, 无锡 15%, 其他 68%  |2   |
|西北|商品_15     |116     |西安 54%, 银川 45%            |1   |
|西北|商品_2      |114     |银川 53%, 西安 46%            |2   |
|西北|商品_22     |113     |西安 54%, 银川 45%            |3   |
|华南|商品_23     |224     |厦门 29%, 福州 24%, 其他 47%  |1   |
|华南|商品_65     |222     |深圳 27%, 厦门 26%, 其他 47%  |2   |
|华南|商品_50     |212     |福州 27%, 深圳 25%, 其他 48%  |3   |
|华北|商品_42     |264     |郑州 25%, 保定 25%, 其他 50%  |1   |
|华北|商品_99     |264     |北京 24%, 郑州 23%, 其他 53%  |1   |
|华北|商品_19     |260     |郑州 23%, 保定 20%, 其他 57%  |3   |
|东北|商品_41     |169     |哈尔滨 35%, 大连 34%, 其他 31%|1   |
|东北|商品_91     |165     |哈尔滨 35%, 大连 32%, 其他 33%|2   |
|东北|商品_58     |159     |沈阳 37%, 大连 32%, 其他 31%  |3   |
|东北|商品_93     |159     |哈尔滨 38%, 大连 37%, 其他 25%|3   |
|华中|商品_62     |117     |武汉 51%, 长沙 48%            |1   |
|华中|商品_4      |113     |长沙 53%, 武汉 46%            |2   |
|华中|商品_57     |111     |武汉 54%, 长沙 45%            |3   |
|华中|商品_29     |111     |武汉 50%, 长沙 49%            |3   |
+----+------------+--------+------------------------------+----+
only showing top 20 rows
SparkSQL