Hive原理及SQL优化

Posted 2021-10-08 东海陈光剑

1.Hive原理

Hive是构建在Hadoop上的数据仓库软件框架，支持使用SQL来读，写和管理大规模数据集合。Hive入门非常简单，功能非常强大，所以非常流行。

通常来说，Hive只支持数据查询和加载，但后面的版本也支持了插入，更新和删除以及流式api。Ｈive具有目前Hadoop上最丰富最全的SQL语法，也拥有最慢最稳定的执行。是目前Hadoop上几乎标准的ETL和数据仓库工具。

Hive这个特点与其它AdHoc查询工具如Impala(拉式获取数据\\利用内存\\渐进输出结果)，Spark SQL或者Presto有着应用场景的区别，也就是虽然都是即席查询工具，前者适用与稳定作业执行，调度以及ETL，或者更倾向于交户式。一个典型的场景是分析师使用Impala去探测数据，验证想法，并把数据产品部署在Hive上执行。

1.1Hadoop基本原理

在我们讲Hive原理和查询优化前，让我们先回顾一下Hadoop基本原理。

Hadoop是一个分布式系统，有HDFS和Yarn。HDFS用于执行存储，Yarn用于资源调度和计算。ＭapReduce是跑在Yarn上的一种计算作业，此外还有Spark等。

Hive通常意义上来说，是把一个SQL转化成一个分布式作业，如MapReduce，Spark或者Tez。无论Hive的底层执行框架是MapReduce、Spark还是Tez，其原理基本都类似。

而目前，由于MapReduce稳定，容错性好，大量数据情况下使用磁盘，能处理的数据量大，所以目前Hive的主流执行框架是MapReduce，但性能相比Spark和Tez也就较低，等下讲到Group By和JOIN原理时会解释这方面的原因。

目前的Hive除了支持在MapReduce上执行，还支持在Spark和Tez 上执行。我们以MapReduce为例来说明的Hive的原理。先回顾一下 MapReduce 原理。

两个Mapper各自输入一块数据，由键值对构成，对它进行加工（加上了个字符n），然后按加工后的数据的键进行分组，相同的键到相同的机器。这样的话，第一台机器分到了键nk1和nk3，第二台机器分到了键nk2。

接下来再在这些Reducers上执行聚合操作（这里执行的是是count），输出就是nk1出现了2次，nk3出现了1次，nk2出现了3次。从全局上来看，MapReduce就是一个分布式的GroupBy的过程。

从上图可以看到，Global Shuffle左边，两台机器执行的是Map。Global Shuffle右边，两台机器执行的是Reduce。所以Hive，实际上就是一个编译器，一个翻译机。把SQL翻译成MapReduce之类的作业。

1.2Hive架构

下面这个旧一点的图片来自Facebook

从架构图上可以很清楚地看出Hive和Hadoop（MapReduce，HDFS）的关系。

Hive是最上层，即客户端层或者作业提交层。 

MapReduce/Yarn是中间层，也就是计算层。 

HDFS是底层，也就是存储层。

从Facebook的图上可以看出，Hive主要有QL，MetaStore和Serde三大核心组件构成。QL就是编译器，也是Hive中最核心的部分。Serde就是Serializer和Deserializer的缩写，用于序列化和反序列化数据，即读写数据。MetaStore对外暴露Thrift API，用于元数据的修改。比如表的增删改查，分区的增删改查，表的属性的修改，分区的属性的修改等。

1.3Hive的数据模型

Hive的数据存储在HDFS上，基本存储单位是表或者分区，Hive内部把表或者分区称作SD，即Storage Descriptor。一个SD通常是一个HDFS路径，或者其它文件系统路径。SD的元数据信息存储在Hive MetaStore中，如文件路径，文件格式，列，数据类型，分隔符。Hive默认的分格符有三种，分别是^A、^B和^C，即ASCii码的1、2和3，分别用于分隔列，分隔列中的数组元素，和元素Key-Value对中的Key和Value。

还记得大明湖畔暴露Thrift API的MetaStore么？嗯，是她，就是它！所有的数据能不能认得出来全靠它！

Hive的核心是Driver，Driver的核心是SemanticAnalyzer。 Hive实际上是一个SQL到Hadoop作业的编译器。 Hadoop上最流行的作业就是MapReduce，当然还有其它，比如Tez和Spark。Hive目前支持ＭapReduce, Tez, Spark三种作业，其原理和刚才回顾的MapReduce过程类似，只是在执行优化上有区别。

1.4Hive的编译流程

Hive作业的执行过程实际上是SQL翻译成作业的过程？那么，它是怎么翻译的？

一条SQL，进入的Hive。经过上述的过程，其实也是一个比较典型的编译过程变成了一个作业。

首先，Driver会输入一个字符串SQL，然后经过Parser变成AST（abstract syntax tree 抽象语法树），这个变成AST的过程是通过Antlr（antlr是指可以根据输入自动生成语法树并可视化的显示出来的开源语法分析器）来完成的，也就是Anltr根据语法文件来将SQL变成AST。

AST进入SemanticAnalyzer（核心）变成QB，也就是所谓的QueryBlock。一个最简的查询块，通常来讲，一个From子句会生成一个QB。生成QB是一个递归过程，生成的 ＱＢ经过GenLogicalPlan过程，变成了一个Operator图，也是一个有向无环图。

OP DAG经过逻辑优化器，对这个图上的边或者结点进行调整，顺序修订，变成了一个优化后的有向无环图。这些优化过程可能包括谓词下推（Predicate Push Down），分区剪裁（Partition Prunner），关联排序（Join Reorder）等等

经过了逻辑优化，这个有向无环图还要能够执行。所以有了生成物理执行计划的过程。GenTasks。Ｈive的作法通常是碰到需要分发的地方，切上一刀，生成一道MapReduce作业。如Group By切一刀，Join切一刀，Distribute By切一刀，Distinct切一刀。

这么很多刀砍下去之后，刚才那个逻辑执行计划，也就是那个逻辑有向无环图，就被切成了很多个子图，每个子图构成一个结点。这些结点又连成了一个执行计划图，也就是Task Tree.

把这些个Task Tree 还可以有一些优化，比如基于输入选择执行路径，增加备份作业等。进行调整。这个优化就是由Physical Optimizer来完成的。经过Physical Optimizer，这每一个结点就是一个MapReduce作业或者本地作业，就可以执行了。

这就是一个SQL如何变成MapReduce作业的过程。要想观查这个过程的最终结果，可以打开Hive，输入Explain ＋ 语句，就能够看到。

1.5group by的执行任务

Hive最重要的部分是Group By和Join。下面分别讲解一下：

首先是Group By

例如我们有一条SQL语句：

INSERT INTO TABLE pageid_age_sum

SELECT pageid, age, count(1)

FROM pv_users

GROUP BY pageid, age;

把每个网页的阅读数按年龄进行分组统计。由于前面介绍了，MapReduce就是一个Group By的过程，这个SQL翻译成MapReduce就是相对简单的。

我们在Map端，每一个Map读取一部分表的数据，通常是64M或者256M，然后按需要Group By的Key分发到Reduce端。经过Shuffle Sort，每一个Key再在Reduce端进行聚合（这里是Count)，然后就输出了最终的结果。

1.6 distinct的执行任务

值得一提的是，Distinct在实现原理上与Group By类似。当Group By遇上 Distinct……例如：SELECT pageid, COUNT(DISTINCT userid) FROM page_view GROUP BY pageid

Hive 实现成MapReduce的原理如下：

也就是说Map分发到Reduce的时候，会使用pageid和userid作为联合分发键，再去聚合（Count），输出结果。

介绍了这么多原理，重点还是为了使用，为了适应场景和业务，为了优化。从原理上可以看出，当遇到Group By的查询时，会按Group By 键进行分发？如果键很多，撑爆了机器会怎么样？

对于Impala，或Spark，为了快，key在内存中，爆是经常的。爆了就失败了。对于Hive，Key在硬盘，本身就比Impala, Spark的处理能力大上几万倍。但……不幸的是，硬盘也有可能爆。

1.7 join的执行任务

例如这样一个查询：INSERT INTO TABLE pv_users

SELECT pv.pageid, u.age

FROM page_view pv JOIN user u ON (pv.userid = u.userid);

把访问和用户表进行关联，生成访问用户表。Hive的Join也是通过MapReduce来完成的。

就上面的查询，在ＭapReduce的Join的实现过程如下：

Ｍap端会分别读入各个表的一部分数据，把这部分数据进行打标，例如pv表标1，user表标2.

Map读取是分布式进行的。标完完后分发到Reduce端，Reduce 端根据Join Key，也就是关联键进行分组。然后按打的标进行排序，也就是图上的Shuffle Sort。

在每一个Reduce分组中，Key为111的在一起，也就是一台机器上。同时，pv表的数据在这台机器的上端，user表的数据在这台机器的下端。

这时候，Reduce把pv表的数据读入到内存里，然后逐条与硬盘上user表的数据做Join就可以了。

从这个实现可以看出，我们在写Hive Join的时候，应该尽可能把小表（分布均匀的表）写在左边，大表（或倾斜表）写在右边。这样可以有效利用内存和硬盘的关系，增强Hive的处理能力。

同时由于使用Join Key进行分发， Hive也只支持等值Join，不支持非等值Join。由于Join和Group By一样存在分发，所以也同样存在着倾斜的问题。所以Join也要对抗倾斜数据，提升查询执行性能。

1.8 Map join的执行任务

通常，有一种执行非常快的Join叫Map Join 。

手动的Map Join SQL如下（pv是小表）：

INSERT INTO TABLE pv_users

SELECT /*+ MAPJOIN(pv) */ pv.pageid, u.age

FROM page_view pv JOIN user u

ON (pv.userid = u.userid);

还是刚才的例子，用Map Join执行

Map Join通常只适用于一个大表和一个小表做关联的场景，例如事实表和维表的关联。

原理如上图，用户可以手动指定哪个表是小表，然后在客户端把小表打成一个哈希表序列化文件的压缩包，通过分布式缓存均匀分发到作业执行的每一个结点上。然后在结点上进行解压，在内存中完成关联。

Map Join全过程不会使用Reduce，非常均匀，不会存在数据倾斜问题。默认情况下，小表不应该超过25M。在实际使用过程中，手动判断是不是应该用Map Join太麻烦了，而且小表可能来自于子查询的结果。

Hive有一种稍微复杂一点的机制，叫Auto Map Join

还记得原理中提到的物理优化器？Physical Optimizer么？它的其中一个功能就是把Join优化成Auto Map Join

图上左边是优化前的，右边是优化后的

优化过程是把Join作业前面加上一个条件选择器ConditionalTask和一个分支。左边的分支是MapJoin，右边的分支是Common Join(Reduce Join)

看看左边的分支是不是和我们上上一张图很像？

这个时候，我们在执行的时候，就由这个Conditional Task 进行实时路径选择，遇到小于25兆走左边，大于25兆走右边。所谓，男的走左边，女的走右边，人妖走中间。

在比较新版的Hive中，Auto Mapjoin是默认开启的。如果没有开启，可以使用一个开关， set hive.auto.convert.join=true 开启。

当然，Join也会遇到和上面的Group By一样的倾斜问题。

Ｈive 也可以通过像Group By一样两道作业的模式单独处理一行或者多行倾斜的数据。

hive 中设定

set hive.optimize.skewjoin = true;

set hive.skewjoin.key = skew_key_threshold （default = 100000）

其原理是就在Reduce Join过程，把超过十万条的倾斜键的行写到文件里，回头再起一道Join单行的Map Join作业来单独收拾它们。最后把结果取并集就是了。如上图所示。

1.9Hive适合做什么？

由于多年积累，Hive比较稳定，几乎是Hadoop上事实的SQL标准。 Ｈive适合离线ETL，适合大数据离线Ad-Hoc查询。适合特大规模数据集合需要精确结果的查询。对于交互式Ad-Hoc查询，通常还会有别的解决方案，比如Impala, Presto等等。

特大规模的离线数据处理，尤其是大表关联，特大规模数据聚集，很适合使用Hive。讲了这么多原理，最重要的还是应用，还是创造价值。

对Hive来说，数据量再大，都不怕。数据倾斜，是大难题。但有很多优化方法和业务改进方法可以避过。Hive执行稳定，函数多，扩展性强，数据吞吐量大，了解原理，有助于用好和选型。

2.SQL优化

HIVE优化可以从两方面入手，减少计算和加速计算。

减少计算包括：减少数据量、数据复用，避免重复计算、多粒度逐步聚合和业务裁剪。

加速计算包括：使用内存计算、并发计算。

2.1 limit优化

limit不启用优化的情况

a.语句中带有join，group by，列as别名，order by，where等的，不启用优化。

b.子查询中的limit不会做优化，会扫描所有数据，扫描完了以后取limit指定的条数。

c. 查询视图 limit不会启用优化

 对于a这种情况，都会用MR处理，并且优化会比较复杂，因为group by，order by这样的，必须扫描所有数据，要保证结果正确又尽量少计算，会让查询优化器变得非常复杂。对于b这种情况，与a类似，都要走MR，另外一方面，子查询中带limit本身用法比较少见，所以也没有优化。

推荐使用:

( 这些语句会直接读取HDFS文件，不走MR，会很快)

select * from taba  limit 1;

 select a from taba partition(p) limit 1;

2.2使用分区/列修剪

      做好列裁剪和filter操作，尤其是只读取需要的分区。支持分区自动修剪，因此，一般情况下，只要分区字段在where子句中，TDW就会自动过滤掉不需要的分区。但是，如果分区字段位于in以及其他函数中，那么分区自动修剪将失效。另外，between、or、and都支持分区自动修剪。

推荐使用:

(tdbank_imp_date 为表的分区字段,对分区字段不要使用函数，不要用substr)： 

where tdbank_imp_date in('2018030100','2018030101','2018030102') 

where tdbank_imp_date='2018030100'

where tdbank_imp_date between  '2018030100'  and '2018043023'

2.3禁止出现笛卡尔积

笛卡尔积只有1个reduce任务(一个整的大文件)，会导致计算超慢，甚至可能计算不出来或者导致节点挂掉。

以下3种形式的SQL会导致笛卡尔积：

select * from gbk，utf8 where gbk.key= utf8.key and gbk.key > 10;

select * from gbk join utf8 where gbk.key= utf8.key and gbk.key > 10;

 tablea join tableb join tablec join ... on tablea.col1 = tableb.col2 and ...

推荐使用:

(1) select * from gbk join utf8 on gbk.key= utf8.key where gbk.key > 10;

(2) tablea join tableb on （tablea.col1 = tableb.col2 and ... ） join tablec on ...join ... on ...

(3) select * from 

(select * from gbk where gbk.key>10) gbk 

join 

(select * from utf8)

on gbk.key=utf8.key 

2.4表关联的优化

2.4.1有小表且数据条数不超过2w行

 推荐使用Map Join

map join的必要条件：

       a.参与连接的小表的行数，以不超过2万条为宜。

       b.连接类型是inner join、right outer join（小表不能是右表）、left outer join（小表不能是左表）、left semi join。

使用方法示例：

例如(其中pv是小表，会把pv表生成hash表，压缩)：

      SELECT /*+ MAPJOIN(pv) */   

                    pv.pageid, u.age                                 

      FROM page_view pv

                 JOIN user u

                 ON (pv.userid = u.userid);

注：当大表存在数据倾斜时，如果小表符合map join的要求，使用map join会极大加速计算。

2.4.2小表连接大表

    将较大的表放在join操作符的右边，这样生成的查询计划效率较高，执行速度快，不容易出错；

   在Join 操作的Reduce 阶段，位于Join 操作符左边的表的内容会首先被加载进内存（容器满后存入硬盘），然后对右侧表进行流式处理，将条目少的表放在左边，可以有效减少发生磁盘IO和OOM 错误的几率。

推荐使用：

将较大的表放在join操作符的右边  

2.5数据倾斜的优化

     常见的数据倾斜问题，一般发生在group by 或者join操作上，表现为一个或几个reduce一直没办法做完，原因是key分布不均，某个或某几个key的数据特别大。可以对key的数据量排序来检验是否有数据倾斜问题：

SELECT  key1,key2, count(1) as cnt

FROM test_tatble

GROUP BY key1,key2

ORDER BY cnt DESC LIMIT 50

2.5.1有小表使用MAPJOIN

当需要join的表有一个小表时就很适合用内存计算来完成，也即使用MAPJOIN。

2.5.2 去除数据倾斜数据，数据为脏数据

特别留意关联的key 里面有大量0、空值；做过滤 ，然后再distinct输出

2.5.3单独计算导致数据倾斜的key再合并数据

根据key值拆开成2个集合，然后再union all 起来 

2.5.4 groupby时的数据倾斜，参数设置

set hive.groupby.skewindata=true;

{:problematic sql}

set hive.groupby.skewindata=false;

查询计划会生成两个MR Job，其中在第一个MR Job中，Map的输出结果会随机分配到Reduce端，每个Reduce做部分聚合计算，然后输出结果，从而达到负载均衡的目的；在第二MR Job中，根据预处理的数据结果按照分组 Key 将数据分配到相应的Reduce任务中，完成最终的聚合计算。

2.5.5设置reduce数，使得reducekey分散

通过设置reduce任务数提高并行度来加速执行：

set mapred.reduce.tasks=N; //执行语句之前

set mapred.reduce.tasks=-1; //执行语句之后恢复原状

注意：请合理设置N的大小，最好设置为上述参数的大小。一定不要超过999！

2.5.6 将数据倾斜的key随机化

将空值key转换成一个字符串加上随机数，从而将倾斜的数据分到不同的Reduce任务上

select*fromtable1 a left join table2 b on case when a.vopenid is null then concat('random',rand()) else a.vopenid end = b.vopenid;

2.6先聚合后连接

这个原则很简单，因为join key可能存在倾斜，因此，只要可能，最好先对join  key进行处理一下再进行join操作，避免数据倾斜。

一般来说，join后面会跟着一些聚合函数操作，这个原则是尽量将聚合操作提前做，使得在做join的时候join key可以是单一的。

聚合有数据压缩的作用。“先聚合，后连接”可以减少聚合和连接时的数据量

2.8开启并行执行

   支持并行执行机制，以下功能都可以通过并行执行加速：union all、join（参与join的是复杂查询）、cube、rollup、groupingsets、with等。可见，很多常用的功能都可以加速。默认该并行开关是关闭的，若需要并行执行，可以设置 set hive.exec.parallel=true;打开开关，但消耗资源成本会增加。

2.9任务分析常用语句

（1）想知道分区修剪是否起作用么？

答案：用explain语句吧，看看要读取哪些目录就知道了。另外，通过这个语句的执行结果，还可以检查你的查询计划是否合理。

例句：

Explain select t.col1,t.col3 from dbname::tablename t where t.ftime=‘20130104’

（2）show rowcount的作用

Show rowcount dbname::tablename //显示整个表有多少行

通过这个命令可以知道一个表有多少行记录，有了它的帮助就可以在连接时基本保证小表连接大表，也可以知道是否适合使用map join。

Show rowcount extended dbname::tablename  //按分区显示每个分区有多少行

这个命令更有用的形式是show rowcount extended tablename，这个命令可以按分区来显示每个分区有多少行记录。

注意：本命令和下一条命令只对结构化存储文件生效。如对文本文件执行该命令将报错。

（3）show tablesize的作用

Show tablesize dbname::tablename

通过这个命令可以知道一个表有多大，单位是字节。

扩展形式：

Show tablesize extended dbname::tablename

说明：

通过该命令的结果，可以估算出大约需要多少个map任务，现在一般256MB/512MB一个map任务。如果你想知道join的时候需要多少个map任务，只需要把每个表需要的map任务数求和就可以了。

需要的map数太多可不是好事，通常map数超过1万就是较大的任务了。否则，你需要耐心等待。

2.10临时表的应用

多次需要用到的数据，最好使用临时表保存起来

总结：

用集合的思维分解问题 ， 用同类型key做关联；

能少join就尽量少join，想办法实现数据复用 ； 

尽最大能力限制子查询输入/输出的数据量 ；

尽量避免udf函数的使用，尽量利用内存机制；

遇到数据倾斜，先清除脏数据，再做优化；

3.常见sql的标准写法

(1) 日志数据提取

需要注意分区/列修剪

select filed1,filed2,filed3 from tableA where tdbank_imp_date between '2018092700' and '2018092723' and ....

(2) 去重统计

先distinct 减少输出的数据量

select count(*) from (

select distinct  openid from  tablelog where tdbank_imp_date between '2018092700' and '2018092723' and ....

)

(3) 查询最近一个月既玩了A游戏也玩了B游戏的用户数量

能一个查询搞定的 尽量一个查询搞定

select count(suin) from iplat group by suin having (count(case when vgameappid='xxx'  then 1 end )>0 and count(case when vgameappid='yyy' then 1 end)>0);

参考资料：

https://blog.csdn.net/LW_GHY/article/details/51469753?utm_source=copy

http://km.oa.com/group/2430/articles/show/127863