什么是Spark SQL
DataFrame和Dataset
Spark SQL基本使用
Spark SQL数据源
Spark SQL内置函数


什么是Spark SQL

Spark SQL  结构化数据处理 
Schema信息
json parquet avro csv
与基础的Spark RDD API不同，Spark SQL提供了对结构化数据的查询和计算接口

Spark SQL的主要特点
1 将SQL查询与Spark应用程序无缝结合
  使用SQL在Spark程序中查询结构化数据
  与Hive区别
  Hive将SQL翻译成MapReduce作业
  Spark SQL底层使用 Spark RDD 
  results = spark.sql( "SELECT * FROM people")
2 以相同的方式连接到多种数据源
  Spark SQL提供了访问各种数据源的通用方法
  数据源包括 Hive Avro Parquet ORC JSON JDBC等
  
  //读取JSON文件
  val userScoreDF = spark.read.json("hdfs://centos01:9000/people.json")
  //创建临时视图user_score
  userScoreDF.createTempView("user_score")
  //根据name关联查询
  val resDF=spark.sql(
  "SELECT i.age,i.name,c.score FROM user_info i JOIN user_score c ON i.name=c.name"
  )
                    
3 在现有的数据仓库上运行SQL或HiveQL查询
  Spark SQL支持HiveQL语法以及Hive SerDes和UDF（用户自定义函数）
  允许访问现有的Hive仓库


DataFrame和Dataset

DataFrame是Spark SQL提供的一个编程抽象
与RDD类似，也是一个分布式的数据集合
DataFrame的数据都被组织到有名字的列中，就像关系型数据库中的表一样
多种数据都可以转化为DataFrame
例如：Spark计算过程中生成的RDD、结构化数据文件、Hive中的表、外部数据库等
DataFrame在RDD的基础上添加了数据描述信息（Schema）
因此看起来更像是一张数据库表


使用DataFrame API结合SQL处理结构化数据比RDD更加的容易
Spark优化器会自动对其优化 
Dataset也是一个分布式数据集，是Spark1.6中添加的一个新的API
相对于RDD，Dataset提供了强类型支持
使用Dataset API同样会经过Spark SQL优化器的优化，提高执行效率

DataFrame 是 Dataset[Row] 的一个别名


Spark SQL基本使用

Spark Shell 内置变量  
sc  SparkContext 
spark  SparkSession 
SparkSession是在SparkContext基础上的封装，应用程序的入口仍然是SparkContext
SparkSession 可调用DataFrame和Dataset API
支持从不同的数据源加载数据，并把数据转换成DataFrame，然后使用SQL语句来操作DataFrame 

HDFS文件 /input/person.txt 
1,zhangsan,25
2,lisi,22
3,wangwu,30

SparkSession  read.textFile()
scala> val d1=spark.read.textFile("hdfs://centos01:9000/input/person.txt")
d1: org.apache.spark.sql.Dataset[String] = [value: string]

csv() jdbc() json()等方法读取csv文件 jdbc数据源 json文件等数据

scala> d1.show()
+-------------+                                                                 
|        value|
+-------------+
|1,zhangsan,25|
|2,lisi,22|
|3,wangwu,30|
+-------------+

列名默认为 value


给Dataset添加元数据信息
定义一个样例类Person，用于存放数据描述信息（Schema）
scala> case class Person(id:Int,name:String,age:Int)
导入SparkSession的隐式转换，以便后续使用Dataset的算子
scala> import spark.implicits._
调用Dataset的map() 转换成 Person

scala> val personDataset=d1.map(line=>{
       val fields = line.split(",")
       val id = fields(0).toInt
       val name = fields(1)
       val age = fields(2).toInt
       Person(id, name, age)
       })
       
scala> personDataset.show()
+---+--------+---+
| id|    name|age|
+---+--------+---+
|  1|zhangsan| 25|
|  2|    lisi| 22|
|  3|  wangwu| 30|
+---+--------+---+


将Dataset转为DataFrame

scala> val pdf = personDataset.toDF()
pdf: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]

执行SQL查询
scala> pdf.createTempView("v_person")

scala> val result = spark.sql("select * from v_person order by age desc")
result: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]

scala> result.show()
+---+--------+---+
| id|    name|age|
+---+--------+---+
|  3|  wangwu| 30|
|  1|zhangsan| 25|
|  2|    lisi | 22|
+---+--------+---+




Spark SQL数据源

通过DataFrame 对各种数据源进行操作
使用相关转换算子进行操作，也可以创建临时视图 然后使用SQL查询
数据加载和写入 
load() 和 save() 
load()  加载外部数据源为一个DataFrame
save()  将一个DataFrame写入到指定的数据源

默认 Parquet格式

val spark = SparkSession.builder()  
  .appName("SparkSQLDataSource")
  .master("local[*]")
  .getOrCreate()
// 加载parquet格式的文件，返回一个DataFrame集合
val usersDF = spark.read.load("hdfs://centos01:9000/users.parquet")
usersDF.show()

//查询DataFrame中的id列和name列 ，并写入HDFS
usersDF.select("id","name").write.save("hdfs://centos01:9000/result")

//创建临时视图
usersDF.createTempView("t_user")
// 执行SQL查询，并将结果写入 HDFS
spark.sql("SELECT id,name FROM t_user")
.write.save("hdfs://centos01:9000/result")

手动指定数据源
使用format()方法 手动指定数据源
数据源需要使用完全限定名（例如org.apache.spark.sql.parquet
对于Spark SQL的内置数据源，可以使用缩写名（json parquet jdbc orc libsvm csv text）

val peopleDFCsv=spark.read.format("csv").load("hdfs://centos01:9000/people.csv")
可以使用option()方法向指定的数据源传递所需参数
例如JDBC数据源 账号、密码等参数

val jdbcDF = spark.read.format("jdbc")
  .option("url", "jdbc:mysql://192.168.1.69:3306/spark_db")
  .option("driver","com.mysql.jdbc.Driver")
  .option("dbtable", "student")
  .option("user", "root")
  .option("password", "123456")
  .load()



数据写入模式
mode() 指定如何处理已经存在的数据，枚举类SaveMode 
（1）SaveMode.ErrorIfExists   默认值 如果数据已经存在，则抛出异常
（2）SaveMode.Append        
（3）SaveMode.Overwrite    覆盖（包括数据或表的Schema）
（4）SaveMode.Ignore       如果数据或表已经存在，则不会写入  

HDFS文件 JSON格式  /people.json 

{"name":"Michael"}
{"name":"Andy", "age":30}
{"name":"Justin", "age":19}

val peopleDF = spark.read.format("json").load("hdfs://centos01:9000/people.json")
peopleDF.select("name")
  .write.mode(SaveMode.Overwrite).format("json")
  .save("hdfs://centos01:9000/result")

分区自动推断
表分区是Hive等系统中常用的优化查询效率的方法（
在分区表中，数据通常存储在不同的分区目录中
分区目录命名  分区列名=值 
people 表 gender和country作为分区列
存储数据的目录结构
path
└── to
    └── people
        ├── gender=male
        │   ├── ...
        │   │
        │   ├── country=US
        │   │   └── data.parquet
        │   ├── country=CN
        │   │   └── data.parquet
        │   └── ...
        └── gender=female
            ├── ...
            │
            ├── country=US
            │   └── data.parquet
            ├── country=CN
            │   └── data.parquet
            └── ...


对于所有内置的数据源（包括Text/CSV/JSON/ORC/Parquet）
Spark SQL都能够根据目录名自动发现和推断分区信息

(1) 在本地（或HDFS）新建以下三个目录及文件
其中的目录people代表表名，gender和country代表分区列
people.json存储实际人口数据：

D:\people\gender=male\country=CN\people.json
D:\people\gender=male\country=US\people.json
D:\people\gender=female\country=CN\people.json
三个people.json文件的数据分别如下：
{"name":"zhangsan","age":32}
{"name":"lisi", "age":30}
{"name":"wangwu", "age":19}

{"name":"Michael"}
{"name":"Jack", "age":20}
{"name":"Justin", "age":18}

{"name":"xiaohong","age":17}
{"name":"xiaohua", "age":22}
{"name":"huanhuan", "age":16}

val usersDF = spark.read.format("json").load("D:\\people") 
usersDF.printSchema()  
usersDF.show()  

Schema信息 
root
 |-- age: long (nullable = true)
 |-- name: string (nullable = true)
 |-- gender: string (nullable = true)
 |-- country: string (nullable = true)

表数据
+----+--------+------+-------+
| age|    name|gender|country|
+----+--------+------+-------+
|  17|xiaohong|female|     CN|
|  22| xiaohua|female|     CN|
|  16|huanhuan|female|     CN|
|  32|zhangsan|  male|     CN|
|  30|     lisi|  male|     CN|
|  19|   wangwu|  male|     CN|
|null| Michael|  male|     US|
|  20|     Jack|  male|     US|
|  18|   Justin|  male|     US|
+----+--------+------+-------+

读取数据时，自动推断出了两个分区列gender和country，将该两列的值添加到了DataFrame中

Parquet文件
列式存储格式

Spark SQL支持对Parquet文件的读写,并且可以自动保存源数据的Schema
写入Parquet文件时，为了提高兼容性，所有列都会自动转换为 可为空 状态
加载和写入Parquet文件时，可以使用load()方法和save()
还可以直接使用Spark SQL内置的parquet()方法 
//读取Parquet文件为一个DataFrame
val usersDF = spark.read.parquet("hdfs://centos01:9000/users.parquet")
//将DataFrame相关数据保存为Parquet文件，包括Schema信息
usersDF.select("id","name")
  .write.parquet("hdfs://centos01:9000/result")


JSON数据集
JSON文件的每一行必须包含一个独立有效的JSON对象
一行一个json对象

{"name":"zhangsan","age":32}
{"name":"lisi", "age":30}
{"name":"wangwu", "age":19}



val spark = SparkSession.builder()
        .appName("SparkSQLDataSource")
        .config("spark.sql.parquet.mergeSchema",true)
        .master("local[*]")
        .getOrCreate()
 
      // 1. 创建用户基本信息表 
      import spark.implicits._
      //创建用户信息Dataset集合
      val arr=Array(
         "{'name':'zhangsan','age':20}",
         "{'name':'lisi','age':18}"
      )
      val userInfo: Dataset[String] = spark.createDataset(arr)
      //将Dataset[String]转为DataFrame
      val userInfoDF = spark.read.json(userInfo)
      //创建临时视图user_info
      userInfoDF.createTempView("user_info")
      //显示数据
      userInfoDF.show()
      // +---+--------+
      // |age|    name|
      // +---+--------+
      // | 20|zhangsan|
      // | 18|    lisi|
      // +---+--------+

 // 2. 创建用户成绩表 
      //读取JSON文件
      val userScoreDF = spark.read.json("D:\\people\\score.json")
      //创建临时视图user_score
      userScoreDF.createTempView("user_score")
      userScoreDF.show()
      // +--------+-----+
      // |    name|score|
      // +--------+-----+
      // |zhangsan|   98|
      // |    lisi|   88|
      // |  wangwu|   95|
      //      +--------+-----+
      // 3. 根据name字段关联查询 
      val resDF=spark.sql("SELECT i.age,i.name,c.score FROM user_info i " +
                              "JOIN user_score c ON i.name=c.name") 
      resDF.show()
      // +---+--------+-----+
      // |age|    name|score|
      // +---+--------+-----+
      // | 20|zhangsan|   98|
      // | 18|    lisi|   88|
      // +---+--------+-----+
      
      


Hive表


Hive有大量依赖项，这些依赖项不包括在默认的Spark发行版中
如果在classpath上配置了这些Hive依赖项，Spark将自动加载它们
Hive依赖项必须出现在所有Worker节点上，因为需要访问Hive序列化和反序列化库（SerDes）



val spark = SparkSession
  .builder()
  .appName("Spark Hive Demo")
  .enableHiveSupport()//开启Hive支持
  .getOrCreate()

创建Hive表students，并指定字段分隔符为制表符 
spark.sql("CREATE TABLE IF NOT EXISTS students (name STRING, age INT) " +
  "ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'")


导入本地数据到Hive表
本地文件/home/hadoop/students.txt 
zhangsan    20
lisi    25
wangwu  19

spark.sql("LOAD DATA LOCAL INPATH '/home/hadoop/students.txt'  INTO TABLE students")

查询表数据
spark.sql("SELECT * FROM students").show()
+--------+---+
|    name|age|
+--------+---+
|zhangsan| 20|
|     lisi| 25|
|   wangwu| 19|
+--------+---+

创建表的同时指定存储格式
创建一个Hive表hive_records，数据存储格式为Parquet（默认为普通文本格式） 
spark.sql("CREATE TABLE hive_records(key STRING, value INT) STORED AS PARQUET")


将DataFrame写入Hive表
使用saveAsTable() 将一个DataFrame写入到指定的Hive表中

//加载students表的数据为DataFrame
val studentsDF = spark.table("students")
//将DataFrame写入表hive_records中
studentsDF.write.mode(SaveMode.Overwrite).saveAsTable("hive_records")
//查询hive_records表数据并显示到控制
spark.sql("SELECT * FROM hive_records").show()

Hive数据源，提交之前需要做好Hive数据仓库、元数据库等的配置


JDBC


url 连接的 JDBC URL
driver  JDBC 驱动的类名
user    数据库用户名
password    数据库密码
dbtable 
数据库表名或能代表一张数据库表的子查询
只使用数据库表名，将查询整张表的数据
查询部分数据或多表关联查询，可以使用SQL查询

注意，不允许同时指定 dbtable 和 query 属性

query   指定查询的SQL语句

不允许同时指定 query 和 partitionColumn 属性
当需要指定 partitionColumn 属性时
可以使用 dbtable 属性指定子查询 并使用子查询的别名对分区列进行限定

partitionColumn，lowerBound,upperBound  
若有一个被指定则必须全部指定，且必须指定numPartitions属性
如何在从多个Worker中并行读取数据时对表进行分区
partitionColumn必须是表中的数字、日期或时间戳列
lowerBound和upperBound只是用来决定分区跨度
而不是用于过滤表中的行 ,因此表中的所有行都将被分区并返回


numPartitions   
对表进行并行读写数据时的最大分区数
决定了并发JDBC连接的最大数量
如果要写入数据的分区数量超过了此限制的值
在写入之前可以调用coalesce(numpartition)将分区数量减少到此限制的值



使用JDBC API对MySQL表student和表score进行关联查询 
val jdbcDF = spark.read.format("jdbc")
  .option("url", "jdbc:mysql://192.168.1.69:3306/spark_db")
  .option("driver","com.mysql.jdbc.Driver")
  .option("dbtable", "(select st.name,sc.score from student st,score sc " +
    "where st.id=sc.id) t")
  .option("user", "root")
  .option("password", "123456")
  .load()
dbtable  是一个子查询，相当于SQL查询中的FROM关键字后的一部分
使用query属性编写完整SQL语句进行查询也能达到同样的效果 

val jdbcDF = spark.read.format("jdbc")
  .option("url", "jdbc:mysql://192.168.1.234:3306/spark_db")
  .option("driver","com.mysql.jdbc.Driver")
  .option("query", "select st.name,sc.score from student st,score sc " +
    "where st.id=sc.id")
  .option("user", "root")
  .option("password", "123456")
  .load()



Spark SQL内置函数
org.apache.spark.sql.functions

函数主要分为10类
UDF函数 聚合函数 日期函数 排序函数 
非聚合函数 数学函数 混杂函数 窗口函数 字符串函数 集合函数
大部分函数与Hive相同

内置函数两种使用方式
编程方式
SQL语句中使用 


import org.apache.spark.sql.functions._
df.select(lower(col("name")).as("name")).show()


UDF  User Defined Functions
常用的聚合函数 count  countDistinct avg max min

自定义聚合函数 UDAF  User Defined Aggregate Functions

UDF主要针对单个输入，返回单个输出
UDAF 可以针对多个输入进行聚合计算返回单个输出，功能更加强大

UDAF 继承抽象类UserDefinedAggregateFunction 

开窗函数
row_number()
使用该函数可以在查询结果中对每个分组的数据，按照其排序的顺序添加一列行号（从1开始）
根据行号可以方便的对每一组数据取前N行（分组取TOPN）
row_number() over (partition by 列名 order by 列名 desc) 行号列别名


partition by 按照某一列进行分组。
order by     分组后按照某一列进行组内排序
desc         降序，默认升序