sparkSQL基础
目录
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:弹性分布式数据集;不可变、可分区、元素可以并行计算的集合。
优点:
- RDD编译时类型安全:编译时能检查出类型错误;
- 面向对象的编程风格:直接通过类名点的方式操作数据。
缺点:
- 序列化和反序列化的性能开销很大,大量的网络传输;
- 构建对象占用了大量的heap堆内存,导致频繁的GC(程序进行GC时,所有任务都是暂停)
RDD的数据结构为:
(2)DataFrame
DataFrame以RDD为基础的分布式数据集。
优点:
- DataFrame带有元数据schema,每一列都带有名称和类型。
- DataFrame引入了off-heap,构建对象直接使用操作系统的内存,不会导致频繁GC。
- DataFrame可以从很多数据源构建;
- DataFrame把内部元素看成Row对象,表示一行行的数据。
- DataFrame=RDD+schema
缺点:
- 编译时类型不安全;
- 不具有面向对象编程的风格。
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)