Spark SQL 学习总结

（一）Spark SQL

Spark SQL和我们之前讲Hive的时候说的hive on spark是不一样的。
hive on spark是表示把底层的mapreduce引擎替换为spark引擎。
而Spark SQL是Spark自己实现的一套SQL处理引擎。
Spark SQL是Spark中的一个模块，主要用于进行结构化数据的处理。它提供的最核心的编程抽象，就是DataFrame。
DataFrame=RDD+Schema 。
它其实和关系型数据库中的表非常类似，RDD可以认为是表中的数据，Schema是表结构信息。
DataFrame可以通过很多来源进行构建，包括：结构化的数据文件，Hive中的表，外部的关系型数据库，以及RDD
Spark1.3出现的 DataFrame ，Spark1.6出现了 DataSet ，在Spark2.0中两者统一，DataFrame等于DataSet[Row]

（二）SParkSession

要使用Spark SQL，首先需要创建一个SpakSession对象
SparkSession中包含了SparkContext和SqlContext
所以说想通过SparkSession来操作RDD的话需要先通过它来获取SparkContext
这个SqlContext是使用sparkSQL操作hive的时候会用到的。

使用案例
添加依赖：

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.4.3</version>
</dependency>

object SqlDemoScala {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local")val sparkSession=SparkSession.builder().appName("sqlDemoScala").config(conf).getOrCreate()//读取json文件，获取Dataframeval stuDf = sparkSession.read.json("D:\\student.json")//查看Dataframe中的数据stuDf.show()sparkSession.stop()}}

结果：

+---+------+------+
|age|  name|   sex|
+---+------+------+
| 19|  jack|  male|
| 18|   tom|  male|
| 27|jessic|female|
| 18|  hehe|female|
| 15|  haha|  male|
+---+------+------+

（三）DataFrame常见算子操作

• printSchema()
• show()
• select()
• filter()、where()
• groupBy()
• count()

使用案例

object DataFrameOpScala {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setMaster("local")
//创建SparkSession对象，里面包含SparkContext和SqlContext
val sparkSession = SparkSession.builder()
.appName("DataFrameOpScala")
.config(conf)
.getOrCreate()val stuDf = sparkSession.read.json("D:\\student.json")
//打印schema信息
stuDf.printSchema()
//默认显示所有数据，可以通过参数控制显示多少条
stuDf.show(2)
//查询数据中的指定字段信息
stuDf.select("name","age").show()
//在使用select的时候可以对数据做一些操作，需要添加隐式转换函数，否则语法报错
import sparkSession.implicits._
stuDf.select($"name",$"age" + 1).show()
//对数据进行过滤，需要添加隐式转换函数，否则语法报错
stuDf.filter($"age">18).show()
//where底层调用的就是filter
stuDf.where($"age">18).show()
//对数据进行分组求和
stuDf.groupBy("age").count().show()
sparkSession.stop()
}
}

（四）DataFrame的sql操作

想要实现直接支持sql语句查询DataFrame中的数据
需要两步操作

1. 先将DataFrame注册为一个临时表
2. 使用sparkSession中的sql函数执行sql语句

案例实现：

object DataFrameSqlScala {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setMaster("local")
//创建SparkSession对象，里面包含SparkContext和SqlContext
val sparkSession = SparkSession.builder()
.appName("DataFrameSqlScala")
.config(conf)
.getOrCreate()
val stuDf = sparkSession.read.json("D:\\student.json")
//将DataFrame注册为一个临时表
stuDf.createOrReplaceTempView("student")
//使用sql查询临时表中的数据
sparkSession.sql("select age,count(*) as num from student group by age")
.show()
sparkSession.stop()
}
}

（五）RDD转换为DataFrame

为什么要将RDD转换为DataFrame?
在实际工作中我们可能会先把hdfs上的一些日志数据加载进来，然后进行一些处理，最终变成结构化的数据，希望对这些数据做一些统计分析，当然了我们可以使用spark中提供transformation算子来实现，只不过会有一些麻烦，毕竟是需要写代码的，如果能够使用sql实现，其实是更加方便的。所以可以针对我们前面创建的RDD，将它转换为DataFrame，这样就可以使用dataFrame中的一些算子或者直接写sql来操作数据了。
Spark SQL支持这两种方式将RDD转换为DataFrame

1. 反射方式
2. 编程方式

（1）反射方式

这种方式是使用反射来推断RDD中的元数据。
基于反射的方式，代码比较简洁，也就是说当你在写代码的时候，已经知道了RDD中的元数据，这样的话使用反射这种方式是一种非常不错的选择。

Scala具有隐式转换的特性，所以spark sql的scala接口是支持自动将包含了case class的RDD转换为DataFrame的
下面来举一个例子：

object RddToDataFrameByReflectScala {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setMaster("local")
//创建SparkSession对象，里面包含SparkContext和SqlContext
val sparkSession = SparkSession.builder()
.appName("RddToDataFrameByReflectScala")
.config(conf)
.getOrCreate()
//获取SparkContext
val sc = sparkSession.sparkContext
val dataRDD = sc.parallelize(Array(("jack",18),("tom",20),("jessic",30)))
//基于反射直接将包含Student对象的dataRDD转换为DataFrame
//需要导入隐式转换
import sparkSession.implicits._
val stuDf = dataRDD.map(tup=>Student(tup._1,tup._2)).toDF()
//下面就可以通过DataFrame的方式操作dataRDD中的数据了
stuDf.createOrReplaceTempView("student")
//执行sql查询
val resDf = sparkSession.sql("select name,age from student where age > 18
//将DataFrame转化为RDD
val resRDD = resDf.rdd
//从row中取数据，封装成student，打印到控制台
resRDD.map(row=>Student(row(0).toString,row(1).toString.toInt))
.collect()
.foreach(println(_))
//使用row的getAs()方法，获取指定列名的值
resRDD.map(row=>Student(row.getAs[String]("name"),row.getAs[Int]("age")))
.collect()
.foreach(println(_))
sparkSession.stop()
}
}
//定义一个Student
case class Student(name: String,age: Int)

（2）编程方式

接下来是编程的方式
这种方式是通过编程接口来创建DataFrame，你可以在程序运行时动态构建一份元数据，就是Schema，然后将其应用到已经存在的RDD上。这种方式的代码比较冗长，但是如果在编写程序时，还不知道RDD的元数据，只有在程序运行时，才能动态得知其元数据，那么只能通过这种动态构建元数据的方式。
也就是说当case calss中的字段无法预先定义的时候，就只能用编程方式动态指定元数据了

案例：

object RddToDataFrameByProgramScala {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setMaster("local")
//创建SparkSession对象，里面包含SparkContext和SqlContext
val sparkSession = SparkSession.builder()
.appName("RddToDataFrameByProgramScala")
.config(conf)
.getOrCreate()
//获取SparkContext
val sc = sparkSession.sparkContext
val dataRDD = sc.parallelize(Array(("jack",18),("tom",20),("jessic",30)))
//组装rowRDD
val rowRDD = dataRDD.map(tup=>Row(tup._1,tup._2))
//指定元数据信息【这个元数据信息就可以动态从外部获取了，比较灵活】
val schema = StructType(Array(
StructField("name",StringType,true),
StructField("age",IntegerType,true)
))
//组装DataFrame
val stuDf = sparkSession.createDataFrame(rowRDD,schema)
//下面就可以通过DataFrame的方式操作dataRDD中的数据了
stuDf.createOrReplaceTempView("student")
//执行sql查询
val resDf = sparkSession.sql("select name,age from student where age > 18
//将DataFrame转化为RDD
val resRDD = resDf.rdd
resRDD.map(row=>(row(0).toString,row(1).toString.toInt))
.collect()
.foreach(println(_))
sparkSession.stop()
}
}

（六）load和save操作

对于Spark SQL的DataFrame来说，无论是从什么数据源创建出来的DataFrame，都有一些共同的load和save操作。
load操作主要用于加载数据，创建出DataFrame；
save操作，主要用于将DataFrame中的数据保存到文件中。

（1）load操作

我们前面操作json格式的数据的时候好像没有使用load方法，而是直接使用的json方法，这是什么特殊用法吗？
查看json方法的源码会发现，它底层调用的是format和load方法
注意：如果看不到源码，需要点击idea右上角的download source提示信息下载依赖的源码。

此时如果不指定format，则默认读取的数据源格式是parquet，也可以手动指定数据源格式。Spark SQL
内置了一些常见的数据源类型，比如json, parquet, jdbc, orc, csv, text
通过这个功能，就可以在不同类型的数据源之间进行转换了。

案例：

object LoadAndSaveOpScala {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setMaster("local")
//创建SparkSession对象，里面包含SparkContext和SqlContext
val sparkSession = SparkSession.builder()
.appName("LoadAndSaveOpScala")
.config(conf)
.getOrCreate()
//读取数据
val stuDf = sparkSession.read
.format("json")
.load("D:\\student.json")
//保存数据
stuDf.select("name","age")
.write
.format("csv")
.save("hdfs://bigdata01:9000/out-save001")
sparkSession.stop()
}
}

执行代码，查看结果，csv文件是使用逗号分隔的：

jack,19
tom,18
jessic,27
hehe,18
haha,15

（2）SaveMode

Spark SQL对于save操作，提供了不同的save mode。
主要用来处理，当目标位置已经有数据时应该如何处理。save操作不会执行锁操作，并且也不是原子的，因此是有一定风险出现脏数据的。

SaveMode 解释
SaveMode.ErrorIfExists (默认) 如果目标位置已经存在数据，那么抛出一个异常
SaveMode.Append 如果目标位置已经存在数据，那么将数据追加进去
SaveMode.Overwrite 如果目标位置已经存在数据，那么就将已经存在的数据删
SaveMode.Ignore 如果目标位置已经存在数据，那么就忽略，不做任何操作

在LoadAndSaveOpScala中增加SaveMode的设置，重新执行，验证结果将SaveMode设置为Append，如果目标已存在，则追加

（七）内置函数

Spark提供了很多内置函数

种类 函数
聚合函数 avg, count, countDistinct, first, last, max, mean, min, sum,
集合函数 array_contains, explode, size
日期/时间函数 datediff, date_add, date_sub, add_months, last_day, next_day,
数学函数 abs, ceil, floor, round
混合函数 if, isnull, md5, not, rand, when
字符串函数 concat, get_json_object, length, reverse, split, upper
窗口函数 denseRank, rank, rowNumber

其实这里面的函数和hive中的函数是类似的
注意：SparkSQL中的SQL函数文档不全，其实在使用这些函数的时候，大家完全可以去查看hive中sql的文档，使用的时候都是一样的。