前言

在没有sparkSql之前，进行数据分析一般通过MR，SQL
但是MR运行慢，操作较为复杂，即效率比较低。
SQL运行速度较快，操作也方便。但是面向的是关系型数据库，并不适合大数据开发
模仿SQL–>HIVE
HIVE是将HIVE SQL 转换成MapReduce然后提交到集群上执行，虽然简化了Mapredece程序复杂性，但是由于MapReduce计算模型自身限制，速度并不是很快
spark模仿hive，自己形成了SparkSQL，将Spark SQL 转化为RDD，然后提交到集群执行，执行效率非常快

什么是Spark SQL

Spark SQL 是Spark用来处理结构化数据的一个模块，
但是Spark 中数据（RDD）并没有结构，怎么办呢？
在SparkSQL中Spark为我们提供了两个新的抽象
DataFrame
DataSet

Spark SQL 特点
1）Integrated(易整合)
2）Uniform Data Access(统一的数据访问方式)
3）Hive Integration(集成 Hive)
4）Standard Connectivity(标准的连接方式)

DataFrame

与RDD类似，DataFrame 是一个分布式数据容器，DataFrame 更像传统数据库的二位表格，除了数据以外，还记录数据的结构信息（即schema），强调的是结构，和属性没有太大关系

上图直观地体现了DataFrame和RDD的区别。
1）左侧的RDD[Person]虽然以Person为类型参数，但Spark框架本身不了解Person类的内部结构。
2）而右侧的DataFrame却提供了详细的结构信息，使得 Spark SQL 可以清楚地知道该数据集中包含哪些列，每列的名称和类型各是什么。
3）DataFrame是为数据提供了Schema的视图。可以把它当做数据库中的一张表来对待，
4）DataFrame也是懒执行的

性能上比 RDD要高，主要原因： 优化的执行计划：查询计划通过Spark catalyst optimiser进行优化

比如下面一个例子:

为了说明查询优化，我们来看上图展示的人口数据分析的示例。图中构造了两个DataFrame，将它们join之后又做了一次filter操作。
如果原封不动地执行这个执行计划，最终的执行效率是不高的。因为join是一个代价较大的操作，也可能会产生一个较大的数据集。
如果我们能将filter下推到 join下方，先对DataFrame进行过滤，再join过滤后的较小的结果集，便可以有效缩短执行时间。
而Spark SQL的查询优化器正是这样做的。简而言之，逻辑查询计划优化就是一个利用基于关系代数的等价变换，将高成本的操作替换为低成本操作的过程。

DataFrame 基本操作

SparkSession

在老的版本中，SparkSQL 提供两种 SQL 查询起始点：一个叫SQLContext，用于Spark 自己提供的 SQL 查询；一个叫 HiveContext，用于连接 Hive 的查询。
从2.0开始, SparkSession是 Spark 最新的 SQL 查询起始点，实质上是SQLContext和HiveContext的组合，所以在SQLContext和HiveContext上可用的 API 在SparkSession上同样是可以使用的。
SparkSession内部封装了SparkContext，所以计算实际上是由SparkContext完成的。
当我们使用 spark-shell 的时候, spark 会自动的创建一个叫做spark的SparkSession, 就像我们以前可以自动获取到一个sc来表示SparkContext

创建DataFrame

有了 SparkSession 之后, 通过 SparkSession有 3 种方式来创建DataFrame:

1） 通过 Spark 的数据源创建

spark支持的数据源：

通过spark.read 读取的文件会自动转化为DataFrame
案例：
创建一个json文件

{"name" = "Michae","salary"=3000}
{"name" = "Andy","salary"=4500}
{"name" = "Justin","salary"=3500}
{"name" = "Berta","salary"=4000}

命令行模式：

scala> val df = spark.read.json("file:///.../test.json")
scala> df.show结果：
+-------+------+
|   name|salary|
+-------+------+
|Michael|  3000|
|   Andy|  4500|
| Justin|  3500|
|  Berta|  4000|
+-------+------+

将查询结果转化为视图createTempView()，没有转化为临时表createTempTable()是因为视图不可变，

scala> df.createTempView("test")
scala> spark.sql("select * from test").show
+-------+------+
|   name|salary|
+-------+------+
|Michael|  3000|
|   Andy|  4500|
| Justin|  3500|
|  Berta|  4000|
+-------+------+
scala> spark.sql("select name from test").show
+-------+
|   name|
+-------+
|Michael|  
|   Andy|  
| Justin|  
|  Berta| 
+-------+

注意:⭐⭐⭐⭐

1. 临时视图只能在当前 Session 有效, 在新的 Session 中无效.
2. 可以创建全局视图. 访问全局视图需要全路径:如
   global_temp.xxx

scala> val df = spark.read.json("file///..../people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]#创建全局的临时表
scala> df.createGlobalTempView("people")#访问需要全路径global_temp.people
scala> spark.sql("select * from global_temp.people")
res31: org.apache.spark.sql.DataFrame = [age: bigint, name: string]scala> res31.show
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+scala> spark.newSession.sql("select * from global_temp.people")
res33: org.apache.spark.sql.DataFrame = [age: bigint, name: string]scala> res33.show
+----+-------+
| age|   name|
+----+-------+
|null|Michael|
|  30|   Andy|
|  19| Justin|
+----+-------+

DSL语法风格(了解)

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

scala> val df = spark.read.json("/.../people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]scala> df.printSchema
root
|-- age: long (nullable = true)
|-- name: string (nullable = true)

使用DSL查询

1. 只查询name列数据

scala> df.select($"name").show
+-------+
|   name|
+-------+
|Michael|
|   Andy|
| Justin|
+-------+scala> df.select("name").show
+-------+
|   name|
+-------+
|Michael|
|   Andy|
| Justin|
+-------+

2. 查询name和age

scala> df.select("name", "age").show
+-------+----+
|   name| age|
+-------+----+
|Michael|null|
|   Andy|  30|
| Justin|  19|
+-------+----+

3. 查询name和age + 1

scala> df.select($"name", $"age" + 1).show
+-------+---------+
|   name|(age + 1)|
+-------+---------+
|Michael|     null|
|   Andy|       31|
| Justin|       20|
+-------+---------+

注意:
4. 设计到运算的时候, 每列都必须使用$

5. 查询age大于20的数据

scala> df.filter($"age" > 21).show
+---+----+
|age|name|
+---+----+
| 30|Andy|
+---+----+

6. 按照age分组，查看数据条数

scala> df.groupBy("age").count.show
+----+-----+
| age|count|
+----+-----+
|  19|    1|
|null|    1|
|  30|    1|
+----+-----+

2） RDD转化为DataFrame

注意：如果需要RDD与DF或者DS之间操作，那么都需要引入 import spark.implicits._ 【spark不是包名，而是sparkSession对象的名称】
前置条件：导入隐式转换并创建一个RDD

通过手动确定转换

案例一：

scala> import spark.implicits._
scala> val rdd = sc.makeRDD(List(1,2,3,4))
scala> val df = rdd.toDF("id")
scala> df.show()
+----+
|  id|
+----+
|   1|  
|   2| 
|   3|
|   4|
+----+

案例二：

scala> import spark.implicits._
scala> val rdd = sc.makeRDD(List((1,"zhangsan",20),(2,"lisi",18),(3,"wangwu",25)))
scala> val df = rdd.toDF("id","name","age")
scala> df.show()
+-------+---------+------+
|   id  |     name|   age|
+-------+---------+------+
|      1| zhangsan|    20|
|      2|     lisi|    18|
|      3|   wangwu|    25|
+-------+---------+------+

通过反射转化（用到样例类）

scala> case class People(name: String, age: Int)
scala> val rdd = sc.makeRDD(List( ("zhangsan",20), ("lisi", 25), ("wangwu", 18) ) )
scala> val peopleRDD = rdd.map( t => { People(t._1, t._2) } ) 
#此处toDF没有指定结构，因为peopleRDD已经有了结构
scala> val df = peopleRDD.toDF
scala> df.show()
+---------+------+
|     name|   age|
+---------+------+
| zhangsan|    20|
|     lisi|    18|
|   wangwu|    25|
+---------+------+

3)通过hive创建RDD

其它操作

DataFrame转化为RDD

直接使用rdd

scala> val dfToRDD = df.rdd
dfToRDD: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[19] at rdd at <console>:29scala> dfToRDD.collect
res13: Array[org.apache.spark.sql.Row] = Array([Michael, 29], [Andy, 30], [Justin, 19])

DataSet

1. 是DataFrame API的一个扩展，是 SparkSQL 最新的数据抽象(1.6新增)。
2. 用户友好的API风格，既具有类型安全检查也具有DataFrame的查询优化特性。
3. Dataset支持编解码器，当需要访问非堆上的数据时可以避免反序列化整个对象，提高了效率。
4. 样例类被用来在DataSet中定义数据的结构信息，样例类中每个属性的名称直接映射到DataSet中的字段名称。
5. DataFrame是DataSet的特列，DataFrame=DataSet[Row] ，所以可以通过as方法将DataFrame转换为DataSet。Row是一个类型，跟Car、Person这些的类型一样，所有的表结构信息都用Row来表示。
6. DataSet是强类型的。比如可以有DataSet[Car]，DataSet[Person].
7. DataFrame只是知道字段，但是不知道字段的类型，所以在执行这些操作的时候是没办法在编译的时候检查是否类型失败的，比如你可以对一个String进行减法操作，在执行的时候才报错，而DataSet不仅仅知道字段，而且知道字段类型，所以有更严格的错误检查。就跟JSON对象和类对象之间的类比。

DataSet基本操作

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

创建

1）创建一个样例类

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

2）创建DataSet

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

3）访问：
直接访问

scala> caseClassDS.show

转换为临时视图访问

scala> caseClassDS.createTempView("test")
scala> spark.sql(select * from test).show

RDD转换为DataSet

RDD没有结构，也没有类型，
DataSet有结构，也有类型。

SparkSQL能够自动将包含有case类的RDD转换成DataFrame，case类定义了table的结构，case类属性通过反射变成了表的列名。
#创建一个样例类

scala> case class Person(name: String, age: Int)
scala> val rdd = sc.makeRDD(List( ("zhangsan",20), ("lisi", 25), ("wangwu", 18) ) )
scala> val mapRDD = rdd.map{ t => Person{ t._1, t._2 } }
scala> val ds = mapRDD.toDS
scala> ds.show+---------+------+
|     name|   age|
+---------+------+
| zhangsan|    20|
|     lisi|    18|
|   wangwu|    25|
+---------+------+

DataSet 转换为RDD

调用rdd方法即可。

scala> ds.rdd

DataFrame 转换为DataSet

#导入隐式转换
scala> import spark.implicits._
#创建样例类
scala> case class Person(name: String, age: Int)
scala> val rdd = sc.makeRDD(List( ("zhangsan",20), ("lisi", 25), ("wangwu", 18) ) )
scala> val df = rdd.toDF("name", "age")
scala> val ds = df.as[Person]
res14: org.apache.spark.sql.Dataset[Person] = [age: int, name: string]

DataSet 转换为 DataFrame

#导入隐式转换
scala> import spark.implicits._
scala> ds.toDF

RDD.DataFrame.DataSet 小结

三者的共性

1、RDD、DataFrame、Dataset全都是spark平台下的分布式弹性数据集，为处理超大型数据提供便利
2、三者都有惰性机制，在进行创建、转换，如map方法时，不会立即执行，只有在遇到Action如foreach时，三者才会开始遍历运算。
3、三者都会根据spark的内存情况自动缓存运算，这样即使数据量很大，也不用担心会内存溢出。
4、三者都有partition的概念
5、三者有许多共同的函数，如filter，排序等
想查看有哪些算子可以操作，可以输入
rdd.
df.
ds.
然后按tab键，自动补全方法
6、在对DataFrame和Dataset进行操作许多操作都需要这个包进行支持

import spark.implicits._

7、DataFrame和Dataset均可使用模式匹配获取各个字段的值和类型
DataFrame:

testDF.map{case Row(col1:String,col2:Int)=>println(col1);println(col2)col1case _=>""}

Dataset:

case class Coltest(col1:String,col2:Int)extends Serializable //定义字段名和类型testDS.map{case Coltest(col1:String,col2:Int)=>println(col1);println(col2)col1case _=>""}

三者的区别

1. RDD:

1）RDD一般和spark mlib同时使用
2）RDD不支持sparksql操作

2. DataFrame:

1）与RDD和Dataset不同，DataFrame每一行的类型固定为Row，每一列的值没法直接访问，只有通过解析才能获取各个字段的值，如：

testDF.foreach{line =>val col1=line.getAs[String]("col1")val col2=line.getAs[String]("col2")
}

2）DataFrame与Dataset一般不与spark mlib同时使用
3）DataFrame与Dataset均支持sparksql的操作，比如select，groupby之类，还能注册临时表/视窗，进行sql语句操作，如：
dataDF.createOrReplaceTempView(“tmp”)
spark.sql(“select ROW,DATE from tmp where DATE is not null order by DATE”).show(100,false)
4）DataFrame与Dataset支持一些特别方便的保存方式，比如保存成csv，可以带上表头，这样每一列的字段名一目了然

//保存
val saveoptions = Map("header" -> "true", "delimiter" -> "\t", "path" -> "hdfs://hadoop102:9000/test")
datawDF.write.format("com.atguigu.spark.csv").mode(SaveMode.Overwrite).options(saveoptions).save()
//读取
val options = Map("header" -> "true", "delimiter" -> "\t", "path" -> "hdfs://hadoop102:9000/test")
val datarDF= spark.read.options(options).format("com.atguigu.spark.csv").load()

利用这样的保存方式，可以方便的获得字段名和列的对应，而且分隔符（delimiter）可以自由指定。

3. Dataset:

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

case class Coltest(col1:String,col2:Int)extends Serializable //定义字段名和类型
/**rdd("a", 1)("b", 1)("a", 1)
**/
val test: Dataset[Coltest]=rdd.map{line=>Coltest(line._1,line._2)}.toDS
test.map{line=>println(line.col1)println(line.col2)}

可以看出，Dataset在需要访问列中的某个字段时是非常方便的，然而，如果要写一些适配性很强的函数时，如果使用Dataset，行的类型又不确定，可能是各种case class，无法实现适配，这时候用DataFrame即Dataset[Row]就能比较好的解决问题

使用IDEA开发SparkSQL程序

创建一个DataFrame

user.json

[ {"username": "zhangsan","password": "1"},{"username": "lisi","password": "2"},{"username": "wangwu","password": "3"}]

代码：

import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}
import org.slf4j.LoggerFactoryobject HelloWorld {def main(args: Array[String]) {//创建SparkConf()并设置App名称val spark = SparkSession.Builder().appName("Spark SQL basic example").config("spark.some.config.option", "some-value").getOrCreate()// For implicit conversions like converting RDDs to DataFramesimport spark.implicits._//创建DataFrameval df = spark.read.json("in/user.json")// Displays the content of the DataFrame to stdoutdf.show()//将DataFrame转换为一张表df.createOrReplaceTempView("user")//采用sql 的语法访问数据spark.sql("select * from user").showspark.stop()}}

转换

import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}
import org.slf4j.LoggerFactoryobject HelloWorld {def main(args: Array[String]) {//创建SparkConf()并设置App名称val spark = SparkSession.Builder().appName("Spark SQL basic example").config("spark.some.config.option", "some-value").getOrCreate()// 引入隐式转换，这里spark并不是一个包，spark是刚刚创建的SparkSession对象import spark.implicits._//创建RDDval rdd = spark.sparkContext.makeRDD(List( (1,"zhangsan",20), (2,"lisi",18), (3,"wangwu", 25) ) )//转换为DFval df:DataFrame = rdd.toDF("id","name","age")//转换为DSval ds:DataSet[User] = df.as[User]//转换为DFval df1: DataFrmae = ds.toDF() //转换为RDDval rdd1:RDD[Row] = df1.rddrdd1.foreach(row => {//获取数据时，可以通过索引访问数据println(row.getString(1))})//RDD->DataSetval userRDD = rdd.map{case (id, name, age)=>{User(id, name, age)}}val userDS: Data[User] = userRDD.toDs()//DataSt->RDDval rdd2: RDD[User] = userDS.rdd		rdd2.foreach(println)//释放资源spark.stop}
}case class User( id: Int, name: String, age: Int )