Hive SQL的编译过程

　　Hive是基于Hadoop的一个数据仓库系统，在各大公司都有广泛的应用。美团数据仓库也是基于Hive搭建，每天执行近万次的HiveETL计算流程，负责每天数百GB的数据存储和分析。Hive的稳定性和性能对我们的数据分析非常关键。

　　在几次升级Hive的过程中，我们遇到了一些大大小小的问题。通过向社区的咨询和自己的努力，在解决这些问题的同时我们对Hive将SQL编译为MapReduce的过程有了比较深入的理解。对这一过程的理解不仅帮助我们解决了一些Hive的bug，也有利于我们优化HiveSQL，提升我们对Hive的掌控力，同时有能力去定制一些需要的功能。

　　将GroupBy的字段组合为map的输出key，利用MapReduce的排序，在reduce阶段保存LastKey区分不同的key。MapReduce的过程如下（当然这里只是说明Reduce端的非Hash聚合过程）

　　当只有一个distinct字段时，如果不考虑Map阶段的HashGroupBy，只需要将GroupBy字段和Distinct字段组合为map输出key，利用mapreduce的排序，同时将GroupBy字段作为reduce的key，在reduce阶段保存LastKey即可完成去重

　　实现方式有两种：

　　（1）如果仍然按照上面一个distinct字段的方法，即下图这种实现方式，无法跟据uid和date分别排序，也就无法通过LastKey去重，仍然需要在reduce阶段在内存中通过Hash去重

　　（2）第二种实现方式，可以对所有的distinct字段编号，每行数据生成n行数据，那么相同字段就会分别排序，这时只需要在reduce阶段记录LastKey即可去重。

　　这种实现方式很好的利用了MapReduce的排序，节省了reduce阶段去重的内存消耗，但是缺点是增加了shuffle的数据量。

　　需要注意的是，在生成reducevalue时，除第一个distinct字段所在行需要保留value，其余distinct数据行value字段均可为空。

　　Hive使用Antlr实现SQL的词法和语法解析。Antlr是一种语言识别的工具，可以用来构造领域语言。

　　这里不详细介绍Antlr，只需要了解使用Antlr构造特定的语言只需要编写一个语法文件，定义词法和语法替换规则即可，Antlr完成了词法分析、语法分析、语义分析、中间代码生成的过程。

　　Hive中语法规则的定义文件在0.10版本以前是Hive.g一个文件，随着语法规则越来越复杂，由语法规则生成的Java解析类可能超过Java类文件的最大上限，0.11版本将Hive.g拆成了5个文件，词法规则HiveLexer.g和语法规则的4个文件SelectClauseParser.g，FromClauseParser.g，IdentifiersParser.g，HiveParser.g。

　　经过词法和语法解析后，如果需要对表达式做进一步的处理，使用Antlr的抽象语法树语法AbstractSyntaxTree，在语法分析的同时将输入语句转换成抽象语法树，后续在遍历语法树时完成进一步的处理。

　　下面的一段语法是HiveSQL中SelectStatement的语法规则，从中可以看出，SelectStatement包含select,from,where,groupby,having,orderby等子句。

　　（在下面的语法规则中，箭头表示对于原语句的改写，改写后会加入一些特殊词标示特定语法，比如TOK_QUERY标示一个查询块）

　　最终生成的ASTTree如下图右侧（使用AntlrWorks生成，AntlrWorks是Antlr提供的编写语法文件的编辑器），图中只是展开了骨架的几个节点，没有完全展开。

　　子查询1/2，分别对应右侧第1/2两个部分。

　　这里注意一下内层子查询也会生成一个TOK_DESTINATION节点。请看上面SelectStatement的语法规则，这个节点是在语法改写中特意增加了的一个节点。原因是Hive中所有查询的数据均会保存在HDFS临时的文件中，无论是中间的子查询还是查询最终的结果，Insert语句最终会将数据写入表所在的HDFS目录下。

　　详细来看，将内存子查询的from子句展开后，得到如下ASTTree，每个表生成一个TOK_TABREF节点，Join条件生成一个“=”节点。其他SQL部分类，不一一详述。

　　QueryBlock是一条SQL最基本的组成单元，包括三个部分：输入源，计算过程，输出。简单来讲一个QueryBlock就是一个子查询。

　　下图为Hive中QueryBlock相关对象的类图，解释图中几个重要的属性

　　QB#qbp即QBParseInfo保存一个基本SQL单元中的给个操作部分的ASTTree结构，QBParseInfo#nameToDest这个HashMap保存查询单元的输出，key的形式是inclause-i（由于Hive支持MultiInsert语句，所以可能有多个输出），value是对应的ASTNode节点，即TOK_DESTINATION节点。类QBParseInfo其余HashMap属性分别保存输出和各个操作的ASTNode节点的对应关系。

　　Hive最终生成的MapReduce任务，Map阶段和Reduce阶段均由OperatorTree组成。逻辑操作符，就是在Map阶段或者Reduce阶段完成单一特定的操作。

　　基本的操作符包括TableScanOperator，SelectOperator，FilterOperator，JoinOperator，GroupByOperator，ReduceSinkOperator

　　从名字就能猜出各个操作符完成的功能，TableScanOperator从MapReduce框架的Map接口原始输入表的数据，控制扫描表的数据行数，标记是从原表中取数据。JoinOperator完成Join操作。FilterOperator完成过滤操作

　　ReduceSinkOperator将Map端的字段组合序列化为ReduceKey/value,PartitionKey，只可能出现在Map阶段，同时也标志着Hive生成的MapReduce程序中Map阶段的结束。

　　Operator在MapReduce阶段之间的数据传递都是一个流式的过程。每一个Operator对一行数据完成操作后之后将数据传递给childOperator计算。

　　Operator类的主要属性和方法如下

　　由于Hive的MapReduce程序是一个动态的程序，即不确定一个MapReduceJob会进行什么运算，可能是Join，也可能是GroupBy，所以Operator将所有运行时需要的参数保存在OperatorDesc中，OperatorDesc在提交任务前序列化到HDFS上，在MapReduce任务执行前从HDFS读取并反序列化。Map阶段OperatorTree在HDFS上的位置在Job.getConf(“hive.exec.plan”)“/map.xml”

　　由于Join/GroupBy/OrderBy均需要在Reduce阶段完成，所以在生成相应操作的Operator之前都会先生成一个ReduceSinkOperator，将字段组合并序列化为ReduceKey/value,PartitionKey

　　接下来详细分析样例SQL生成OperatorTree的过程

　　先序遍历上一个阶段生成的QB对象

　　表中①的优化器均是一个Job干尽可能多的事情/合并。②的都是减少shuffle数据量，甚至不做Reduce。

　　CorrelationOptimizer优化器非常复杂，都能利用查询中的相关性，合并有相关性的Job，参考HiveCorrelationOptimizer

　　对于样例SQL，有两个优化器对其进行优化。下面分别介绍这两个优化器的作用，并补充一个优化器ReduceSinkDeDuplication的作用

　　ReduceSinkDeDuplication可以合并线性相连的两个RS。实际上CorrelationOptimizer是ReduceSinkDeDuplication的超集，能合并线性和非线性的操作RS，但是Hive先实现的ReduceSinkDeDuplication

　　譬如下面这条SQL语句

　　ReduceSinkDeDuplication优化器检测到：1.pRSKey完全包含cRSKey，且排序顺序一致；2.pRSPartitionKey完全包含cRSPartitionKey。符合优化条件，会对执行计划进行优化。

　　ReduceSinkDeDuplication将childRS和parentheRS与childRS之间的Operator删掉，保留的RS的Key为key,value字段，PartitionKey为key字段。合并后的OperatorTree如下：

　　继续遍历TS[p]的子Operator，将子Operator存入栈opStack中

　　当第一个RS进栈后，即栈opStack={TS[p],FIL[18],RS[4]}时，就会满足下面的规则R2

　　MapJoin简单说就是在Map阶段将小表读入内存，顺序扫描大表完成Join。

　　上图是HiveMapJoin的原理图，出自Facebook工程师LiyinTang的一篇介绍Join优化的slice，从图中可以看出MapJoin分为两个阶段：

　　MapReduceJob在Map阶段，每个Mapper从DistributedCache读取HashTableFiles到内存中，顺序扫描大表，在Map阶段直接进行Join，将数据传递给下一个MapReduce任务。

　　JOIN[8]中有一张表为临时表，先对Stage-2进行深度拷贝（由于需要保留原始执行计划为BackupPlan，所以这里将执行计划拷贝了一份），生成一个MapJoinOperator替代JoinOperator，然后生成一个MapReduceLocalWork读取小表生成HashTableFiles上传至DistributedCache中。

推荐：