Spark SQL 是Spark 用于结构化数据(structured data)处理的 Spark 模块。
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 ;
易整合
无缝的整合了 SQL 查询和 Spark 编程
统一的数据访问
使用相同的方式连接不同的数据源(hbase , hive, Mysql)
兼容 Hive
在已有的仓库上直接运行 SQL 或者 HiveQL (SparkSQL前身就是与Hive结合的Shark)
标准数据连接
通过 JDBC 或者 ODBC 来连接
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 的视图。可以把它当做数据库中的一张表来对待;
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 来表示。获取数据时需要指定顺序;
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为环境对象
Spark SQL 的DataFrame API 允许我们使用 DataFrame 而不用必须去注册临时表或者生成 SQL 表达式。
DataFrame API 既有 transformation 操作也有 action 操作。
在 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
createTempView :创建临时视图,重复的不可创建;
createOrReplaceTempView :创建临时视图,重复替换;
createGlobalTempView :创建全局临时视图,重复的不可创建;
createOrReplaceGlobalTempView :创建全局视图,重复的替换;
注意:普通临时视图是 【Session会话范围内】的(更改会话、结束会话时,临时视图就不在了 !),如果想应用范围内有效,可以使用全局临时视图;
注意: 访问全局表时,要加global_temp.
前置路径访问!( 类似表空间)
scala> spark.sql("SELECT * FROM global_temp.ods1").show()
view视图和table表的区别?
view视图是查询结果集(一个快照),view只可以查不可以改! (数据源在DateFrame里面)
而table可以增删改查!
scala> val rdd=sc.makeRDD(List(1,2,3,4))
scala> val df=rdd.toDF ("id")
DataFrame 其实就是对RDD 的封装,所以可以直接获取内部的RDD;
scala> df.rdd
注意:此时得到的RDD 存储类型为Row
DataSet 是具有强类型的数据集合,需要提供对应的类型信息。
DateFrame只有数据本身的类型;
为什么有强类型的概念?
因为使用强类型访问数据会更方便,类似于一个实体类,比如User 类型可以访问username,age等属性,不用考虑顺序;
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 //(创建DateSet )
注意:在实际使用的时候,很少用到把序列转换成DataSet,更多的是通过RDD 来得到DataSet;
case class User(name: String, age: Long) // 先定义样例类
val rdd=sc.makeRDD (List(User ("zhangsan",30), User("lisi",40) ) ) //用了样例类的方式构建RDD(样例类RDD)
rdd.toDS //将RDD直接转换为DataSet
DataSet 其实也是对 RDD 的封装,所以可以直接获取内部的RDD
val rdd1=ds.rdd
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
DataSet本身就包含了结构信息和数据,即只需要去掉类型,保留结构就是DataFrame了!
ds.toDF
共性:
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 之后可以很自由的获得每一行的信息
转换图: