一、概述
1、介绍
将Spark SQL转换成RDD，然后提交到集群执行【对比hive】
提供2个编程抽象：DataFrame&DataSet
可以使用SQL和DatasetAPI与Spark SQL交互
2、特点
易整合SQL和spark
统一的数据访问方式
兼容hive，可以直接运行SQL或hql
标准数据连接JDBC/ODBC
3、DataFrame
以RDD为基础，每一列都带有名称和类型
支持嵌套数据类型（struct、array和map）
懒执行，但通过查询计划优化（先过滤后join）【基于关系代数的等价变换】，能获得高性能
4、DataSet
分布式数据库集合
DataFrame=DataSet[Row]
二、Spark SQL编程
1、SparkSession查询起始点
老版本有两种：SQLContext 和 HiveContext
SparkSession是Spark最新的SQL查询起始点
SparkSession内部封装了sparkContext，实际上是由sparkContext完成计算
2、DataFrame
【转换重要，DataSet的API更具有函数式】
（1）创建DataFrame
数据源、json、其他RDD转换【spark.read.xxx】
（2）SQL风格语法
创建临时表df.createOrReplaceTempView("people") --session范围内有效
整体范围内有效需要创建全局表：df.createGlobalTempView("people")
SQL查询并显示：
val sqlDF = spark.sql("SELECT * FROM people")
sqlDF.show
（3）特定领域语言(domain-specific language, DSL)DSL风格语法
查看DataFrame的Schema信息：df.printSchema
查看某一列或所有列：
df.select("name").show()
df.select("*").show
df.select($"name",$"age" + 1).show --涉及运算，列名用$
df.filter($"age">19).show
df.groupBy("age").count.show
（4）RDD转换为DataFrame
需要引入 import spark（sparkSession对象）.implicits._
手动转换
peopleRDD.map{x=> val fields=x.split(",");(fields(0),fields(1).trim.toInt)}.toDF("name","age").show
样例类反射转换
//创建样例类
case class People(name:String,age:Int)
//根据样例类转换
peopleRDD.map{x=> var fields=x.split(",");People(fields(0),fields(1).toInt)}.toDF.show
编程方式转换
package day05 import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession} object DataFrameDemo2 {
def main(args: Array[String]): Unit = {
val spark: SparkSession = SparkSession.builder()
.master("local[*]")
.appName("Word Count")
.getOrCreate()
val sc: SparkContext = spark.sparkContext
val rdd: RDD[(String, Int)] = sc.parallelize(Array(("lisi", 10), ("zs", 20), ("zhiling", 40)))
// 映射出来一个 RDD[Row], 因为 DataFrame其实就是 DataSet[Row]
val rowRdd: RDD[Row] = rdd.map(x => Row(x._1, x._2))
// 创建 StructType 类型
val types = StructType(Array(StructField("name", StringType), StructField("age", IntegerType)))
val df: DataFrame = spark.createDataFrame(rowRdd, types)
df.show
}
}
（5）DataFrame转换为RDD
val df = spark.read.json("/opt/module/spark-local/people.json")
直接调用rdd：val dfToRDD = df.rdd
打印RDD：dfToRDD.collect
3、DataSet
（1）创建DataSet
使用样例类序列创建DataSet
//创建样例类
case class Person(name: String, age: Long)
//构造样例类序列
val caseClassDS = Seq(Person("wangyuyan",2)).toDS()
//查看详细信息
caseClassDS.show
使用基本类型的序列创建DataSet
val ds = Seq(1,2,3,4,5,6).toDS
注意：不常通过序列，多用RDD转换
（2）RDD转换为DataSet
//创建RDD
val peopleRDD = sc.textFile("/opt/module/spark-local/people.txt")
//创建样例类
case class Person(name:String,age:Int)
//将RDD转化为DataSet
peopleRDD.map(line => {val fields = line.split(",");Person(fields(0),fields(1). toInt)}).toDS
（3）DataSet转换为RDD
//创建DataSet
val DS = Seq(Person("zhangcuishan", 32)).toDS()
//将DataSet转换为RDD
DS.rdd
4、DataFrame与DataSet的互操作
（1）DataFrame转为DataSet
import spark.implicits._
//创建一个DateFrame
val df = spark.read.json("/opt/module/spark-local/people.json")
//创建一个样例类
case class Person(name: String,age: Long)
//将DataFrame转化为DataSet
df.as[Person]
（2）Dataset转为DataFrame
//创建一个样例类
case class Person(name: String,age: Long)
//创建DataSet
val ds = Seq(Person("zhangwuji",32)).toDS()
//将DataSet转化为DataFrame
var df = ds.toDF
//展示
df.show
5、RDD、DataFrame和DataSet之间的关系
（1）共性
都是spark平台下的分布式弹性数据集
都有惰性机制，遇到action算子才执行计算
自动缓存，避免溢出
都有partition的概念
可使用模式匹配获取各个字段的值和类型
（2）区别
行类型不同
应用场景
（3）相互转换
6、IDEA创建SparkSQL程序
object SparkSQL01_Demo {
def main(args: Array[String]): Unit = {
//创建上下文环境配置对象
val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL01_Demo")
//创建SparkSession对象
val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
//RDD=>DataFrame=>DataSet转换需要引入隐式转换规则，否则无法转换
//spark不是包名，是上下文环境对象名
import spark.implicits._
//读取json文件 创建DataFrame {"username": "lisi","age": 18}
val df: DataFrame = spark.read.json("D:\\dev\\workspace\\spark-bak\\spark-bak-00\\input\\test.json")
//df.show()
//SQL风格语法
df.createOrReplaceTempView("user")
//spark.sql("select avg(age) from user").show
//DSL风格语法
//df.select("username","age").show()
//*****RDD=>DataFrame=>DataSet*****
//RDD
val rdd1: RDD[(Int, String, Int)] = spark.sparkContext.makeRDD(List((1,"qiaofeng",30),(2,"xuzhu",28),(3,"duanyu",20)))
//DataFrame
val df1: DataFrame = rdd1.toDF("id","name","age")
//df1.show()
//DateSet
val ds1: Dataset[User] = df1.as[User]
//ds1.show()
//*****DataSet=>DataFrame=>RDD*****
//DataFrame
val df2: DataFrame = ds1.toDF()
//RDD 返回的RDD类型为Row，里面提供的getXXX方法可以获取字段值，类似jdbc处理结果集，但是索引从0开始
val rdd2: RDD[Row] = df2.rdd
//rdd2.foreach(a=>println(a.getString(1)))
//*****RDD=>DataSe*****
rdd1.map{
case (id,name,age)=>User(id,name,age)
}.toDS()
//*****DataSet=>=>RDD*****
ds1.rdd
//释放资源
spark.stop()
}
}
case class User(id:Int,name:String,age:Int)
7、用户自定义函数
（1）UDF：输入一行，返回一个结果
//创建DataFrame
val df = spark.read.json("/opt/module/spark-local/people.json")
//打印数据
df.show
//注册UDF，功能为在数据前添加字符串
spark.udf.register("addName",(x:String)=> "Name:"+x)
//创建临时表
df.createOrReplaceTempView("people")
//应用UDF
spark.sql("Select addName(name),age from people").show()
（2）UDAF（User Defined Aggregate Function）：输入多行，返回一行【用户自定义聚合函数】
需求：实现求平均年龄
RDD算子方式实现
val res: (Int, Int) = sc.makeRDD(List(("zhangsan", 20), ("lisi", 30), ("wangw", 40))).map {
case (name, age) => {
(age, 1)
}
}.reduce {
(t1, t2) => {
(t1._1 + t2._1, t1._2 + t2._2)
}
}
println(res._1/res._2)
自定义累加器方式实现（减少shuffle提高效率）
var sumAc = new MyAC
sc.register(sumAc)
sc.makeRDD(List(("zhangsan",20),("lisi",30),("wangw",40))).foreach{
case (name,age)=>{
sumAc.add(age)
}
}
println(sumAc.value) class MyAC extends AccumulatorV2[Int,Int]{
内部包含copy、reset、add、merge函数
自定义聚合函数实现-弱类型
//在spark中注册聚合函数
spark.udf.register("avgAge",myAverage) //读取数据 {"username": "zhangsan","age": 20}
val df: DataFrame = spark.read.json("D:\\dev\\workspace\\spark-bak\\spark-bak-00\\input\\test.json") //创建临时视图
df.createOrReplaceTempView("user") //使用自定义函数查询
spark.sql("select avgAge(age) from user").show()
自定义聚合函数实现-强类型
（3）UDTF：输入一行，返回多行
spark用flatMap即可实现该功能
三、SparkSQL数据的加载与保存
1、通用的加载和保存方式
（1）加载数据
spark.read.直接加载
format加载指定数据类型：spark.read.format("json").load ("/opt/module/spark-local/people.json").show
文件上直接进行查询（反引号）：spark.sql("select * from json.`/opt/module/spark-local/people.json`").show
（2）保存数据
write直接保存数据
format指定保存数据类型：df.write.mode("append/覆盖/忽略").json("/opt/module/spark-local/output")
（3）默认数据源：Parquet列式存储
配置项spark.sql.sources.default，可修改默认数据源格式
2、JSON文件
spark SQL能自动推测 JSON数据集的结构，并将它加载为一个Dataset[Row].
peopleDF.createOrReplaceTempView("people")
val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19")
teenagerNamesDF.show()
3、MySQL
可以通过JDBC读写数据，使用spark-shell操作需要指定jar包路径
可以使用通用的load方法或者jdbc方法
（1）从JDBC读数据
//方式2:通用的load方法读取 参数另一种形式
spark.read.format("jdbc")
.options(Map("url"->"jdbc:mysql://hadoop202:3306/test?user=root&password=123456",
"dbtable"->"user","driver"->"com.mysql.jdbc.Driver")).load().show //方式3:使用jdbc方法读取
val props: Properties = new Properties()
props.setProperty("user", "root")
props.setProperty("password", "123456")
val df: DataFrame = spark.read.jdbc("jdbc:mysql://hadoop202:3306/test", "user", props)
df.show
（2）向JDBC写数据
val rdd: RDD[User2] = spark.sparkContext.makeRDD(List(User2("lisi", 20), User2("zs", 30)))
val ds: Dataset[User2] = rdd.toDS
//方式1：通用的方式 format指定写出类型
ds.write
.format("jdbc")
.option("url", "jdbc:mysql://hadoop202:3306/test")
.option("user", "root")
.option("password", "123456")
.option("dbtable", "user")
.mode(SaveMode.Append)
.save()
//方式2：通过jdbc方法
val props: Properties = new Properties()
props.setProperty("user", "root")
props.setProperty("password", "123456")
ds.write.mode(SaveMode.Append).jdbc("jdbc:mysql://hadoop202:3306/test", "user", props)
4、Hive
Spark SQL编译时可以包含 Hive 支持，也可以不包含
（1）使用内嵌hive
直接使用即可
spark.sql("show tables").show
spark.sql("create table aa(id int)")
spark.sql("show tables").show
//向表中加载数据
spark.sql("load data local inpath './ids.txt' into table aa")
（2）外部hive应用
步骤：拷贝驱动和配置文件、提前启动hive
能够查到外部hive的大量表
（3）运行Spark SQL CLI
本地运行Hive元数据服务以及从命令行执行查询任务
执行bin/spark-sql
（4）代码中操作Hive
object SparkSQL08_Hive{
def main(args: Array[String]): Unit = {
//创建上下文环境配置对象
val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL01_Demo")
val spark: SparkSession = SparkSession
.builder()
.enableHiveSupport()
.master("local[*]")
.appName("SQLTest")
.getOrCreate()
spark.sql("show tables").show()
//释放资源
spark.stop()
}
}
四、SparkSQL项目实战
1、准备数据
hive中创建3张表： 1张用户行为表，1张城市表，1 张产品表
（1）需求简介
计算各个区域前三大热门商品，并备注上每个商品在主要城市中的分布比例，超过两个城市用其他显示
|
地区 |
商品名称 |
点击次数 |
城市备注 |
|
华北 |
商品A |
100000 |
北京21.2%，天津13.2%，其他65.6% |
|
华北 |
商品P |
80200 |
北京63.0%，太原10%，其他27.0% |
|
华北 |
商品M |
40000 |
北京63.0%，太原10%，其他27.0% |
|
东北 |
商品J |
92000 |
大连28%，辽宁17.0%，其他 55.0% |
（2）思路分析
查询出来所有的点击记录，并与 city_info 表连接，得到每个城市所在的地区，与 Product_info 表连接得到产品名称
按照地区和商品名称分组，统计出每个商品在每个地区的总点击次数
按照点击次数降序排列
城市备注需要自定义 UDAF 函数
（3）代码实现
val spark: SparkSession = SparkSession
.builder()
.master("local[2]")
.appName("AreaClickApp")
.enableHiveSupport()
.getOrCreate()
spark.sql("use sparkpractice")
// 0 注册自定义聚合函数
spark.udf.register("city_remark", new AreaClickUDAF)
// 1. 查询出所有的点击记录,并和城市表产品表做内连接
spark.sql(
"""
|select
| c.*,
| v.click_product_id,
| p.product_name
|from user_visit_action v join city_info c join product_info p on v.city_id=c.city_id and v.click_product_id=p.product_id
|where click_product_id>-1
""".stripMargin).createOrReplaceTempView("t1") // 2. 计算每个区域, 每个产品的点击量
spark.sql(
"""
|select
| t1.area,
| t1.product_name,
| count(*) click_count,
| city_remark(t1.city_name)
|from t1
|group by t1.area, t1.product_name
""".stripMargin).createOrReplaceTempView("t2") // 3. 对每个区域内产品的点击量进行倒序排列
spark.sql(
"""
|select
| *,
| rank() over(partition by t2.area order by t2.click_count desc) rank
|from t2
""".stripMargin).createOrReplaceTempView("t3") // 4. 每个区域取top3 spark.sql(
"""
|select
| *
|from t3
|where rank<=3
""".stripMargin).show //释放资源
spark.stop()
