【编者按】星环科技从2013年6月开始研发基于Spark的SQL执行引擎，在2013年底推出Transwarp Inceptor 1.0，并落地了国内首个7x24小时的商用项目。经过1年多的持续创新与改进，星环已经在国内落地了数十个Inceptor的商用项目。这是一篇星环Spark解决方案的技术解析，也是Spark用户可以效仿的优化之道。
以下为原文
当前Hadoop技术蓬勃发展，用于解决大数据的分析难题的技术平台开始涌现。Spark凭借性能强劲、高度容错、调度灵活等技术优势已渐渐成为主流技术，业界大部分厂商都提供了基于Spark的技术方案和产品。根据Databricks的统计，目前有11个商业的Spark版本。
在使用Spark作出计算平台的解决方案中，有两种主流编程模型，一类是基于Spark API或者衍生出来的语言，另一种是基于SQL语言。SQL作为数据库领域的事实标准语言，相比较用API（如MapReduce API，Spark API等）来构建大数据分析的解决方案有着先天的优势：一是产业链完善，各种报表工具、ETL工具等可以很好的对接；二是用SQL开发有更低的技术门槛；三是能够降低原有系统的迁移成本等。因此，SQL语言也渐渐成为大数据分析的主流技术标准。本文将深入解析Inceptor的架构、编程模型和编译优化技术，并提供基准测试在多平台上的性能对比数据。
1. Inceptor架构
Transwarp Inceptor是基于Spark的分析引擎，如图1所示，从下往上有三层架构：最下面是存储层，包含分布式内存列式存储（Transwarp Holodesk），可建在内存或者SSD上；中间层是Spark计算引擎层，星环做了大量的改进保证引擎有超强的性能和高度的健壮性；最上层包括一个完整的SQL 99和PL/SQL编译器、统计算法库和机器学习算法库，提供完整的R语言访问接口。
图1：Transwarp Inceptor架构图
Transwarp Inceptor可以分析存储在HDFS、HBase或者Transwarp Holodesk分布式缓存中的数据，可以处理的数据量从GB到数十TB，即使数据源或者中间结果的大小远大于内存容量也可高效处理。另外Transwarp Inceptor通过改进Spark和YARN的组合，提高了Spark的可管理性。同时星环不仅仅是将Spark作为一个缺省计算引擎，也重写了SQL编译器，提供更加完整的SQL支持。
同时，Transwarp Inceptor还通过改进Spark使之更好地与HBase融合，可以为HBase提供完整的SQL支持，包括批量SQL统计、OLAP分析以及高并发低延时的SQL查询能力，使得HBase的应用可以从简单的在线查询应用扩展到复杂分析和在线应用结合的混合应用中，大大拓展了HBase的应用范围。
2. 编程模型
Transwarp Inceptor提供两种编程模型：一是基于SQL的编程模型，用于常规的数据分析、数据仓库类应用市场；二是基于数据挖掘编程模型，可以利用R语言或者Spark MLlib来做一些深度学习、数据挖掘等业务模型。
2.1 SQL模型
Transwarp Inceptor实现了自己的SQL解析执行引擎，可以兼容SQL 99和HiveQL，自动识别语法，因此可以兼容现有的基于Hive开发的应用。由于Transwarp Inceptor完整支持标准的SQL 99标准，传统数据库上运行的业务可以非常方便的迁移到Transwarp Inceptor系统上。此外Transwarp Inceptor支持PL/SQL扩展，传统数据仓库的基于PL/SQL存储过程的应用（如ETL工具）可以非常方便的在Inceptor上并发执行。另外Transwarp Inceptor支持部分SQL 2003标准，如窗口统计功能、安全审计功能等，并对多个行业开发了专门的函数库，因此可以满足多个行业的特性需求。
2.2 数据挖掘计算模型
Transwarp Inceptor实现了机器学习算法库与统计算法库，支持常用机器学习算法并行化与统计算法并行化，并利用Spark在迭代计算和内存计算上的优势，将并行的机器学习算法与统计算法运行在Spark上。例如：机器学习算法库有包括逻辑回归、朴素贝叶斯、支持向量机、聚类、线性回归、关联挖掘、推荐算法等，统计算法库包括均值、方差、中位数、直方图、箱线图等。Transwarp Inceptor可以支持用R语言或者Spark API在平台上搭建多种分析型应用，例如用户行为分析、精准营销、对用户贴标签、进行分类。
3. SQL编译与优化
Transwarp Inceptor研发了一套完整的SQL编译器，包括HiveQL解析器、SQL标准解析器和PL/SQL解析器，将不同的SQL语言解析成中间级表示语言，然后经过优化器转换成物理执行计划。SQL语言解析后经过逻辑优化器生成中间级表示语言，而中间表示语言再经过物理优化器生成最终的物理执行计划。从架构上分，逻辑优化器和物理优化器都包含基于规则的优化模块和基于成本的优化模块。
为了和Hadoop生态更好的兼容，Inceptor为一个SQL查询生成Map Reduce上的执行计划和Spark上的执行计划，并且可以通过一个SET命令在两种执行引擎之间切换。
图2：SQL编译框架
3.1 SQL编译与解析
Transwarp Inceptor的SQL编译器会根据输入的SQL查询的类型来自动选择不同的解析器，如PL/SQL存储过程会自动进入PL/SQL解析器并生成一个Spark RDD的DAG从而在Spark平台上并行计算，标准SQL查询会进入SQL标准解析器生成Spark或Map Reduce执行计划。由于HiveQL和标准的SQL有所出入，为了兼容HiveQL，Transwarp Inceptor保留了HiveQL解析器，并可以对非标准SQL的Hive查询生成Spark或者Map Reduce执行计划。
3.1.1 SQL 标准解析器
Transwarp Inceptor构建了自主研发的SQL标准解析器，用于解析SQL 99 & SQL 2003查询并生成Spark和Map Reduce的执行计划。词法和语法分析层基于Antlr语法来构建词法范式，通过Antlr来生成抽象语义树，并会通过一些上下文的语义来消除冲突并生成正确的抽象语义树。语义分析层解析上层生成的抽象语义树，根据上下文来生成逻辑执行计划并传递给优化器。首先Transwarp Inceptor会将SQL解析成TABLE SCAN、SELECT、FILTER、JOIN、UNION、ORDER BY、GROUP BY等主要的逻辑块，接着会根据一些Meta信息进一步细化各个逻辑块的执行计划。如TABLE SCAN会分成块读取、块过滤、行级别过滤、序列化等多个执行计划。
3.1.2 PL/SQL 解析器
PL/SQL是Oracle对SQL语言的模块化扩展，已经在很多行业中有大规模的应用，是数据仓库领域的重要编程语言。
为了让存储过程在Spark上有较好的性能，PL/SQL解析器会根据存储过程中的上下文关系来生成SQL DAG，然后对各SQL的执行计划生成的RDD进行二次编译，通过物理优化器将一些没有依赖关系的RDD进行合并从而生成一个最终的RDD DAG。因此，一个存储过程被解析成一个大的DAG，从而stage之间可以大量并发执行，避免了多次执行SQL的启动开销并保证了系统的并发性能。
解析并生成SQL级别的执行计划
解析SQL的依赖关系并生成DAG, 再根据各个SQL的执行计划来生成最终存储过程的Spark RDD DAG
3.2 SQL优化器
Transwarp Inceptor使用Spark作为默认计算引擎，并且开发了完善的SQL优化器，因此在大量的客户案例性能测试中，Transwarp Inceptor的性能领先Map Reduce 10-100倍，并超越部分开源MPP数据库。SQL优化器对平台性能的提升居功至伟。
3.2.1 基于规则的优化器（Rule Based Optimizer）
目前为止，Transwarp Inceptor共实现了一百多个优化规则，并且在持续的添加新的规则。按照功能划分，这些规则主要分布在如下几个模块：
文件读取时过滤
在文件读取时过滤数据能够最大化的减少参与计算的数据量从而最为有效的提高性能，因此Transwarp Inceptor提供了多个规则用于生成表的过滤条件。对于一些SQL中的显示条件，Transwarp Inceptor会尽量将过滤前推到读取表中；而对于一些隐式的过滤条件，如可以根据join key生成的过滤规则，Inceptor会根据语义保证正确性的前提下进行规则生成。
过滤条件前置
Transwarp Inceptor能够从复杂的组合过滤条件中筛选出针对特定表的过滤规则，然后通过SQL语义来确定是否能将过滤条件前推到尽量早的时候执行。如果有子查询，过滤条件可以递归前推入最低层的子查询中，从而保证所有的冗余数据被删除。
超宽表的读取过滤
对一些列超多的表进行处理的时候，Transwarp Inceptor首先会根据SQL语义来确定要读取的列，并在读取表的时候进行跨列读取减少IO和内存消耗。而如果表有过滤条件，Inceptor会做进一步优化，首先只读取过滤条件相关的列来确定该行记录是否需要被选择，如果不是就跳过当前行的所有列，因此能够最大程度上的减少数据读取。在一些商业实施中，这些优化规则能够带来5x - 10x的性能提升。
Shuffle Stage的优化与消除
Spark的shuffle实现的效率非常低，需要把结果写磁盘，然后通过HTTP传输。Transwarp Inceptor添加了一些shuffle消除的优化规则，对SQL的DAG中不必要或者是可以合并的shuffle stage进行消除或者合并。对于必须要做Shuffle的计算任务，Inceptor通过DAGScheduler来提高shuffle的效率：Map Task会直接将结果返回给DAGScheduler，然后DAGScheduler将结果直接交给Reduce Task而不是等待所有Map Task结束，这样能够非常明显的提升shuffle阶段的性能。
Partition消除
Transwarp Inceptor提供单一值Partition和Range Partition，并且支持对Partition建Bucket来做多次分区。当Partition过多的时候，系统的性能会因为内存消耗和调度开销而损失。因此，Inceptor提供了多个规则用于消除不必要的Partition，如果上下文中有隐式的对Partition的过滤条件，Inceptor也会生成对partition的过滤规则。
3.2.2 基于成本的优化器（Cost Based Optimizer）
基于规则的优化器都是根据一些静态的信息来产生的，因此很多和动态数据相关的特性是不能通过基于规则的优化来解决，因此Transwarp Inceptor提供了基于成本的优化器来做二次优化。相关的原始数据主要来自Meta-store中的表统计信息、RDD的信息、SQL上下文中的统计信息等。依赖于这些动态的数据，CBO会计算执行计划的物理成本并选择最有效的执行计划。一些非常有效的优化规则包括如下几点：
JOIN顺序调优
在实际的案例中，join是消耗计算量最多的业务，因此对join的优化至关重要。在多表JOIN模型中，Transwarp Inceptor会根据统计信息来预估join的中间结果大小，并选择产生中间数据量最小的join顺序作为执行计划。
JOIN类型的选择
Transwarp Inceptor支持Left-most Join Tree 和 Bush Join Tree，并且会根据统计信息来选择生成哪种Join模型有最佳性能。此外，Transwarp Inceptor会根据原始表或者中间数据的大小来选择是否开启针对数据倾斜模型下的特殊优化等。此外，针对HBase表是否有索引的情况，Transwarp Inceptor会在普通Join和Look-up Join间做个均衡的选择。
并发度的控制
Spark通过线程级并发来提高性能，但是大量的并发可能会带来不必要的调度开销，因此不同的案例在不同并发度下会有最佳性能。Transwarp Inceptor通过对RDD的一些属性进行推算来选择最佳并发控制，对很多的案例有着2x-3x的性能提升。
4.Transwarp Holodesk内存计算引擎
为了有效的降低SQL分析的延时，减少磁盘IO对系统性能的影响，星环科技研发了基于内存或者SSD的存储计算引擎Transwarp Holodesk，通过将表数据直接建在内存或者SSD上以实现SQL查询全内存计算。另外Transwarp Holodesk增加了数据索引功能，支持对多个数据列建索引，从而更大程度的降低了SQL查询延时。
4.1 存储格式
Transwarp Holodesk基于列式存储做了大量的原创性改进带来更高的性能和更低的数据膨胀率。首先数据被序列化后存储到内存或SSD上以节省者资源占用。如图3所示，每个表的数据被存储成若干个Segment，每个Segment被划分成若干个Block，每个Block按照列方式存储于SSD或内存中。另外每个Block的头部都加上Min-Max Filter和Bloom Filter用于过滤无用的数据块，减少不必要的数据进入计算阶段。
Transwarp Holodesk根据查询条件的谓词属性对每个数据块的对应列构建数据索引，索引列采用自己研发的Trie结构进行组织存储，非索引列采用字典编码的方式进行组织存储。Trie不仅能对具有公共前缀的字符串进行压缩，而且可以对输入的字符串排序，从而可以利用二分查找快速查询所需数据的位置，从而快速响应查询需求。
图3：Holodesk存储格式
HDFS 2.6支持Storage Tier让应用程序可以选择存储层为磁盘或者SSD，但是没有专用的存储格式设计是无法有效利用SSD的读写吞吐量和低延，因此现有的Text以及行列混合（ORC/Parquet）都不能有效的利用SSD的高性能。为此验证存储结构对性能的影响，我们将HDFS构建在SSD上并选用某基准测试来做了进一步的性能对比，结果如图4所示：采用文本格式，PCI-E SSD带来的性能提升仅1.5倍；采用专为内存和SSD设计的Holodesk列式存储，其性能相比较SSD上的HDFS提升高达6倍。
图4：SSD上Holodesk对HDFS的性能加速比
4.2 性能优势
某运营商客户在12台x86服务器上搭建了Transwarp Inceptor，将Transwarp Holodesk 配置在PCIE-SSD上，并与普通磁盘表以及DB2来做性能对比测试。最终测试数据如图5所示：
图5：某运营商Holodesk性能测试结果
在纯粹的count测试一项，Holodesk性能相对于磁盘表最高领先32倍；对于join测试一项，Transwarp Holodesk最高领先磁盘表多达12倍；在单表聚合测试中，Holodesk提升倍数达10~30倍。另外Transwarp Holodesk在和DB2的对比中也表现优秀，两个复杂SQL查询在DB2数据库中需要运行1小时以上，但是在使用Transwarp Holodesk均是分钟级和秒级就返回结果。
内存的价格大约是同样容量SSD的十倍左右，为了给企业提供更高性价比的计算方案，Transwarp Holodesk针对SSD进行了大量的优化，使得应用在SSD上运行具有与在内存上比较接近的性能，从而为客户提供了性价比更高的计算平台。
在对TPC-DS的IO密集型查询的测试中，无论上构建在PCI-E SSD还是内存上，Holodesk对比磁盘表有一个数量级上的性能提升；而SSD上的Holodesk性能只比内存差10%左右。
图6：数据在磁盘、SSD和内存中的性能表现
5. 稳定的Spark执行引擎
企业目前应用开源Spark的主要困难在稳定性、可管理性和功能不够丰富上。开源Spark在稳定性上还有比较多的问题，在处理大数据量时可能无法运行结束或出现Out of memory，性能时快时慢，有时比Map/Reduce更慢，无法应用到复杂数据分析业务中。
Transwarp Inceptor针对各种出错场景设计了多种解决方法，如通过基于成本的优化器选择最合适的执行计划、加强对数据结构内存使用效率的有效管理、对常见的内存出错问题通过磁盘进行数据备份等方式，极大提高了Spark功能和性能的稳定性，上述问题都已经解决并经过商业案例的考验。Transwarp Inceptor能稳定的运行7*24小时，并能在TB级规模数据上高效进行各种稳定的统计分析。
6. SQL引擎效能验证
TPC-DS是TPC组织为Decision Support System设计的一个测试集，包含对大数据集的统计／报表生成／联机查询／数据挖掘等复杂应用，测试用的数据有各种不同的分布与倾斜，与真实场景非常接近。随着国内外各代表性的Hadoop发行版厂商以TPC-DS为标准测评产品，TPC-DS也就逐渐成为了业界公认的Hadoop系统测试准则。
6.1 验证对比的平台和配置
我们搭建了两个集群分别用于Transwarp Inceptor与Cloudera Data Hub/Impala的测试。每个集群采用4台普通两路x86服务器搭建，每台服务器硬件配置如下：
CPU两颗Intel® Xeon® Processor E5-2620 v2 (15M Cache, 2.10 GHz, 6 cores)内存每个节点128GB内存
磁盘6块1TB SATA硬盘, 7200RPM
网络千兆全双工网络
考虑到磁盘的容量和HDFS的存储复制模式，我们选择的是500GB的数据总量。SQL测试案例的选择上，在Cloudera Impala中使用的是由Cloudera改动过的TPC-DS测试子集，在Transwarp Inceptor我们选用的是TPC-DS为Oracle生成的测试集合，保留了原有的各种复杂SQL，因此能够客观反映出Inceptor在SQL支持上的情况。
Transwarp Inceptor 4.0
CDH 5.1.3
测试数据容量
500 GB
500 GB
SQL测试案例
官方标准版本
Cloudera定制版本
6.2 Transwarp Inceptor VS Cloudera Impala
Transwarp Inceptor由于有完善的SQL支持，能够运行全部所有的99个SQL查询。而由于Cloudera官方发布的TPC-DS测试集只包含19个SQL案例，因此我们只能运行这19个SQL，实验证明这部分查询在Impala上全部正常运行完成。
图7是所有的测试集合的性能对比图。图中纵坐标小于1表述测试案例中Cloudera Impala性能超过Transwarp Inceptor，而大于1则表示Transwarp Inceptor有更好的性能表现。对于Cloudera Impala不能支持的SQL，我们就标记这个性能比为100。从图中可见，在Cloudera Impala支持的19个SQL中，有8个SQL的表现超过Transwarp Inceptor，2个表现相当，另外9个Transwarp Inceptor比Cloudera Impala表现的更好。
图7：Transwarp Inceptor与Cloudera Impala的性能比较
6.3 Transwarp Inceptor VS Map Reduce
我们使用了同样的硬件和软件配置完成和开源的Hive执行效率相比，Transwarp Inceptor能够带来10x-100x的性能提升。图8是TPC-DS的部分SQL查询在Inceptor和CDH 5.1 Hive的性能提升倍数，其中最大的提升倍数竟可达到123倍。
图8：Transwarp Inceptor与开源Hive的性能比较
7. 结语
随着在大数据领域国内外开始处于同一起跑线，我们相信像星环科技这样国内具有代表性的Hadoop发行版厂商将在中国的广阔市场空间中获得长足发展，并且由于中国市场激烈的竞争与磨练，逐步打磨出超越国外先进厂商的技术与实力。
孙元浩，星环信息科技创始人兼CTO，2013年创办星环科技，致力于Hadoop之上的高效计算引擎和数据分析算法的研发。孙元浩曾是英特尔Hadoop发行版的创始人，带领团队研发企业级Hadoop发行版，并对Hadoop做出多项功能增强和改进，以满足不同行业对Hadoop的功能需求。孙元浩及其团队在中国成功建立几十个Hadoop成功案例，积极推动了大数据方案在中国的落地和实施。
由“2015 OpenStack技术大会”、“2015 Spark技术峰会”、“2015 Container技术峰会” 所组成的 OpenCloud 2015大会于4月17-18日在北京召开。日程已经全部公开！懂行的人都在这里！（优惠票价期，速来）
