标签:spark sparksql
Spark1.1.0中变化较大是sparkSQL和MLlib,sparkSQL1.1.0主要的变动有:
- 增加了JDBC/ODBC Server(ThriftServer),用户可以在应用程序中连接到SparkSQL并使用其中的表和缓存表。
- 增加了对JSON文件的支持
- 增加了对parquet文件的本地优化
- 增加了支持将python、scala、java的lambda函数注册成UDF,并能在SQL中直接引用
- 引入了动态字节码生成技术(bytecode generation,即CG),明显地提升了复杂表达式求值查询的速率。
- 统一API接口,如sql()、SchemaRDD生成等。
- ......
下面分十个小节来介绍sparkSQL1.1.0的架构和使用,希望各位读者joy it!
第一节:为什么sparkSQL 为本篇,介绍sparkSQL的发展历程和性能
第二节:sparkSQL架构 介绍catalyst,然后介绍sqlContext、hiveContext的运行架构及区别
第三节:sparkSQL组件之解析 介绍sparkSQL运行架构中的各个组件的功能和实现
第四节:深入了解sparkSQL之运行 使用hive/console更深入了解各种计划是如何生成的
第五节:测试环境之搭建 介绍后面章节将使用的环境搭建和测试数据
第六节:sparkSQL之基础应用 介绍sqlContext的RDD、Json、parquet使用以及hiveContext使用
第七节:ThriftServer和CLI 介绍TriftServer和CLI的使用,以及如何使用JDBC访问sparkSQL数据
第八节:sparkSQL之综合应用 介绍sparkSQL和MLlib、sparkSQL和GraphX结合使用
第九节:sparkSQL之调优 介绍CG、压缩、序化器、缓存之使用
第十节:总结
至于与hive的兼容性、具体的SQL语法以后有机会再介绍。
本篇为第一节,为什么sparkSQL?
1:sparkSQL的发展历程。
A:hive and shark
sparkSQL的前身是shark。在hadoop发展过程中,为了给熟悉RDBMS但又不理解MapReduce的技术人员提供快速上手的工具,hive应运而生,是当时唯一运行在hadoop上的SQL-on-Hadoop工具。但是,MapReduce计算过程中大量的中间磁盘落地过程消耗了大量的I/O,降低的运行效率,为了提高SQL-on-Hadoop的效率,大量的SQL-on-Hadoop工具开始产生,其中表现较为突出的是:
- MapR的Drill
- Cloudera的Impala
- Shark
其中Shark是伯克利实验室spark生态环境的组件之一,它修改了下图所示的右下角的内存管理、物理计划、执行三个模块,并使之能运行在spark引擎上,从而使得SQL查询的速度得到10-100倍的提升。
但是,随着Spark的发展,对于野心勃勃的Spark团队来说,Shark对于hive的太多依赖(如采用hive的语法解析器、查询优化器等等),制约了Spark的One Stack rule them all的既定方针,制约了spark各个组件的相互集成,所以提出了sparkSQL项目。SparkSQL抛弃原有Shark的代码,汲取了Shark的一些优点,如内存列存储(In-Memory Columnar Storage)、Hive兼容性等,重新开发了SparkSQL代码;由于摆脱了对hive的依赖性,SparkSQL无论在数据兼容、性能优化、组件扩展方面都得到了极大的方便,真可谓“退一步, 海阔天空”。
- 数据兼容方面 不但兼容hive,还可以从RDD、parquet文件、JSON文件中获取数据,未来版本甚至支持获取RDBMS数据以及cassandra等NOSQL数据
- 性能优化方面 除了采取In-Memory Columnar Storage、byte-code generation等优化技术外、将会引进Cost Model对查询进行动态评估、获取最佳物理计划等等
- 组件扩展方面 无论是SQL的语法解析器、分析器还是优化器都可以重新定义,进行扩展
2014年6月1日,Shark项目和SparkSQL项目的主持人Reynold Xin宣布:停止对Shark的开发,团队将所有资源放sparkSQL项目上,至此,Shark的发展画上了句话,但也因此发展出两个直线:SparkSQL和hive on spark。
其中sparkSQL作为Spark生态的一员继续发展,而不再受限于hive,只是兼容hive;而hive on spark是一个hive的发展计划,该计划将spark作为hive的底层引擎之一,也就是说,hive将不再受限于一个引擎,可以采用map-reduce、Tez、spark等引擎。
2:sparkSQL的性能
shark的出现,使得SQL-on-Hadoop的性能比hive有了10-100倍的提高:
那么,摆脱了hive的限制,sparkSQL的性能又有怎么样的表现呢?虽然没有shark相对于hive那样瞩目地性能提升,但也表现得非常优异:
为什么sparkSQL的性能会得到怎么大的提升呢?主要sparkSQL在下面几点做了优化:
A:内存列存储(In-Memory Columnar Storage)
sparkSQL的表数据在内存中存储不是采用原生态的JVM对象存储方式,而是采用内存列存储,如下图所示。
该存储方式无论在空间占用量和读取吞吐率上都占有很大优势。
对于原生态的JVM对象存储方式,每个对象通常要增加12-16字节的额外开销,对于一个270MB的TPC-H lineitem table数据,使用这种方式读入内存,要使用970MB左右的内存空间(通常是2~5倍于原生数据空间);另外,使用这种方式,每个数据记录产生一个JVM对象,如果是大小为200B的数据记录,32G的堆栈将产生1.6亿个对象,这么多的对象,对于GC来说,可能要消耗几分钟的时间来处理(JVM的垃圾收集时间与堆栈中的对象数量呈线性相关)。显然这种内存存储方式对于基于内存计算的spark来说,很昂贵也负担不起。
对于内存列存储来说,将所有原生数据类型的列采用原生数组来存储,将Hive支持的复杂数据类型(如array、map等)先序化后并接成一个字节数组来存储。这样,每个列创建一个JVM对象,从而导致可以快速的GC和紧凑的数据存储;额外的,还可以使用低廉CPU开销的高效压缩方法(如字典编码、行长度编码等压缩方法)降低内存开销;更有趣的是,对于分析查询中频繁使用的聚合特定列,性能会得到很大的提高,原因就是这些列的数据放在一起,更容易读入内存进行计算。
B:字节码生成技术(bytecode generation,即CG)
在数据库查询中有一个昂贵的操作是查询语句中的表达式,主要是由于JVM的内存模型引起的。比如如下一个查询:
在这个查询里,如果采用通用的SQL语法途径去处理,会先生成一个表达式树(有两个节点的Add树,参考后面章节),在物理处理这个表达式树的时候,将会如图所示的7个步骤:
- 调用虚函数Add.eval(),需要确认Add两边的数据类型
- 调用虚函数a.eval(),需要确认a的数据类型
- 确定a的数据类型是Int,装箱
- 调用虚函数b.eval(),需要确认b的数据类型
- 确定b的数据类型是Int,装箱
- 调用Int类型的Add
- 返回装箱后的计算结果
其中多次涉及到虚函数的调用,虚函数的调用会打断CPU的正常流水线处理,减缓执行。
Spark1.1.0在catalyst模块的expressions增加了codegen模块,如果使用动态字节码生成技术(配置spark.sql.codegen参数),sparkSQL在执行物理计划的时候,对匹配的表达式采用特定的代码,动态编译,然后运行。如上例子,匹配到Add方法:
最终实现效果类似如下伪代码:
val a: Int = inputRow.getInt(0)
val b: Int = inputRow.getInt(1)
val result: Int = a + b
resultRow.setInt(0, result)
对于Spark1.1.0,对SQL表达式都作了CG优化,具体可以参看codegen模块。CG优化的实现主要还是依靠scala2.10的运行时放射机制(runtime reflection)。对于SQL查询的CG优化,可以简单地用下图来表示:
C:scala代码优化
另外,sparkSQL在使用Scala编写代码的时候,尽量避免低效的、容易GC的代码;尽管增加了编写代码的难度,但对于用户来说,还是使用统一的接口,没受到使用上的困难。下图是一个scala代码优化的示意图:
sparkSQL1.1入门之一:为什么sparkSQL
标签:spark sparksql
原文地址:http://blog.csdn.net/book_mmicky/article/details/39288715