《Spark入门实战系列.doc》由会员分享，可在线阅读，更多相关《Spark入门实战系列.doc（38页珍藏版）》请在金锄头文库上搜索。
1、Spark入门实战系列-6.SparkSQL（上）-SparkSQL简介【注】该系列文章以及使用到安装包/测试数据 可以在倾情大奉送-Spark入门实战系列获取1、SparkSQL的发展历程1.1Hive and SharkSparkSQL的前身是Shark，给熟悉RDBMS但又不理解MapReduce的技术人员提供快速上手的工具，Hive应运而生，它是当时唯一运行在Hadoop上的SQL-on-Hadoop工具。但是MapReduce计算过程中大量的中间磁盘落地过程消耗了大量的I/O，降低的运行效率，为了提高SQL-on-Hadoop的效率，大量的SQL-on-Hadoop工具开始产生，其中
2、表现较为突出的是：lMapR的DrilllCloudera的ImpalalShark其中Shark是伯克利实验室Spark生态环境的组件之一，它修改了下图所示的右下角的内存管理、物理计划、执行三个模块，并使之能运行在Spark引擎上，从而使得SQL查询的速度得到10-100倍的提升。1.2Shark和SparkSQL但是，随着Spark的发展，对于野心勃勃的Spark团队来说，Shark对于Hive的太多依赖（如采用Hive的语法解析器、查询优化器等等），制约了Spark的One Stack Rule Them All的既定方针，制约了Spark各个组件的相互集成，所以提出了SparkSQL项
3、目。SparkSQL抛弃原有Shark的代码，汲取了Shark的一些优点，如内存列存储（In-Memory Columnar Storage）、Hive兼容性等，重新开发了SparkSQL代码；由于摆脱了对Hive的依赖性，SparkSQL无论在数据兼容、性能优化、组件扩展方面都得到了极大的方便，真可谓“退一步，海阔天空”。l数据兼容方面不但兼容Hive，还可以从RDD、parquet文件、JSON文件中获取数据，未来版本甚至支持获取RDBMS数据以及cassandra等NOSQL数据；l性能优化方面除了采取In-Memory Columnar Storage、byte-code genera
4、tion等优化技术外、将会引进Cost Model对查询进行动态评估、获取最佳物理计划等等；l组件扩展方面无论是SQL的语法解析器、分析器还是优化器都可以重新定义，进行扩展。2014年6月1日Shark项目和SparkSQL项目的主持人Reynold Xin宣布：停止对Shark的开发，团队将所有资源放SparkSQL项目上，至此，Shark的发展画上了句话，但也因此发展出两个直线：SparkSQL和Hive on Spark。其中SparkSQL作为Spark生态的一员继续发展，而不再受限于Hive，只是兼容Hive；而Hive on Spark是一个Hive的发展计划，该计划将Spark作
5、为Hive的底层引擎之一，也就是说，Hive将不再受限于一个引擎，可以采用Map-Reduce、Tez、Spark等引擎。1.3SparkSQL的性能Shark的出现，使得SQL-on-Hadoop的性能比Hive有了10-100倍的提高：那么，摆脱了Hive的限制，SparkSQL的性能又有怎么样的表现呢？虽然没有Shark相对于Hive那样瞩目地性能提升，但也表现得非常优异：为什么SparkSQL的性能会得到怎么大的提升呢？主要SparkSQL在下面几点做了优化：A：内存列存储（In-Memory Columnar Storage）SparkSQL的表数据在内存中存储不是采用原生态的JVM
6、对象存储方式，而是采用内存列存储，如下图所示。该存储方式无论在空间占用量和读取吞吐率上都占有很大优势。对于原生态的JVM对象存储方式，每个对象通常要增加12-16字节的额外开销，对于一个270MB的TPC-H lineitem table数据，使用这种方式读入内存，要使用970MB左右的内存空间（通常是25倍于原生数据空间）；另外，使用这种方式，每个数据记录产生一个JVM对象，如果是大小为200B的数据记录，32G的堆栈将产生1.6亿个对象，这么多的对象，对于GC来说，可能要消耗几分钟的时间来处理（JVM的垃圾收集时间与堆栈中的对象数量呈线性相关）。显然这种内存存储方式对于基于内存计算的Spa
7、rk来说，很昂贵也负担不起。对于内存列存储来说，将所有原生数据类型的列采用原生数组来存储，将Hive支持的复杂数据类型（如array、map等）先序化后并接成一个字节数组来存储。这样，每个列创建一个JVM对象，从而导致可以快速的GC和紧凑的数据存储；额外的，还可以使用低廉CPU开销的高效压缩方法（如字典编码、行长度编码等压缩方法）降低内存开销；更有趣的是，对于分析查询中频繁使用的聚合特定列，性能会得到很大的提高，原因就是这些列的数据放在一起，更容易读入内存进行计算。B：字节码生成技术（bytecode generation，即CG）在数据库查询中有一个昂贵的操作是查询语句中的表达式，主要是由于
8、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类型的Add7.返回装箱后的计算结果其中多次涉及到虚函数的调用，虚函数的调用会打断CPU的正常流水
9、线处理，减缓执行。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 + bresultRow.setInt(0, result)对于Spark1.1.
10、0，对SQL表达式都作了CG优化，具体可以参看codegen模块。CG优化的实现主要还是依靠scala2.10的运行时放射机制（runtime reflection）。对于SQL查询的CG优化，可以简单地用下图来表示：C：Scala代码优化另外，SparkSQL在使用Scala编写代码的时候，尽量避免低效的、容易GC的代码；尽管增加了编写代码的难度，但对于用户来说，还是使用统一的接口，没受到使用上的困难。下图是一个Scala代码优化的示意图：2、SparkSQL运行架构类似于关系型数据库，SparkSQL也是语句也是由Projection（a1，a2，a3）、Data Source（table
11、A）、Filter（condition）组成，分别对应sql查询过程中的Result、Data Source、Operation，也就是说SQL语句按Result-Data Source-Operation的次序来描述的。当执行SparkSQL语句的顺序为：1.对读入的SQL语句进行解析（Parse），分辨出SQL语句中哪些词是关键词（如SELECT、FROM、WHERE），哪些是表达式、哪些是Projection、哪些是Data Source等，从而判断SQL语句是否规范；2.将SQL语句和数据库的数据字典（列、表、视图等等）进行绑定（Bind），如果相关的Projection、Data S
12、ource等都是存在的话，就表示这个SQL语句是可以执行的；3.一般的数据库会提供几个执行计划，这些计划一般都有运行统计数据，数据库会在这些计划中选择一个最优计划（Optimize）；4.计划执行（Execute），按Operation-Data Source-Result的次序来进行的，在执行过程有时候甚至不需要读取物理表就可以返回结果，比如重新运行刚运行过的SQL语句，可能直接从数据库的缓冲池中获取返回结果。2.1Tree和RuleSparkSQL对SQL语句的处理和关系型数据库对SQL语句的处理采用了类似的方法，首先会将SQL语句进行解析（Parse），然后形成一个Tree，在后续的如绑
13、定、优化等处理过程都是对Tree的操作，而操作的方法是采用Rule，通过模式匹配，对不同类型的节点采用不同的操作。在整个sql语句的处理过程中，Tree和Rule相互配合，完成了解析、绑定（在SparkSQL中称为Analysis）、优化、物理计划等过程，最终生成可以执行的物理计划。2.1.1TreelTree的相关代码定义在sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/treeslLogical Plans、Expressions、Physical Operators都可以使用Tree表示lTree的具体操作是通过Tree
14、Node来实现的SparkSQL定义了catalyst.trees的日志，通过这个日志可以形象的表示出树的结构TreeNode可以使用scala的集合操作方法（如foreach, map, flatMap, collect等）进行操作有了TreeNode，通过Tree中各个TreeNode之间的关系，可以对Tree进行遍历操作，如使用transformDown、transformUp将Rule应用到给定的树段，然后用结果替代旧的树段；也可以使用transformChildrenDown、transformChildrenUp对一个给定的节点进行操作，通过迭代将Rule应用到该节点以及子节点。l
15、TreeNode可以细分成三种类型的Node：UnaryNode一元节点，即只有一个子节点。如Limit、Filter操作BinaryNode二元节点，即有左右子节点的二叉节点。如Jion、Union操作LeafNode叶子节点，没有子节点的节点。主要用户命令类操作，如SetCommand2.1.2RulelRule的相关代码定义在sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ruleslRule在SparkSQL的Analyzer、Optimizer、SparkPlan等各个组件中都有应用到lRule是一个抽象类，具体的Rule实现是通过RuleExecutor完成lRule通过定义batch和batchs，可以简便的、模块化地对Tree进行transform操作lRule通过定义Once和FixedPoint，可以对Tree进行一次操作或多次操作（如对某些Tree进行多次迭代操作的时候，达到FixedPoint次数迭代或达到前后两次的树结构没变化才停止操作，具体参看RuleExecutor.apply）2.2sqlContext和hiveContext的运行过程SparkSQL有两个分支，sq
