【注】该系列文章以及使用到安装包/测试数据 可以在《倾情大奉送--Spark入门实战系列》获取

1SparkSQL的发展历程

1.1 Hive and Shark

SparkSQL的前身是Shark,给熟悉RDBMS但又不理解MapReduce的技术人员提供快速上手的工具,Hive应运而生,它是当时唯一运行在Hadoop上的SQL-on-Hadoop工具。但是MapReduce计算过程中大量的中间磁盘落地过程消耗了大量的I/O,降低的运行效率,为了提高SQL-on-Hadoop的效率,大量的SQL-on-Hadoop工具开始产生,其中表现较为突出的是:

l MapRDrill

l ClouderaImpala

l Shark

其中Shark是伯克利实验室Spark生态环境的组件之一,它修改了下图所示的右下角的内存管理、物理计划、执行三个模块,并使之能运行在Spark引擎上,从而使得SQL查询的速度得到10-100倍的提升。

clip_image002

1.2 SharkSparkSQL 

但是,随着Spark的发展,对于野心勃勃的Spark团队来说,Shark对于Hive的太多依赖(如采用Hive的语法解析器、查询优化器等等),制约了SparkOne Stack Rule Them All的既定方针,制约了Spark各个组件的相互集成,所以提出了SparkSQL项目。SparkSQL抛弃原有Shark的代码,汲取了Shark的一些优点,如内存列存储(In-Memory Columnar Storage)、Hive兼容性等,重新开发了SparkSQL代码;由于摆脱了对Hive的依赖性,SparkSQL无论在数据兼容、性能优化、组件扩展方面都得到了极大的方便,真可谓“退一步,海阔天空”。

l数据兼容方面  不但兼容Hive,还可以从RDDparquet文件、JSON文件中获取数据,未来版本甚至支持获取RDBMS数据以及cassandraNOSQL数据;

l性能优化方面  除了采取In-Memory Columnar Storagebyte-code generation等优化技术外、将会引进Cost Model对查询进行动态评估、获取最佳物理计划等等;

l组件扩展方面  无论是SQL的语法解析器、分析器还是优化器都可以重新定义,进行扩展。

201461Shark项目和SparkSQL项目的主持人Reynold Xin宣布:停止对Shark的开发,团队将所有资源放SparkSQL项目上,至此,Shark的发展画上了句话,但也因此发展出两个直线:SparkSQLHive on Spark

clip_image004

其中SparkSQL作为Spark生态的一员继续发展,而不再受限于Hive,只是兼容Hive;而Hive on Spark是一个Hive的发展计划,该计划将Spark作为Hive的底层引擎之一,也就是说,Hive将不再受限于一个引擎,可以采用Map-ReduceTezSpark等引擎。

1.3 SparkSQL的性能

Shark的出现,使得SQL-on-Hadoop的性能比Hive有了10-100倍的提高:

clip_image006

那么,摆脱了Hive的限制,SparkSQL的性能又有怎么样的表现呢?虽然没有Shark相对于Hive那样瞩目地性能提升,但也表现得非常优异:

clip_image008

为什么SparkSQL的性能会得到怎么大的提升呢?主要SparkSQL在下面几点做了优化:

A:内存列存储(In-Memory Columnar Storage

SparkSQL的表数据在内存中存储不是采用原生态的JVM对象存储方式,而是采用内存列存储,如下图所示。

clip_image010

该存储方式无论在空间占用量和读取吞吐率上都占有很大优势。

对于原生态的JVM对象存储方式,每个对象通常要增加12-16字节的额外开销,对于一个270MBTPC-H lineitem table数据,使用这种方式读入内存,要使用970MB左右的内存空间(通常是25倍于原生数据空间);另外,使用这种方式,每个数据记录产生一个JVM对象,如果是大小为200B的数据记录,32G的堆栈将产生1.6亿个对象,这么多的对象,对于GC来说,可能要消耗几分钟的时间来处理(JVM的垃圾收集时间与堆栈中的对象数量呈线性相关)。显然这种内存存储方式对于基于内存计算的Spark来说,很昂贵也负担不起。

对于内存列存储来说,将所有原生数据类型的列采用原生数组来存储,将Hive支持的复杂数据类型(如arraymap等)先序化后并接成一个字节数组来存储。这样,每个列创建一个JVM对象,从而导致可以快速的GC和紧凑的数据存储;额外的,还可以使用低廉CPU开销的高效压缩方法(如字典编码、行长度编码等压缩方法)降低内存开销;更有趣的是,对于分析查询中频繁使用的聚合特定列,性能会得到很大的提高,原因就是这些列的数据放在一起,更容易读入内存进行计算。

B:字节码生成技术(bytecode generation,即CG

在数据库查询中有一个昂贵的操作是查询语句中的表达式,主要是由于JVM的内存模型引起的。比如如下一个查询:

SELECT a + b FROM table

在这个查询里,如果采用通用的SQL语法途径去处理,会先生成一个表达式树(有两个节点的Add树,参考后面章节),在物理处理这个表达式树的时候,将会如图所示的7个步骤:

1.  调用虚函数Add.eval(),需要确认Add两边的数据类型

2.  调用虚函数a.eval(),需要确认a的数据类型

3.  确定a的数据类型是Int,装箱

4.  调用虚函数b.eval(),需要确认b的数据类型

5.  确定b的数据类型是Int,装箱

6.  调用Int类型的Add

7.  返回装箱后的计算结果

其中多次涉及到虚函数的调用,虚函数的调用会打断CPU的正常流水线处理,减缓执行。

Spark1.1.0catalyst模块的expressions增加了codegen模块,如果使用动态字节码生成技术(配置spark.sql.codegen参数),SparkSQL在执行物理计划的时候,对匹配的表达式采用特定的代码,动态编译,然后运行。如上例子,匹配到Add方法:

clip_image012

然后,通过调用,最终调用:

clip_image014

最终实现效果类似如下伪代码:

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优化,可以简单地用下图来表示:

clip_image016

CScala代码优化

另外,SparkSQL在使用Scala编写代码的时候,尽量避免低效的、容易GC的代码;尽管增加了编写代码的难度,但对于用户来说,还是使用统一的接口,没受到使用上的困难。下图是一个Scala代码优化的示意图:

clip_image018

2SparkSQL运行架构

类似于关系型数据库,SparkSQL也是语句也是由Projectiona1a2a3)、Data SourcetableA)、Filtercondition)组成,分别对应sql查询过程中的ResultData SourceOperation,也就是说SQL语句按Result-->Data Source-->Operation的次序来描述的。

clip_image020

 

当执行SparkSQL语句的顺序为:

1.对读入的SQL语句进行解析(Parse),分辨出SQL语句中哪些词是关键词(如SELECTFROMWHERE),哪些是表达式、哪些是Projection、哪些是Data Source等,从而判断SQL语句是否规范;

2.SQL语句和数据库的数据字典(列、表、视图等等)进行绑定(Bind),如果相关的ProjectionData Source等都是存在的话,就表示这个SQL语句是可以执行的;

3.一般的数据库会提供几个执行计划,这些计划一般都有运行统计数据,数据库会在这些计划中选择一个最优计划(Optimize);

4.计划执行(Execute),按Operation-->Data Source-->Result的次序来进行的,在执行过程有时候甚至不需要读取物理表就可以返回结果,比如重新运行刚运行过的SQL语句,可能直接从数据库的缓冲池中获取返回结果。

2.1 TreeRule

SparkSQLSQL语句的处理和关系型数据库对SQL语句的处理采用了类似的方法,首先会将SQL语句进行解析(Parse),然后形成一个Tree,在后续的如绑定、优化等处理过程都是对Tree的操作,而操作的方法是采用Rule,通过模式匹配,对不同类型的节点采用不同的操作。在整个sql语句的处理过程中,TreeRule相互配合,完成了解析、绑定(在SparkSQL中称为Analysis)、优化、物理计划等过程,最终生成可以执行的物理计划。

2.1.1 Tree

l  Tree的相关代码定义在sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees

l  Logical PlansExpressionsPhysical Operators都可以使用Tree表示

l  Tree的具体操作是通过TreeNode来实现的

Ø  SparkSQL定义了catalyst.trees的日志,通过这个日志可以形象的表示出树的结构

Ø  TreeNode可以使用scala的集合操作方法(如foreach, map, flatMap, collect等)进行操作