1.什么是spark sql?
Spark SQL是Spark用来处理结构化数据的一个模块.包括DataFrame和DataSet，并且作为分布式SQL查询引擎的作用。
2.什么DataFrame?
DataFrame也是一个分布式数据容器。然而DataFrame更像传统数据库的二维表格，除了数据以外，还记录数据的结构信息，即schema
Spark session 是spark sql编程的起点。SparkSession内部封装了sparkContext，所以计算实际上是由sparkContext完成的。
3.DataFrame创建?
在Spark SQL中SparkSession是创建DataFrame和执行SQL的入口，创建DataFrame有三种方式：
1.通过Spark的数据源进行创建
2.从一个存在的RDD进行转换
1.通过Spark的数据源进行创建
（1）查看Spark数据源进行创建的文件格式
scala> spark.read.
csv format jdbc json load option options orc parquet schema table text textFile
（2）读取json文件创建DataFrame
scala> val df = spark.read.json("/opt/module/spark/examples/src/main/resources/people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
（3）展示结果
scala> df.show
±—±------+
| age| name|
±—±------+
|null|Michael|
| 30| Andy|
| 19| Justin|
±—±------+
DataFrame
SQL风格语法(主要)
1）创建一个DataFrame
scala> val df = spark.read.json("/opt/module/spark/examples/src/main/resources/people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
2）对DataFrame创建一个临时表
scala> df.createOrReplaceTempView(“people”)
3）通过SQL语句实现查询全表
scala> val sqlDF = spark.sql(“SELECT * FROM people”)
sqlDF: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
4）结果展示
scala> sqlDF.show
±—±------+
| age| name|
±—±------+
|null|Michael|
| 30| Andy|
| 19| Justin|
±—±------+
注意：临时表是Session范围内的，Session退出后，表就失效了。如果想应用范围内有效，可以使用全局表。注意使用全局表时需要全路径访问，如：global_temp.people
5）对于DataFrame创建一个全局表
scala> df.createGlobalTempView(“people”)
6）通过SQL语句实现查询全表
scala> spark.sql(“SELECT * FROM global_temp.people”).show()
±—±------+
| age| name|
±—±------+
|null|Michael|
| 30| Andy|
| 19| Justin|
scala> spark.newSession().sql(“SELECT * FROM global_temp.people”).show()
±—±------+
| age| name|
±—±------+
|null|Michael|
| 30| Andy|
| 19| Justin|
±—±------+
4.RDD与DataFrame和Dataset转换
4.1 rdd->dataframe
注意：如果需要RDD与DF或者DS之间操作，那么都需要引入 import spark.implicits._ 【spark不是包名，而是sparkSession对象的名称】
前置条件：导入隐式转换并创建一个RDD
方式一:Rdd->dataframe:1.导入包2.手动去取值，并且调用toDF()
scala> import spark.implicits._
import spark.implicits._
scala> val peopleRDD = sc.textFile(“examples/src/main/resources/people.txt”)
peopleRDD: org.apache.spark.rdd.RDD[String] = examples/src/main/resources/people.txt MapPartitionsRDD[3] at textFile at :27
1）通过手动确定转换
scala> peopleRDD.map{x=>val para = x.split(",");(para(0),para(1).trim.toInt)}.toDF(“name”,“age”)
res1: org.apache.spark.sql.DataFrame = [name: string, age: int]
方式二:
过反射确定（需要用到样例类）
（1）创建一个样例类
scala> case class People(name:String, age:Int)
（2）根据样例类将RDD转换为DataFrame
scala> peopleRDD.map{ x => val para = x.split(",");People(para(0),para(1).trim.toInt)}.toDF
res2: org.apache.spark.sql.DataFrame = [name: string, age: int]
4.2.DateFrame转换为RDD
直接调用rdd即可
1）创建一个DataFrame
scala> val df = spark.read.json("/opt/module/spark/examples/src/main/resources/people.json")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
2）将DataFrame转换为RDD
scala> val dfToRDD = df.rdd
dfToRDD: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[19] at rdd at :29
3）打印RDD
scala> dfToRDD.collect
res13: Array[org.apache.spark.sql.Row] = Array([Michael, 29], [Andy, 30], [Justin, 19])
4.3 DataSet(用toDs)
Dataset是具有强类型的数据集合，需要提供对应的类型信息
4.3.1 创建
1）创建一个样例类
scala> case class Person(name: String, age: Long)
defined class Person
2）创建DataSet–sequence
scala> val caseClassDS = Seq(Person(“Andy”, 32)).toDS()
caseClassDS: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]
4.3.2RDD转换为DataSet
Rdd->dataframe->dataset
SparkSQL能够自动将包含有case类的RDD转换成DataFrame，case类定义了table的结构，case类属性通过反射变成了表的列名。
1）创建一个RDD
scala> val peopleRDD = sc.textFile(“examples/src/main/resources/people.txt”)
peopleRDD: org.apache.spark.rdd.RDD[String] = examples/src/main/resources/people.txt MapPartitionsRDD[3] at textFile at :27
2）创建一个样例类
scala> case class Person(name: String, age: Long)
defined class Person
3）将RDD转化为DataSet–>dataframe->dataset
scala> peopleRDD.map(line => {val para = line.split(",");Person(para(0),para(1).trim.toInt)}).toDS()
4.3.3 DataSet转换为RDD
调用rdd方法即可。
1）创建一个DataSet
scala> val DS = Seq(Person(“Andy”, 32)).toDS()
DS: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]
2）将DataSet转换为RDD
scala> DS.rdd
res11: org.apache.spark.rdd.RDD[Person] = MapPartitionsRDD[15] at rdd at :28
代码层面上的
DataSet转DataFrame
这个很简单，因为只是把case class封装成Row
（1）导入隐式转换
import spark.implicits._
（2）转换
val testDF = testDS.toDF
DataFrame转DataSet
（1）导入隐式转换
import spark.implicits._
（2）创建样例类
case class Coltest(col1:String,col2:Int)extends Serializable //定义字段名和类型
（3）转换
val testDS = testDF.as[Coltest]
DataFrame:
1）与RDD和Dataset不同，DataFrame每一行的类型固定为Row，每一列的值没法直接访问，只有通过解析才能获取各个字段的值，如：
testDF.foreach{line =>val col1=line.getAs[String]("col1")val col2=line.getAs[String]("col2")
}
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)
IDEA创建SparkSQL程序
IDEA中程序的打包和运行方式都和SparkCore类似，Maven依赖中需要添加新的依赖项：
<dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.11</artifactId><version>2.1.1</version>
</dependency>
程序如下：
package com.atguigu.sparksqlimport 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._val df = spark.read.json("data/people.json")// Displays the content of the DataFrame to stdoutdf.show()df.filter($"age" > 21).show()df.createOrReplaceTempView("persons")spark.sql("SELECT * FROM persons where age > 21").show()spark.stop()}}