1、 Spark sql -- 代替hive的（并非完全代替）
(1) Spark sql 和 hive 区别 :
两者都是写sql的，区别是计算引擎不一样
hive -- 计算引擎是MapReduce ，是通过MR做计算的
Spark sql -- 计算引擎是Saprk Core，是通过Spark Core做计算的
Spark sql 功能比 hive 强大 : 并非只能写sql
hive只能在shell行写sql
spark可以在代码中写sql
(2) Spark sql结构 :
1、 Data Source API(读数据) : 可以读取 csv(文本文件)、 json、 jdbc 等各种各样的数据做处理
2、 Data Frame API(提供了两个API):
Dataframe DSL -- 写代码 (DSL : 类SQL语法，与SQL差不多，但它是代码)
Spark SQL and HQL -- 写SQL
(3) DataFrame : 数据框（二维的表结构，类似hive的一张表）
写SQL的前提 : 有表
DataFrame 是基于 RDD 做了封装， 在上面提供了 列名和列类型 的概念，即表的结构的概念。
可以基于 DataFrame 去写 SQL 。
2、 写Spark SQL :
在spark sql中， shuffle之后分区数不是由前面的RDD决定的，而是有默认值， 默认200个。 可以指定参数修改。
(1) 导入Spark SQL依赖 -- 在Spark项目的pom文件中加入
<!-- Spark sql核心依赖 -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.4.5</version>
</dependency>
(2) 在Spark项目下创建sql包 -- 新的模块一定要新建新的包
*项目名称一定要小写，多个单词之间用-分割 ： s1-v1_1.2
*包名也要小写，一般是公司域名倒写 ： com.shujia.spark
(3) 创建Spark sql环境 :
val spark: SparkSession = SparkSession
.builder() // 构建
.appName("wordCount")
.master("local")
// 设置 sparkSQL 在 shuffle 之后 DF 的分区数，默认是200
.config("spark.sql.shuffle.partitions", 1)
.getOrCreate() // 当前环境有SparkSession就获取, 反之则创建
(4) 返回值不再是 RDD, 而是 DataFrame (DF)
查看数据不再是 foreach(), 而是 show()
(5) 针对于sql语句有多行的情况， 可以使用 """ """ 格式书写
val wordCountDF = spark.sql(
"""
|select word,count(1) as c from (
|select explode(split(line,',')) word from lines
|) as d
|group by word
|""".stripMargin) // stripMargin : 删除"|" 并合并以上sql语句
(6) 创建 DataFrame 的方式:
1、 读取 csv 格式的数据创建 DF
val studentDF: DataFrame = spark
.read
.format("csv")
.option("sep", ",") //列的分割方式
.schema("id STRING, name STRING, age INT, gender STRING, clazz STRING") // 指定字段名和字段类型， 必须按照数据顺序指定
.load("data/students.txt") //指定读取的路径
2、 读取 json 格式的数据构建 DF
(spark 会自动解析json格式)
val studentJsonDF: DataFrame = spark
.read
.format("json")
.load("data/a.json")
3、 读取 jdbc 数据构建 DF
(通过网络远程读取 mysql 中的数据, 需要添加mysql依赖)
val jdbcDF: DataFrame = spark.read
.format("jdbc")
.option("url", "jdbc:mysql://master:3306")
.option("dbtable", "bigdata.students")
.option("user", "root")
.option("password", "123456")
.load()
4、 读取 parquet 格式的数据构建 DF
(parquet格式的数据中自带 列名 和 列类型，
parquet会对数据进行压缩， 体积变小， 解压和压缩需要时间)
// 保存一个parquet格式的文件
studentDF
.write
.format("parquet")
.mode(SaveMode.Overwrite)
.save("data/parquet")
// 读取parquet格式的数据
val parquetDF: DataFrame = spark
.read
.format("parquet")
.load("data/parquet")
3、 DSL语法 -- 类sql语法
// spark sql 中必须要导入隐式转换， 才可以使用 $方法 获取列对象
import spark.implicits._
//导入 DSL 所有的函数
import org.apache.spark.sql.functions._
(1) show : 查看前面20条数据, 相当于action算子
action算子 -- 每一个Action算子都会触发一个job
(2) select : 选择字段, 和 sql 中 select 是一样
(3) $ : 是一个方法，作用是通过列名获取列的对象
studentDF.select($"id", $"age" + 2 as "age").show()
(4) where : 过滤数据
= : 赋值 == : 判断 === : 等于
(5) group by : 分组
(6) agg : 分组之后进行聚合计算
只能在分组后使用， 即一般跟在group函数后面
studentDF
.groupBy($"clazz")
// 分组之后做聚合计算 -- 可以写多个
.agg(count($"clazz") as "c", avg($"age") as "avgAge")
.show()
(7) join : 表关联
(8) 开窗函数 -- 统计每个班级总分前2的学生
withColumn : 给 DF 增加新的列
joinDF
// 按照 id 和 班级 分组
.groupBy($"id", $"clazz")
// 对分数求和
.agg(sum($"sco") as "sumSco")
// 使用开窗函数 -- row_number() over (partition by clazz order by sumSco desc)
// .select($"id", $"clazz", $"sumSco", row_number() over Window.partitionBy($"clazz").orderBy($"sumSco".desc) as "r")
// 在前面 DF 的基础上增加列 ( 上面的简写, 省去写 $"id", $"clazz", $"sumSco" 步骤, 直接将 "r" 加在 "sumSco" 后面 )
.withColumn("r", row_number() over Window.partitionBy($"clazz").orderBy($"sumSco".desc))
// 取 班级前2
.where($"r" <= 2).show()
(9) orderBy : 排序
DSL 语法 与 SQL 的异同 :
1、 DSL 和 SQL 功能相同, 但写法不同, 代码更简洁
2、 DSL 不需要做 子查询
