一. 概述
Spark SQL 是Spark 用于结构化数据(structured data)处理的 Spark 模块。
1.1 Hive and SparkSQL
SparkSQL 的前身是 Shark,而 Shark 是伯克利实验室 Spark 生态环境的组件之一,是基于Hive 所开发的工具;
但是,随着Spark 的发展,对于野心勃勃的Spark 团队来说,Shark 对于 Hive 的太多依赖(如采用 Hive 的 语法解析器、查询优化器等等),制约了 Spark 的One Stack Rule Them All 的既定方针,制约了 Spark 各个组件的相互集成,所以提出了 SparkSQL 项目。SparkSQL 抛弃原有 Shark 的代码,汲取了 Shark 的一些优点,如内存列存储(In-Memory ColumnarStorage)、Hive 兼容性等,重新开发了SparkSQL 代码;由于摆脱了对Hive 的依赖性,SparkSQL无论在数据兼容、性能优化、组件扩展方面都得到了极大的方便,真可谓“退一步,海阔天空”。
SparkSQL 作为 Spark 生态的一员继续发展,而不再受限于 Hive,只是兼容 Hive; Hive on Spark 是一个Hive 的发展计划,该计划将 Spark 作为Hive 的底层引擎之一,也就是说,Hive 将不再受限于一个引擎,可以采用 Map-Reduce、Tez、Spark 等引擎。 对于开发人员来讲,SparkSQL 可以简化RDD 的开发,提高开发效率,且执行效率非常快,所以实际工作中,基本上采用的就是 SparkSQL。
hive: 将SQL转换为MapReduce来执行; Spark: SparkSQL是为了简化RDD的开发,把RDD做了一套封装, 写SparkSQL时会自动把封装好的模型转化为RDD ! 而封装的模型就是DateFrame和DateSet ;
1.2 SparkSQL 特点
易整合 无缝的整合了 SQL 查询和 Spark 编程
统一的数据访问 使用相同的方式连接不同的数据源(hbase , hive, Mysql)
兼容 Hive 在已有的仓库上直接运行 SQL 或者 HiveQL (SparkSQL前身就是与Hive结合的Shark)
标准数据连接 通过 JDBC 或者 ODBC 来连接
1.3 DataFrame 是什么 ?
RDD只关心数据,DataFrame更关心元数据的结构信息,类似数据库的表;
DataFrame 是一种以 RDD 为基础的分布式数据集,类似于传统数据库中的二维表格。
DataFrame 与 RDD 的主要区别在于: DataFrame 带有 schema 元信息,即 DataFrame 所表示的二维表数据集的每一列都带有名称和类型。而RDD只关心数据而不关心数据的结构。
同时,与Hive 类似,DataFrame 也支持嵌套数据类型(struct、array 和 map)。从 API 易用性的角度上看,DataFrame API 提供的是一套高层的关系操作,比函数式的 RDD API 要更加友好,门槛更低。 如图,RDD中只有一个个Person 泛型。数据都是Person的对象,不关心数据的属性。 DataFrame像二维表格, 会保存数据的元数据信息;
左侧的 RDD[Person]虽然以 Person 为类型参数,但 Spark 框架本身不了解Person 类的内部结构。 而右侧的DataFrame 却提供了详细的结构信息,使得 Spark SQL 可以清楚地知道该数据集中包含哪些列,每列的名称和类型各是什么。 DataFrame 是为数据提供了 Schema 的视图。可以把它当做数据库中的一张表来对待;
1.4 DataSet 是什么 ?
DataSet 把一行数据当成一个对象来用;
DataSet 是分布式数据集合。DataSet 是Spark 1.6 中添加的一个新抽象,是DataFrame的一个扩展; 它提供了RDD 的优势(强类型,使用强大的 lambda 函数的能力)以及SparkSQL 优化执行引擎的优点。DataSet 也可以使用功能性的转换(操作 map,flatMap,filter等等);
DataSet 是DataFrame API 的一个扩展,是SparkSQL 最新的数据抽象;
DataSet 是强类型的。比如可以有 DataSet[Car],DataSet[Person]
DataFrame 是DataSet 的特列,DataFrame=DataSet[Row] ,所以可以通过 as 方法将DataFrame 转换DataSet。 Row 是一个类型,跟 Car、Person 这些的类型一样,所有的表结构信息都用 Row 来表示。获取数据时需要指定顺序;
二. 核心编程
2.1 环境
Spark Core 中,如果想要执行应用程序,需要首先构建上下文环境对象 SparkContext (sc); 而Spark SQL 其实可以理解为对 Spark Core 的一种封装,不仅仅在模型上进行了封装,上下文环境对象也进行了封装。
SparkSession 是 Spark 最新的 SQL 查询起始点,实质上是 SQLContext 和HiveContext 的组合,所以在 SQLContex 和HiveContext 上可用的API 在 SparkSession 上同样是可以使用的。 SparkSession 内部封装了 SparkContext,所以计算实际上是由 sparkContext 完成的;
sc即选择Spark Context为环境对象 spark即选择Spark Session为环境对象
2.2 DataFrame
Spark SQL 的DataFrame API 允许我们使用 DataFrame 而不用必须去注册临时表或者生成 SQL 表达式。 DataFrame API 既有 transformation 操作也有 action 操作。
2.2.1 创建 DataFrame
在 Spark SQL 中, SparkSession 是创建DataFrame 和执行 SQL 的入口, 创建 DataFrame有三种方式: ①通过Spark 的数据源进行创建; (Spark.read等) ②从一个存在的RDD 进行转换;(将RDD转换为SparkSQL,然后创建DateFrame) ③还可以从HiveTable 进行查询返回。
-
例:读取json文件为DataFrame (Json格式: 字段 冒号 值) 读取json文件后,DataFrame会自动识别出数据的结构 从文件中读取,只能识别为bigint ; 查看df信息: df.show 需求:查看DataFrame df的age信息: scala> df.createTempView(“user”) scala> spark.sql(“select age from user”).show
2.2.2 创建视图
createTempView :创建临时视图,重复的不可创建;
createOrReplaceTempView :创建临时视图,重复替换;
createGlobalTempView :创建全局临时视图,重复的不可创建;
createOrReplaceGlobalTempView :创建全局视图,重复的替换;
注意:普通临时视图是 【Session会话范围内】的(更改会话、结束会话时,临时视图就不在了 !),如果想应用范围内有效,可以使用全局临时视图;
注意: 访问全局表时,要加global_temp. 前置路径访问!( 类似表空间)
scala> spark.sql("SELECT * FROM global_temp.ods1").show()
view视图和table表的区别? view视图是查询结果集(一个快照),view只可以查不可以改! (数据源在DateFrame里面) 而table可以增删改查!
2.2.3 RDD 转换为 DataFrame
scala> val rdd=sc.makeRDD(List(1,2,3,4))
scala> val df=rdd.toDF ("id")
2.2.4 DataFrame 转换为 RDD
DataFrame 其实就是对RDD 的封装,所以可以直接获取内部的RDD;
scala> df.rdd
注意:此时得到的RDD 存储类型为Row
2.3 DataSet
DataSet 是具有强类型的数据集合,需要提供对应的类型信息。 DateFrame只有数据本身的类型;
为什么有强类型的概念? 因为使用强类型访问数据会更方便,类似于一个实体类,比如User 类型可以访问username,age等属性,不用考虑顺序;
2.3.1 创建 DataSet
1)使用样例类序列创建 DataSet
scala> case class Person(name: String, age: Long)
scala> val list=List(Person("zhangsan",30),Person("lisi",40))
scala> val ds=list.toDS
注意:在实际使用的时候,很少用到把序列转换成DataSet,更多的是通过RDD 来得到DataSet;
2.3.2 RDD 转换为 DataSet
case class User(name: String, age: Long)
val rdd=sc.makeRDD (List(User ("zhangsan",30), User("lisi",40) ) )
rdd.toDS
2.3.3 DataSet 转换为 RDD
DataSet 其实也是对 RDD 的封装,所以可以直接获取内部的RDD
val rdd1=ds.rdd
2.4 DataFrame 和 DataSet 转换
DataFrame 其实是DataSet 的特例,所以它们之间是可以互相转换的。
2.4.1 DataFrame 转换为DataSet
先建立 DataFrame:
val rdd = sc.makeRDD( List(("zhangsan",30),("lisi",49) ) ) /创建RDD
/ 从内存读取时,可以自动识别数据的类型,此处识别为Int,如果是文件读入则识别为bigint
val df =rdd.toDF("name","age") /将RDD转换为DataFram
转换为 DataSet: df.as[User]
case class User(name:String, age:Int) / 使用样例类创建User类 ,User类的属性要和DataFram的列对应上!
val ds = df.as[User] / 将DataFrame 转换为User类型的DataSet
2.4.2 DataSet 转换为DataFrame
DataSet本身就包含了结构信息和数据,即只需要去掉类型,保留结构就是DataFrame了!
ds.toDF
2.5 RDD、DataFrame、DataSet 三者的关系
共性:
-
RDD、DataFrame、DataSet 全都是 spark 平台下的分布式、弹性 数据集,为处理超大型数据提供便利; 三者都有惰性机制,在进行创建、转换,如 map 方法时,不会立即执行,只有在遇到Action 如 foreach 时,三者才会开始遍历运算; -
三者有许多共同的函数,如 filter,排序等; //因为底层都是RDD -
三者都有 partition 分区的概念 -
三者都会根据 Spark 的内存情况自动缓存运算,这样即使数据量很大,也不用担心会内存溢出;
区别:
-
RDD (底层核心内容) RDD 一般和 spark mllib 同时使用 RDD 不支持 sparksql 操作 -
DataFrame 与 RDD 和 Dataset 不同,DataFrame 每一行的类型固定为Row(行对象),每一列的值没法直接访问,只有通过解析才能获取各个字段的值 DataFrame 与DataSet 一般不与 spark mllib 同时使用 DataFrame 与DataSet 均支持 SparkSQL 的操作 DataFrame 与DataSet 支持一些特别方便的保存方式,比如保存成 csv,可以带上表头,这样每一列的字段名一目了然 -
DataSet Dataset 和DataFrame 拥有完全相同的成员函数,区别只是每一行的数据类型不同。 DataFrame 其实就是DataSet 的一个特例 type DataFrame = Dataset[Row] DataFrame 也可以叫Dataset[Row],每一行的类型是 Row,不解析,每一行究竟有哪些字段,各个字段又是什么类型都无从得知,只能用上面提到的 getAS 方法或者共性中的第七条提到的模式匹配拿出特定字段。而Dataset 中,每一行是什么类型是不一定的,在自定义了 case class 之后可以很自由的获得每一行的信息
转换图:
|