目录

1.sparksql概述

1.1、什么是Spark SQL

1.2、SparkSQL的数据源

1.3、SparkSQL底层架构

2. sparksql的四大特性

2.1、易整合

2.2、统一的数据源访问

2.3、兼容hive

2.4、支持标准的数据库连接

3. DataFrame概述

3.1、DataFrame是什么

3.2、DataFrame和RDD的优缺点

3.4、DataFrame常用操作

4. DataSet概述

4.1、DataSet是什么

4.2、RDD、DataFrame、DataSet的区别

4.3、构建DataSet（四种方法）

---

1.sparksql概述

1.1、什么是Spark SQL

• Spark SQL is Apache Spark's module for working with structured data.
• SparkSQL是apache Spark用来处理结构化数据的一个模块
• SparkSQL支持查询原生的RDD。 RDD是Spark平台的核心概念，是Spark能够高效的处理大数据的各种场景的基础。
• 能够在Scala中写SQL语句。支持简单的SQL语法检查，能够在Scala中写Hive语句访问Hive数据，并将结果取回作为RDD使用。

1.2、SparkSQL的数据源

SparkSQL的数据源可以是JSON类型的字符串，JDBC,Parquent,Hive，HDFS等。

1.3、SparkSQL底层架构

（1）首先拿到sql后解析一批未被解决的逻辑计划；

（2）再经过分析得到分析后的逻辑计划；

（3）再经过一批优化规则转换成一批最佳优化的逻辑计划；

（4）再经过SparkPlanner的策略转化成一批物理计划；

（5）最后经过消费模型转换成一个个的Spark任务执行。

2. sparksql的四大特性

2.1、易整合

• 将SQL查询与Spark程序无缝混合；
• 可以使用不同的语言进行代码开发：
  • java
  • scala
  • python

2.2、统一的数据源访问

• 以相同的方式连接到任何数据源；
• sparksql后期可以采用一种统一的方式去对接任意的外部数据源：

                   dataFrame = sparkSession.read.文件格式的方法名("该文件格式的路径")

2.3、兼容hive

• sparksql可以支持hivesql

2.4、支持标准的数据库连接

• sparksql支持标准的数据库连接JDBC或者ODBC

3. DataFrame概述

3.1、DataFrame是什么

• DataFrame是一种以RDD为基础的分布式数据集，类似于传统数据库的二维表格；
• DataFrame带有Schema元信息，即DataFrame所表示的二维表数据集的每一列都带有名称和类型，但底层做了更多的优化；
• DataFrame可以从很多数据源构建：
  • 比如：已经存在的RDD、结构化文件、外部数据库、Hive表。
• RDD可以把它内部元素看成是一个java对象；
• DataFrame可以把内部是一个Row对象，它表示一行一行的数据；
• 可以把DataFrame这样去理解
  • RDD+schema元信息
• dataFrame相比于rdd来说，多了对数据的描述信息（schema元信息）

3.2、DataFrame和RDD的优缺点

（1）RDD

 

优点：

           1、编译时类型安全：开发会进行类型检查，在编译的时候及时发现错误；

           2、具有面向对象编程的风格。

缺点：

           1、构建大量的java对象占用了大量heap堆空间，导致频繁的GC：

                 RDD它的数据量比较大，后期都需要存储在heap堆中，而heap堆中的内存空间有限，出现频繁的垃圾回收（GC），程序在进行垃圾回收的过程中，所有的任务都是暂停，影响程序执行的效率。

           2、数据的序列化和反序列性能开销很大：

                  在分布式程序中，对象(对象的内容和结构)是先进行序列化，发送到其他服务器，进行大量的网络传输，然后接受到这些序列化的数据之后，再进行反序列化来恢复该对象。

（2）DataFrame

• DataFrame引入了schema元信息和off-heap(堆外空间)

优点：

           1、DataFrame引入off-heap，大量的对象构建直接使用操作系统层面上的内存，不在使用heap堆中的内存，这样一来heap堆中的内存空间就比较充足，不会导致频繁GC，程序的运行效率比较高，它是解决了RDD构建大量的java对象占用了大量heap堆空间，导致频繁的GC这个缺点。

           2、DataFrame引入了schema元信息---就是数据结构的描述信息，后期spark程序中的大量对象在进行网络传输的时候，只需要把数据的内容本身进行序列化就可以，数据结构信息可以省略掉。这样一来数据网络传输的数据量是有所减少，数据的序列化和反序列性能开销就不是很大了。它是解决了RDD数据的序列化和反序列性能开销很大这个缺点。

缺点：

• DataFrame引入了schema元信息和off-heap(堆外)它是分别解决了RDD的缺点，同时它也丢失了RDD的优点。

           1、编译时类型不安全

                 编译时不会进行类型的检查，这里也就意味着前期是无法在编译的时候发现错误，只有在运行的时候才会发现

           2、不在具有面向对象编程的风格

3.3、 读取文件构建DataFrame

（1） 读取文本文件创建DataFrame

第一种方式：读取text文件

//创建dataFrame
val personDF=spark.read.text("/person.txt")
//打印schema信息
personDF.printSchema
//展示数据
personDF.show

第二种方式：调用toDF方法将rdd转换成dataFrame

//加载数据
val rdd1=sc.textFile("/person.txt").map(x=>x.split(" "))
//定义一个样例类
case class Person(id:String,name:String,age:Int)
//把rdd与样例类进行关联
val personRDD=rdd1.map(x=>Person(x(0),x(1),x(2).toInt))
//把rdd转换成DataFrame
val personDF=personRDD.toDF

//打印schema信息
personDF.printSchema
//展示数据
personDF.show

（2）读取json文件创建DataFrame

val peopleDF=spark.read.json("/people.json")  //读取json文件创建dataFrame
 
//打印schema信息
peopleDF.printSchema
//展示数据
peopleDF.show

（3）读取parquet文件创建DataFrame

//创建DataFrame
val usersDF=spark.read.parquet("/users.parquet")

//打印schema信息
usersDF.printSchema
//展示数据
usersDF.show

（4）读取JDBC中的数据创建DataFrame(MySql为例)

val mysqlDF:DataFrame = sparkSession.read.jdbc(url,tableName,properties)

（5）读取Hive中的数据加载成DataFrame

 val dataFrame = sparkSession.sql("select * from people")

（6） 直接创建

 val dataFrame:DataFrame = sparkSession.createDataFrame(rowRDD,schema)

3.4、DataFrame常用操作

（1）DSL风格语法

sparksql中的DataFrame自身提供了一套自己的Api，可以去使用这套api来做相应的处理。

//加载数据
val rdd1=sc.textFile("/person.txt").map(x=>x.split(" "))
//定义一个样例类
case class Person(id:String,name:String,age:Int)
//把rdd与样例类进行关联
val personRDD=rdd1.map(x=>Person(x(0),x(1),x(2).toInt))
//把rdd转换成DataFrame
val personDF=personRDD.toDF

//打印schema信息
personDF.printSchema
//展示数据
personDF.show

//查询指定的字段
personDF.select("name").show
personDF.select($"name").show
personDF.select(col("name")).show

//实现age+1
 personDF.select($"name",$"age",$"age"+1).show

//实现age大于30过滤
 personDF.filter($"age" > 30).show

//按照age分组统计次数
 personDF.groupBy("age").count.show 

//按照age分组统计次数降序
 personDF.groupBy("age").count().sort($"count".desc).show

（2）SQL风格语法

• 可以把DataFrame注册成一张表，然后通过sparkSession.sql(sql语句)操作。

//DataFrame注册成表
personDF.createTempView("person")

//使用SparkSession调用sql方法统计查询
spark.sql("select * from person").show
spark.sql("select name from person").show
spark.sql("select name,age from person").show
spark.sql("select * from person where age >30").show
spark.sql("select count(*) from person where age >30").show
spark.sql("select age,count(*) from person group by age").show
spark.sql("select age,count(*) as count from person group by age").show
spark.sql("select * from person order by age desc").show

4. DataSet概述

4.1、DataSet是什么

DataSet是分布式的数据集合，Dataset提供了强类型支持，也是在RDD的每行数据加了类型约束。

DataSet是在Spark1.6中添加的新的接口。它集中了RDD的优点（强类型和可以用强大lambda函数）以及使用了Spark SQL优化的执行引擎。

4.2、RDD、DataFrame、DataSet的区别

（1）RDD

RDD：弹性分布式数据集；不可变、可分区、元素可以并行计算的集合。

优点：

1. RDD编译时类型安全：编译时能检查出类型错误；
2. 面向对象的编程风格：直接通过类名点的方式操作数据。

缺点：

1. 序列化和反序列化的性能开销很大，大量的网络传输；
2. 构建对象占用了大量的heap堆内存，导致频繁的GC（程序进行GC时，所有任务都是暂停）

RDD的数据结构为：

（2）DataFrame

DataFrame以RDD为基础的分布式数据集。

优点：

1. DataFrame带有元数据schema，每一列都带有名称和类型。
2. DataFrame引入了off-heap，构建对象直接使用操作系统的内存，不会导致频繁GC。
3. DataFrame可以从很多数据源构建；
4. DataFrame把内部元素看成Row对象，表示一行行的数据。
5. DataFrame=RDD+schema

缺点：

1. 编译时类型不安全；
2. 不具有面向对象编程的风格。

DataFrame的数据结构为：（类似于二维表）

（3）Dataset

DataSet包含了DataFrame的功能，Spark2.0中两者统一，DataFrame表示为DataSet[Row]，即DataSet的子集。
（1）DataSet可以在编译时检查类型；
（2）并且是面向对象的编程接口。

（DataSet 结合了 RDD 和 DataFrame 的优点，并带来的一个新的概念 Encoder。当序列化数据时，Encoder 产生字节码与 off-heap 进行交互，能够达到按需访问数据的效果，而不用反序列化整个对象。）

Dataset中的数据结构：

或者

（4）三者之间的转换：

（1）RDD转换成DataFrame或DataSet，需先定义一个样例类，再将RDD与样例类进行关联，再调用.toDF方法或.toDS方法。

            //定义一个样例类
           case class Person(id:String,name:String,age:Int)
          //把rdd与样例类进行关联
           val personRDD=rdd1.map(x=>Person(x(0),x(1),x(2).toInt))
           //把rdd转换成DataFrame

            //需要手动导入隐式转换
           import spark.implicits._
           val personDF=personRDD.toDF

（2）而DataFrame或DataSet转换成RDD，只需调用.rdd方法即可。

          val rdd1=dataFrame.rdd

           val rdd2=dataSet.rdd

（3）DataFrame转换成DataSet：val dataSet=dataFrame.as[强类型]

（4）DataSet转换成DataFrame：val dataFrame=dataSet.toDF

4.3、构建DataSet（四种方法）

（1）通过sparkSession调用createDataset方法

 ds=spark.createDataset(1 to 10) //scala集合
 val ds=spark.createDataset(sc.textFile("/person.txt"))  //rdd

（2）使用scala集合和rdd调用toDS方法

textFile("/person.txt").toDS
List(1,2,3,4,5).toDS

（3）把一个DataFrame转换成DataSet

dataSet=dataFrame.as[强类型]

（4）通过一个DataSet转换生成一个新的DataSet

List(1,2,3,4,5).toDS.map(x=>x*10)