盘点SQL on Hadoop中用到的主要技术

　　编者按：自打Hive出现之后，经过几年的发展，SQL on Hadoop相关的系统已经百花齐放，速度越来越快，功能也越来越齐全。本文并不是要去比较所谓“交互式查询哪家强”，而是试图梳理出一个统一的视角，来看看各家系统有哪些技术上相通之处。

以下是作者原文：

　　考虑到系统使用的广泛程度与成熟度，在具体举例时一般会拿Hive和Impala为例，当然在调研的过程中也会涉及到一些其他系统，如Spark SQL，Presto，TAJO等。而对于HAWQ这样的商业产品和apache drill这样成熟度还不是很高的开源方案就不做过多了解了。

系统架构

Runtime Framework v.s. MPP

　　在SQL on Hadoop系统中，有两种架构，一种是基于某个运行时框架来构建查询引擎，典型案例是Hive；另一种是仿照过去关系数据库的MPP架构。前者现有运行时框架，然后套上SQL层，后者则是从头打造一个一体化的查询引擎。有时我们能听到一种声音，说后者的架构优于前者，至少在性能上。那么是否果真如此？

　　一般来说，对于SQL on Hadoop系统很重要的一个评价指标就是：快。后面提到的所有内容也大多是为了查询速度更快。在Hive逐渐普及之后，就逐渐有了所谓交互式查询的需求，因为无论是BI系统，还是Ad-Hoc，都不能按照离线那种节奏玩。这时候无论是有实力的大公司（比如Facebook），还是专业的供应商（比如Cloudera），都试图去解决这个问题。短期可以靠商业方案或者关系数据库去支撑一下，但是长远的解决方案就是参考过去的MPP数据库架构打造一个专门的系统，于是就有了Impala，Presto等等。从任务执行的角度说，这类引擎的任务执行其实跟DAG模型是类似的，当时也有Spark这个DAG模型的计算框架了，但这终究是别人家的孩子，而且往Spark上套SQL又是Hive的那种玩法了。于是在Impala问世之后就强调自己“计算全部在内存中完成”，性能也是各种碾压当时还只有MR作为计算模型的Hive。那么Hive所代表的“基于已有的计算模型”方式是否真的不行？

　　不可否认，按照这种方式去比较，那么类MPP模式确实有很多优势：

• DAG v.s. MR：最主要的优势，中间结果不写磁盘（除非内存不够），一气呵成。
• 流水线计算：上游stage一出结果马上推送或者拉到下一个stage处理，比如多表join时前两个表有结果直接给第三个表，不像MR要等两个表完全join完再给第三个表join。
• 高效的IO：本地查询没有多余的消耗，充分利用磁盘。这个后面细说。
• 线程级别的并发：相比之下MR每个task要启动JVM，本身就有很大延迟，占用资源也多。

　　当然MPP模式也有其劣势，一个是扩展性不是很高，这在关系数据库时代就已经有过结论；另一个是容错性差，对于Impala来说一旦运行过程中出点问题，整个查询就挂了。

　　但是，经过不断的发展，Hive也能跑在DAG框架上了，不仅有Tez，还有Spark。上面提到的一些劣势，其实大都也可以在计算模型中解决，只不过考虑到计算模型的通用性和本身的设计目标，不会去专门满足（所以如果从这个角度分类，Impala属于“专用系统”，Spark则属于“通用系统”）。在最近Cloudera做的benchmark中，虽然Impala仍然一路领先，但是基于Spark的Spark SQL完全不逊色于Presto，基于Tez的Hive也不算很差，至少在并发模式下能超过Presto，足见MPP模式并不是绝对占上风的。所以这种架构上的区别在我看来并不是制胜的关键，至少不是唯一的因素，真正要做到快速查询，各个方面的细节都要有所把握。后面说的都是这些细节。

核心组件

　　不管是上面提到的那种架构，一个SQL on Hadoop系统一般都会有一些通用的核心组件，这些组件根据设计者的考虑放在不同的节点角色中，在物理上节点都按照Master/Worker的方式去做，如果Master压力太大，一些本来适合放在Master上的组件可以放到一个辅助Master上。

• UI层负责提供用户输入查询的接口。一般有Web/GUI，命令行，编程方式3类。
• QL层负责把用户提交的查询解析成可以运行的执行计划（比如MR Job）。这部分在后面会专门提到。
• 执行层就是运行具体的Job。一般会有一个Master负责Query的运行管理，比如申请资源，观察进度等等，同时Master也负责最终聚合局部结果到全局结果。而每个节点上会有相应的Worker做本地计算。
• IO层提供与存储层交互的接口。对于HDFS来说，需要根据I/O Format把文件转换成K/V，Serde再完成K/V到数据行的映射。对于非HDFS存储来说就需要一些专门的handler/connector。
• 存储层一般是HDFS，但也有可以查询NoSQL，或者关系数据库的。
• 系统另外还需要一个元数据管理服务，管理表结构等。

执行计划

编译流程

从SQL到执行计划，大致分为5步。

• 第一步将SQL转换成抽象语法树AST。这一步一般都有第三方工具库可以完成，比如antlr。
• 第二步对AST进行语义分析，比如表是否存在，字段是否存在，SQL语义是否有误（比如select中被判定为聚合的字段在group by中有没有出现）。
• 第三步生成逻辑执行计划,这是一个由逻辑操作符组成的DAG。比如对于Hive来说扫表会产生TableScanOperator，聚合会产生GroupByOperator。对于类MPP系统来说，情况稍微有点不同。逻辑操作符的种类还是差不多，但是会先生成单机版本，然后生成多机版本。多机版本主要是把aggregate，join，还有top n这几个操作并行化，比如aggregate会分成类似MR那样的本地aggregate，shuffle和全局aggregate三步。
• 第四步做逻辑执行计划做优化，这步在下面单独介绍。
• 第五步把逻辑执行计划转换成可以在机器上运行的物理计划。对于Hive来说，就是MR/Tez Job等；对于Impala来说，就是plan fragment。其他类MPP系统也是类似的概念。物理计划中的一个计算单元（或者说Job），有“输入，处理，输出”三要素组成，而逻辑执行计划中的operator相对粒度更细，一个逻辑操作符一般处于这三要素之一的角色。

　　下面分别举两个例子，直观的认识下SQL、逻辑计划、物理计划之间的关系，具体解释各个operator的话会比较细碎，就不展开了。

Hive on MR：

1  select count(1) from status_updates where ds = '2009-08-01'

　　Presto（引用自美团技术团队，其中SubPlan就是物理计划的一个计算单元）：

[js] view plaincopy

1. select c1.rank, count(*)   
2. from dim.city c1 join dim.city c2 on c1.id = c2.id   
3. where c1.id > 10 group by c1.rank limit 10;  

优化器

　　关于执行计划的优化，虽然不一定是整个编译流程中最难的部分，但却是最有看点的部分，而且目前还在不断发展中。Spark系之所以放弃Shark另起炉灶做Spark SQL，很大一部分原因是想自己做优化策略，避免受Hive的限制，为此还专门独立出优化器组件Catalyst（当然Spark SQL目前还是非常新，其未来发展给人不少想象空间）。总之这部分工作可以不断的创新，优化器越智能，越傻瓜化，用户就越能解放出来解决业务问题。

　　早期在Hive中只有一些简单的规则优化，比如谓词下推（把过滤条件尽可能的放在table scan之后就完成），操作合并（连续的filter用and合并成一个operator，连续的projection也可以合并）。后来逐渐增加了一些略复杂的规则，比如相同key的join + group by合并为1个MR，还有star schema join。在Hive 0.12引入的相关性优化（correlation optimizer）算是规则优化的一个高峰，他能够减少数据的重复扫描，具体来说，如果查询的两个部分用到了相同的数据，并且各自做group by / join的时候用到了相同的key，这个时候由于数据源和shuffle的key是一样的，所以可以把原来需要两个job分别处理的地方合成一个job处理。

　　比如下面这个SQL：

[js] view plaincopy

1. SELECT   
2.     sum(l_extendedprice) / 7.0 as avg_yearly   
3. FROM   
4.      (SELECT l_partkey, l_quantity, l_extendedprice   
5.       FROM lineitem JOIN part ON (p_partkey=l_partkey)   
6.       WHERE p_brand='Brand#35' AND p_container = 'MED PKG')touter   
7. JOIN   
8.      (SELECT l_partkey as lp, 0.2 * avg(l_quantity) as lq   
9.       FROM lineitem GROUP BY l_partkey) tinner   
10. ON (touter.l_partkey = tinnter.lp)   
11. WHERE touter.l_quantity < tinner.lq  

　　但是，基于规则的优化（RBO）不能解决所有问题。在关系数据库中早有另一种优化方式，也就是基于代价的优化CBO。CBO通过收集表的数据信息（比如字段的基数，数据分布直方图等等）来对一些问题作出解答，其中最主要的问题就是确定多表join的顺序。CBO通过搜索join顺序的所有解空间（表太多的情况下可以用有限深度的贪婪算法），并且算出对应的代价，可以找到最好的顺序。这些都已经在关系数据库中得到了实践。

　　目前Hive已经启动专门的项目，也就是Apache Optiq来做这个事情，而其他系统也没有做的很好的CBO，所以这块内容还有很大的进步空间。