浅谈Hive on Spark的原理

Hive是基于Hadoop平台的数据仓库，最初由Facebook开发，在通过多年发展以后，已经成为Hadoop事实上的SQL引擎标准。相较于其余诸如Impala、Shark（SparkSQL的前身）等引擎而言，Hive拥有更为普遍的用户基础以及对SQL语法更全面的支持。Hive最初的计算引擎为MapReduce，受限于其自身的Map+Reduce计算模式，以及不够充分的大内利用，MapReduce的性能难以获得提高。
Hortonworks于2013年提出将Tez做为另外一个计算引擎以提升Hive的性能。Spark则是最初由加州大学伯克利分校开发的分布式计算引擎，借助于其灵活的DAG执行模式、对内存的充分利用，以及RDD所能表达的丰富语义，Spark受到了Hadoop社区的普遍关注。在成为Apache顶级项目以后，Spark更是集成了流处理、图计算、机器学习等功能，是业界公认最具潜力的下一代通用计算框架。鉴于此，Hive社区于2014年推出了Hive on Spark项目（HIVE-7292），将Spark做为继MapReduce和Tez以后Hive的第三个计算引擎。该项目由Cloudera、Intel和MapR等几家公司共同开发，并受到了来自Hive和Spark两个社区的共同关注。目前Hive onSpark的功能开发已基本完成，并于2015年1月初合并回trunk，预计会在Hive下一个版本中发布。本文将介绍Hive onSpark的设计架构，包括如何在Spark上执行Hive查询，以及如何借助Spark来提升Hive的性能等。另外本文还将介绍Hive onSpark的进度和计划，以及初步的性能测试数据。
背景
Hive on Spark是由Cloudera发起，由Intel、MapR等公司共同参与的开源项目，其目的是把Spark做为Hive的一个计算引擎，将Hive的查询做为Spark的任务提交到Spark集群上进行计算。尚学堂陈老师提示经过该项目，能够提升Hive查询的性能，同时为已经部署了Hive或者Spark的用户提供了更加灵活的选择，从而进一步提升Hive和Spark的普及率。
在介绍Hive on Spark的具体设计以前，先简单介绍一下Hive的工做原理，以便于你们理解如何把Spark做为新的计算引擎供给Hive使用。
在Hive中， 一条SQL语句从用户提交到计算并返回结果，大体流程以下图所示（Hive 0.14中引入了基于开销的优化器（Cost BasedOptimizer，CBO），优化的流程会略有不一样。
语法分析阶段，Hive利用Antlr将用户提交的SQL语句解析成一棵抽象语法树（Abstract Syntax Tree，AST）。
生成逻辑计划包括经过Metastore获取相关的元数据，以及对AST进行语义分析。获得的逻辑计划为一棵由Hive操做符组成的树，Hive操做符即Hive对表数据的处理逻辑，好比对表进行扫描的TableScanOperator，对表作Group的GroupByOperator等。
逻辑优化即对Operator Tree进行优化，与以后的物理优化的区别主要有两点：一是在操做符级别进行调整；二是这些优化不针对特定的计算引擎。好比谓词下推（Predicate Pushdown）就是一个逻辑优化：尽早的对底层数据进行过滤以减小后续须要处理的数据量，这对于不一样的计算引擎都是有优化效果的。
生成物理计划即针对不一样的引擎，将Operator Tree划分为若干个Task，并按照依赖关系生成一棵Task的树（在生成物理计划以前，各计算引擎还能够针对自身需求，对Operator Tree再进行一轮逻辑优化）。好比，对于MapReduce，一个GROUP BY+ORDER BY的查询会被转化成两个MapReduce的Task，第一个进行Group，第二个进行排序。
物理优化则是各计算引擎根据自身的特色，对Task Tree进行优化。好比对于MapReduce，Runtime Skew Join的优化就是在原始的Join Task以后加入一个Conditional Task来处理可能出现倾斜的数据。
最后按照依赖关系，依次执行Task Tree中的各个Task，并将结果返回给用户。每一个Task按照不一样的实现，会把任务提交到不一样的计算引擎上执行。
整体设计
Hive on Spark整体的设计思路是，尽量重用Hive逻辑层面的功能；从生成物理计划开始，提供一整套针对Spark的实现，好比SparkCompiler、SparkTask等，这样Hive的查询就能够做为Spark的任务来执行了。如下是几点主要的设计原则。
尽量减小对Hive原有代码的修改。这是和以前的Shark设计思路最大的不一样。Shark对Hive的改动太大以致于没法被Hive社区接受，Hive on Spark尽量少改动Hive的代码，从而不影响Hive目前对MapReduce和Tez的支持。同时，Hive on Spark保证对现有的MapReduce和Tez模式在功能和性能方面不会有任何影响。
对于选择Spark的用户，应使其可以自动的获取Hive现有的和将来新增的功能。
尽量下降维护成本，保持对Spark依赖的松耦合。
基于以上思路和原则，具体的一些设计架构以下。
新的计算引擎
Hive的用户能够经过hive.execution.engine来设置计算引擎，目前该参数可选的值为mr和tez。为了实现Hive onSpark，咱们将spark做为该参数的第三个选项。要开启Hive on Spark模式，用户仅需将这个参数设置为spark便可。
以Hive的表做为RDD
Spark以分布式可靠数据集（Resilient Distributed Dataset，RDD）做为其数据抽象，所以咱们须要将Hive的表转化为RDD以便Spark处理。本质上，Hive的表和Spark的HadoopRDD都是HDFS上的一组文件，经过InputFormat和RecordReader读取其中的数据，所以这个转化是天然而然的。
使用Hive原语
这里主要是指使用Hive的操做符对数据进行处理。Spark为RDD提供了一系列的转换（Transformation），其中有些转换也是面向SQL的，如groupByKey、join等。但若是使用这些转换（就如Shark所作的那样），就意味着咱们要从新实现一些Hive已有的功能；并且当Hive增长新的功能时，咱们须要相应地修改Hive on Spark模式。有鉴于此，咱们选择将Hive的操做符包装为Function，而后应用到RDD上。这样，咱们只须要依赖较少的几种RDD的转换，而主要的计算逻辑仍由Hive提供。
因为使用了Hive的原语，所以咱们须要显式地调用一些Transformation来实现Shuffle的功能。下表中列举了Hive onSpark使用的全部转换。