1、Spark SQL简介
Spark SQL是Spark用来处理结构化数据的一个模块，它提供了一个编程抽象叫做DataFrame并且作为分布式SQL查询引擎的作用。
为什么要学习Spark SQL？我们已经学习了Hive，它是将Hive SQL转换成MapReduce然后提交到集群上执行，大大简化了编写MapReduce的程序的复杂性，由于MapReduce这种计算模型执行效率比较慢。所以Spark SQL的应运而生，它是将Spark SQL转换成RDD，然后提交到集群执行，执行效率非常快！同时Spark SQL也支持从Hive中读取数据。
Spark SQL的特点：
容易整合（集成）
统一的数据访问方式
兼容Hive
标准的数据连接
2、基本概念：Datasets和DataFrames
DataFrame
DataFrame是组织成命名列的数据集。它在概念上等同于关系数据库中的表，但在底层具有更丰富的优化。DataFrames可以从各种来源构建，
例如：
结构化数据文件
hive中的表
外部数据库或现有RDDs
DataFrame API支持的语言有Scala，Java，Python和R。
从上图可以看出，DataFrame多了数据的结构信息，即schema。RDD是分布式的 Java对象的集合。DataFrame是分布式的Row对象的集合。DataFrame除了提供了比RDD更丰富的算子以外，更重要的特点是提升执行效率、减少数据读取以及执行计划的优化
Datasets
Dataset是数据的分布式集合。Dataset是在Spark 1.6中添加的一个新接口，是DataFrame之上更高一级的抽象。它提供了RDD的优点（强类型化，使用强大的lambda函数的能力）以及Spark SQL优化后的执行引擎的优点。一个Dataset 可以从JVM对象构造，然后使用函数转换（map， flflatMap，fifilter等）去操作。 Dataset API 支持Scala和Java。 Python不支持Dataset API。
3、测试数据
使用员工表的数据，并已经将其保存到了HDFS上。
4、创建DataFrames
（*）通过Case Class创建DataFrames
① 定义case class（相当于表的结构：Schema）
注意：由于mgr和comm列中包含null值，简单起见，将对应的case class类型定义为String
case class
Emp(empno:Int,Ename:String,job:String,mgr:String,hiredate:String,sal:Int,comm:String,deptno:Int)
② 将HDFS上的数据读入RDD，并将RDD与case Class关联
val lines = sc.textFile("hdfs://hadoop111:9000/scott/emp.csv").map(_.split(","))
val allEmp = lines.map(x =>Emp(x(0).toInt,x(1),x(2),x(3),x(4),x(5).toInt,x(6),x(7).toInt,))
③ 将RDD转换成DataFrames
val allEmpDF = allEmp.toDF
④ 通过DataFrames查询数据
（*）使用SparkSession
① 什么是SparkSession
Apache Spark 2.0引入了SparkSession，其为用户提供了一个统一的切入点来使用Spark的各项功能，并且允许用户通过它调用DataFrame和Dataset相关API来编写Spark程序。最重要的是，它减少了用户需要了解的一些概念，使得我们可以很容易地与Spark交互。
在2.0版本之前，与Spark交互之前必须先创建SparkConf和SparkContext。然而在Spark 2.0中，我们可以通过SparkSession来实现同样的功能，而不需要显式地创建SparkConf, SparkContext 以及SQLContext，因为这些对象已经封装在SparkSession中。
② 创建StructType，来定义Schema结构信息
注意，需要：import org.apache.spark.sql.types._
③ 读入数据并且切分数据
④ 将RDD中的数据映射成Row
⑤ 创建DataFrames
val df = spark.createDataFrame(rowRDD,myschema)
再举一个例子，使用JSon文件来创建DataFame
① 源文件：
$SPARK_HOME/examples/src/main/resources/people.json
②读取文件
val df = spark.read.json("源文件")
③ 查看数据和Schema信息
5、DataFrame操作
DataFrame操作也称为无类型的Dataset操作
（*）查询所有的员工姓名
（*）查询所有的员工姓名和薪水，并给薪水加100块钱
（*）查询工资大于2000的员工
（*）求每个部门的员工人数
其他的例子，请参考API：
http://spark.apache.org/docs/2.1.0/api/scala/index.html#org.apache.spark.sql.Dataset
（*）在DataFrame中使用SQL语句
① 将DataFrame注册成表（视图）：
df.createOrReplaceTempView("emp")
② 执行查询：
spark.sql("select * from emp").show
spark.sql("select * from emp where deptno=10").show
spark.sql("select deptno,sum(sal) from emp group by deptno").show
6、Global Temporary View
上面使用的是一个在Session生命周期中的临时views。在Spark SQL中，如果你想拥有一个临时的view，并想在不同的Session中共享，而且在application的运行周期内可用，那么就需要创建一个全局的临时view。并记得使用的时候加上global_temp作为前缀来引用它，因为全局的临时view是绑定到系统保留的数据库global_temp上。
① 创建一个普通的view和一个全局的view
df.createOrReplaceTempView("emp1")
df.createGlobalTempView("emp2")
② 在当前会话中执行查询，均可查询出结果。
spark.sql("select * from emp1").show
spark.sql("select * from global_temp.emp2").show
③ 开启一个新的会话，执行同样的查询
spark.newSession.sql("select * from emp1").show **（运行出错）**
spark.newSession.sql("select * from global_temp.emp2").show
7、创建Datasets
DataFrame的引入，可以让Spark更好的处理结构数据的计算，但其中一个主要的问题是：缺乏编译时类型安全。为了解决这个问题，Spark采用新的Dataset API (DataFrame API的类型扩展)。
Dataset是一个分布式的数据收集器。这是在Spark1.6之后新加的一个接口，兼顾了RDD的优点（强类型，可以使用功能强大的lambda）以及Spark SQL的执行器高效性的优点。所以可以把DataFrames看成是一种特殊的Datasets，即：Dataset(Row)
（*）创建DataSet，方式一：使用序列
1、定义case class
case class MyData(a:Int,b:String)
2、生成序列，并创建DataSet
val ds = Seq(MyData(1,"Tom"),MyData(2,"Mary")).toDS
3、查看结果
（*）创建DataSet，方式二：使用JSON数据
1、定义case class
case class Person(name: String, gender: String)
2、通过JSON数据生成DataFrame
val df = spark.read.json(sc.parallelize("""{"gender": "Male", "name": "Tom"}""" :: Nil))
3、将DataFrame转成DataSet
df.as[Person].show
df.as[Person].collect
（*）创建DataSet，方式三：使用HDFS数据
1、读取HDFS数据，并创建DataSet
val linesDS = spark.read.text("hdfs://hadoop111:9000/data/data.txt").as[String]
2、对DataSet进行操作：分词后，查询长度大于3的单词
val words = linesDS.flatMap(_.split(" ")).filter(_.length > 3)
words.show
words.collect
3、执行WordCount程序
val result = linesDS.flatMap(_.split(" ")).map((_,1)).groupByKey(x => x._1).count
result.show
排序：result.orderBy($"value").show
8、Datasets的操作案例
（*）使用emp.json 生成DataFrame
val empDF = spark.read.json("/root/resources/emp.json")
查询工资大于3000的员工
empDF.where($"sal" >= 3000).show
（*）创建case class
case class
Emp(empno:Long,ename:String,job:String,hiredate:String,mgr:String,sal:Long,comm:String,deptno:Long)
（*）生成DataSets，并查询数据
val empDS = empDF.as[Emp]
查询工资大于3000的员工
empDS.filter(_.sal > 3000).show
查看10号部门的员工
empDS.filter(_.deptno == 10).show
（*）多表查询
1、创建部门表
val deptRDD=sc.textFile("/root/temp/dept.csv").map(_.split(","))
case class Dept(deptno:Int,dname:String,loc:String)
val deptDS = deptRDD.map(x=>Dept(x(0).toInt,x(1),x(2))).toDS
2、创建员工表
case class
Emp(empno:Int,ename:String,job:String,mgr:String,hiredate:String,sal:Int,comm:St
ring,deptno:Int)
val empRDD = sc.textFile("/root/temp/emp.csv").map(_.split(","))
val empDS = empRDD.map(x =>
Emp(x(0).toInt,x(1),x(2),x(3),x(4),x(5).toInt,x(6),x(7).toInt)).toDS
3、执行多表查询：等值链接
val result = deptDS.join(empDS,"deptno")
另一种写法：注意有三个等号
val result = deptDS.joinWith(empDS,deptDS("deptno")===empDS("deptno"))
joinWith和join的区别是连接后的新Dataset的schema会不一样
（*）查看执行计划：result.explain
