这是本人的学习过程，看到的同道中人祝福你们心若有所向往，何惧道阻且长；
但愿每一个人都像星星一样安详而从容的，不断沿着既定的目标走完自己的路程；
最后想说一句君子不隐其短，不知则问，不能则学。
如果大家觉得我写的还不错的话希望可以收获关注、点赞、收藏（谢谢大家）
一、SparkSQL概述
1.1 什么是SparkSQL
Spark是用于结构化数据处理的Spark模块。与基本的Spark RDD API不同，SparkSQL提供的接口为Spark提供了有关数据结构和正在执行的计算的更多信息。在内部，SparkSQL使用这些额外的信息来执行额外的优化。与SparkSQL交互的方式有很多种，包括SQL和DatasetAPI。结算时，使用相同的执行引擎，与你用于表计算的API/语言无关。
1.2 为什么要有SparkSQL
1.3 SparkSQL的发展
1、发展历史
RDD（Spark1.0）=> Dateframe（Spark1.3） =>Dataset（Spark1.6）
如果同样的数据都给到这三个数据结构，它们分别计算之后，都会给出相同的结果。
不同的是它们执行效率和执行方式。在现在的版本中，dataset性能最好，已经成为了唯一使用的接口。其中Dataframe已经在底层被看作是特殊泛型的DataSet。
2、三者的共性
（1）RDD、DataFrame、DataSet全都是Spark平台下的分布式弹性数据集，为处理大型数据通过便利。
（2）三者都有惰性机制，在进行创建、转换，如map方法时，不会立即执行，只有在遇到Action行动算子实，三者才会开始遍历运算。
（3）三者有许多共同的函数，例如filter，sortby等
（4）三者都会根据Spark的内存情况自动缓存运算。
（5）三者都有分区的概念
1.4 SparkSQL的特点
1、易整合：无缝的整合了SQL查询和Spark编程
2、统一的数据访问方式：使用相同的方式连接不同的数据源
3、兼容Hive：在已有的仓库上直接运行SQL或者HQL
4、标准的数据连接：通过JDBC或者ODBC来连接
二、SparkSQL 编程
2.1 SparkSession 新的起始点
在老的版本中，SparkSQL提供两种SQL查询起始点：
（1） 一个叫SQLContext，用于Spark自己提供的SQL查询；
（2）一个叫HiveContext，用于连接Hive的查询。
SparkSession是Spark最新的SQL查询起始点，实质上是SQLContext和HiveContext的组合，所以在SQLContext和HiveContext上可用的API在SparkSession上同样是可以使用的。
SparkSession内部封装了SparkContext，所以计算实际上是由SparkContext完成的。当我们使用spark-shell的时候，Spark框架会自动的创建一个名称叫做Spark的SparkSession，就像我们以前可以自动获取到一个sc来表示SparkContext。
DataFrame
Spark SQL的DataFrame API 允许我们使用 DataFrame 而不用必须去注册临时表或者生成 SQL 表达式。DataFrame API 既有 transformation操作也有action操作。
创建DataFrame
在Spark SQL中SparkSession是创建DataFrame和执行SQL的入口，创建DataFrame有三种方式：通过Spark的数据源进行创建；从一个存在的RDD进行转换；还可以从Hive Table进行查询返回。
1)从Spark数据源进行创建
查看Spark支持创建文件的数据源格式
spark.read. +tab键
2.在spark的bin/data目录（windows环境）中创建user.json文件
{"username":"zhangsan","age":20}
3.读取json文件创建DataFrame
scala> val df = spark.read.json("data/user.json")
df: org.apache.spark.sql.DataFrame = [age: bigint， username: string]
注意：如果从内存中获取数据，spark可以知道数据类型具体是什么。如果是数字，默认作为Int处理；但是从文件中读取的数字，不能确定是什么类型，所以用bigint接收，可以和Long类型转换，但是和Int不能进行转换
SQL语法
SQL语法风格是指我们查询数据的时候使用SQL语句来查询，这种风格的查询必须要有临时视图或者全局视图来辅助
1)读取JSON文件创建DataFrame
scala> val df = spark.read.json("data/user.json") df: org.apache.spark.sql.DataFrame = [age: bigint， username: 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|username| +---+--------+ | 20|zhangsan| | 30| lisi| | 40| wangwu| +---+--------+
注意：普通临时表是Session范围内的，如果想应用范围内有效，可以使用全局临时表。使用全局临时表时需要全路径访问，如：global_temp.people
5)对于DataFrame创建一个全局表
scala> df.createGlobalTempView("people")
6)通过SQL语句实现查询全表
scala> spark.sql("SELECT * FROM global_temp.people").show() +---+--------+ |age|username| +---+--------+ | 20|zhangsan| | 30| lisi| | 40| wangwu| +---+--------+
scala> spark.newSession().sql("SELECT * FROM global_temp.people").show() +---+--------+ |age|username| +---+--------+ | 20|zhangsan| | 30| lisi| | 40| wangwu| +---+--------+
DSL语法
DataFrame提供一个特定领域语言(domain-specific language, DSL)去管理结构化的数据。可以在 Scala, Java, Python 和 R 中使用 DSL，使用 DSL 语法风格不必去创建临时视图了
1)创建一个DataFrame
scala> val df = spark.read.json("data/user.json") df: org.apache.spark.sql.DataFrame = [age: bigint， name: string]
2)查看DataFrame的Schema信息
scala> df.printSchema root |-- age: Long (nullable = true) |-- username: string (nullable = true)
3)只查看"username"列数据，
scala> df.select("username").show() +--------+ |username| +--------+ |zhangsan| | lisi| | wangwu| +--------+
4)查看"username"列数据以及"age+1"数据
注意:涉及到运算的时候, 每列都必须使用$, 或者采用引号表达式：单引号+字段名
scala> df.select($"username",$"age" + 1).show scala> df.select('username, 'age + 1).show() scala> df.select('username, 'age + 1 as "newage").show() +--------+---------+ |username|(age + 1)| +--------+---------+ |zhangsan| 21| | lisi| 31| | wangwu| 41| +--------+---------+
5)查看"age"大于"30"的数据
scala> df.filter($"age">30).show +---+---------+ |age| username| +---+---------+ | 40| wangwu| +---+---------+
6)按照"age"分组，查看数据条数
scala> df.groupBy("age").count.show +---+-----+ |age|count| +---+-----+ | 20| 1| | 30| 1| | 40| 1| +---+-----+
RDD转换为DataFrame
在IDEA中开发程序时，如果需要RDD与DF或者DS之间互相操作，那么需要引入 import spark.implicits._
这里的spark不是Scala中的包名，而是创建的sparkSession对象的变量名称，所以必须先创建SparkSession对象再导入。这里的spark对象不能使用var声明，因为Scala只支持val修饰的对象的引入。
spark-shell中无需导入，自动完成此操作。
scala> val idRDD = sc.textFile("data/id.txt") scala> idRDD.toDF("id").show +---+ | id| +---+ | 1| | 2| | 3| | 4| +---+
实际开发中，一般通过样例类将RDD转换为DataFrame
scala> case class User(name:String, age:Int) defined class User scala> sc.makeRDD(List(("zhangsan",30), ("lisi",40))).map(t=>User(t._1, t._2)).toDF.show +--------+---+ | name|age| +--------+---+ |zhangsan| 30| | lisi| 40| +--------+---+
DataFrame转换为RDD
DataFrame其实就是对RDD的封装，所以可以直接获取内部的RDD
scala> val df = sc.makeRDD(List(("zhangsan",30), ("lisi",40))).map(t=>User(t._1, t._2)).toDF df: org.apache.spark.sql.DataFrame = [name: string, age: int] scala> val rdd = df.rdd rdd: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[46] at rdd at <console>:25 scala> val array = rdd.collect array: Array[org.apache.spark.sql.Row] = Array([zhangsan,30], [lisi,40])
注意：此时得到的RDD存储类型为Row
scala> array(0) res28: org.apache.spark.sql.Row = [zhangsan,30] scala> array(0)(0) res29: Any = zhangsan scala> array(0).getAs[String]("name") res30: String = zhangsan
DataSet
DataSet是具有强类型的数据集合，需要提供对应的类型信息。
scala> case class Person(name: String, age: Long) defined class Person scala> val caseClassDS = Seq(Person("zhangsan",2)).toDS() caseClassDS: org.apache.spark.sql.Dataset[Person] = [name: string, age: Long] scala> caseClassDS.show +---------+---+ | name|age| +---------+---+ | zhangsan| 2| +---------+---+
2）使用基本类型的序列创建DataSet
scala> val ds = Seq(1,2,3,4,5).toDS ds: org.apache.spark.sql.Dataset[Int] = [value: int] scala> ds.show +-----+ |value| +-----+ | 1| | 2| | 3| | 4| | 5| +-----+
注意：在实际使用的时候，很少用到把序列转换成DataSet，更多的是通过RDD来得到DataSet
