全面总结Hive性能优化（一）

朱雀 2023-02-27 12:23 29阅读 0赞

对于hive来说，相信大家都明白，  
Hive是基于Hadoop的一个数据仓库工具；它是MapReduce的一个封装，底层就是MapReduce程序；  
Hive可以将结构化的数据文件（按照各字段分类的数据）映射成一张虚表，并提供类SQL查询功能。

由于Hive的执行依赖于底层的MapReduce作业,因此对Hadoop作业的优化或者对MapReduce作业的调整是提高Hive性能的基础。  
所以我们可以通过一系列的调优方法来使其提高效率。

#### 为什么Hive运行性能低下？ ####

简单来说hive用查询语句查询数据时，会花费大量的时间，尤其是数据量大的时候，低效的查询语句查询不仅耗时长，还会浪费运行资源，有可能查询的时候没有限制分区，导致任务起了好几万个map，导致造成太多的小文件，用过hive的朋友，我想或多或少都有类似的经历：一天下来，没跑几次hive，就到下班时间了。hive在极大数据或者数据不平衡等情况下，经常表现一般，所以很多时候必须进行优化来提高效率。

**Hive的性能优化可以从以下几方面入手**

首先我们需要知道，在Hadoop分布式计算框架下会衍生哪些问题：

**1.数据倾斜**

所谓的「数据倾斜」，用通俗的话来讲，就是“一个人累死，其他人闲死”，在
      MapReduce程序中，大量的相同key被partition分配到一个分区里，
      使这个节点承受着巨大的压力，而其他节点计算完毕后要一直等待这个忙碌的节点，
      这样一来也拖累了整体的计算时间，使数据的生产效率十分低下。

**2.job数比较多的作业运行效率相对比较低**

原因是mapreduce作业初始化的时间是比较长的。
     即使只有几百行的表，如果进行多次关联多次汇总，产生十几个jobs，同样会耗时很长。

**3.分组太少，不必要的分组，分配reduce不均造成计算压力大**

count(distinct)在数据量大的情况下，效率较低，如果是多count(distinct...)效率更低，
     因为count(distinct)是按group by字段分组，按distinct字段排序，一般这种分布方式是很倾斜的；
         假设需要计算男女的UV，像淘宝一天30亿的PV，
     如果按性别分组，分配2个reduce，每个reduce要处理15亿数据。

**4.数据倾斜是导致效率大幅降低的主要原因**

可以采用多一次 Map/Reduce 的方法， 避免倾斜。

清楚了出现效率低下的主要原因，那么去优化的话也就知道从哪入手了。

## 调优方案 ##

#### 1.减少处理数据量 ####

**列裁剪**

Hive 在读数据的时候，可以只读取查询中所需要用到的列，而忽略其它列.

SELECT a,b FROM z WHERE e<10;

在实施此项查询中，z表有 5 列（a，b，c，d，e），Hive 只读取查询逻辑中真实需要 的 3 列 a、b、e，  
而忽略列 c，d；这样做节省了读取开销，中间表存储开销和数据整合开销。

**分区裁剪**

可以在查询的过程中减少不必要的分区

SELECT * FROM (SELECTT a1,COUNT(1) FROM T GROUP BY a1) subq
    WHERE subq.prtn=100; 
    
    #  多余分区
    SELECT * FROM T1 JOIN (SELECT * FROM T2) subq 
    ON (T1.a1=subq.a2) WHERE subq.prtn=100;

查询语句若将“subq.prtn=100”条件放入子查询中更为高效，可以减少读入的分区 数目.  
Hive 自动执行这种裁剪优化

**分区参数为：hive.optimize.pruner=true**

#### 2.Join原则 ####

在编写带有 join 操作的代码语句时，应该将条目少的表/子查询放在 Join 操作符的左边。  
因为在 Reduce 阶段，位于 Join 操作符左边的表的内容会被加载进内存，载入条目较少的表 可以有效减少 OOM（out of memory）即内存溢出。所以对于同一个 key 来说，对应的 value 值小的放前，大的放后，这便是“小表放前”原则。若一条语句中有多个 Join，依据 Join 的条件相同与否，有不同的处理方法。

如果 Join 的条件相同：  
1.不管有多少个表，都会则会合并为一个 Map-Reduce  
2.产生一个 Map-Reduce 任务，而不是 ‘n’ 个  
3.在做 OUTER JOIN 的时候也是一样

如果 Join 的条件不相同：  
Map-Reduce 的任务数目和 Join 操作的数目是对应的

#### 3.Join阶段数据倾斜的优化 ####

可按照表的大小和join方式的不同对应有多种优化：

**map join**

map join可以把小表全部读入到内存，然后发送到大表所在的服务器处理，
    所有的逻辑都在map端完成，不需要reduce端，从而避免倾斜。

**适用场景**：  
大表 inner join 小表  
大表 left join 小表  
小表 right join 大表  
**注意：map join有以下限制**：  
hive小表默认大小不能超过25M，可以通过 **hive.mapjoin.smalltable.filesize** 进行调整  
hive 0.7版本之后，可以通过设置 **set hive.auto.convert.join = true** 自动优化  
相关的参数为：

*  **hive.join.emit.interval = 1000**
 *  **hive.mapjoin.size.key = 10000**
 *  **hive.mapjoin.cache.numrows = 10000**

在Hive0.11后，Hive默认启动该优化，也就是不在需要显式的使用mapjoin标记，  
其会在必要的时候触发该优化操作将普通join转换成mapjoin。  
可以通过以下两个属性来设置该优化的触发时机

*  **set hive.auto.convert.join = true**
 *  **set hive.mapjoin.smalltable.filesize = 25000000 (默认 25M)**

**map join + left join**

主要是为了解决小表在left outer join左侧从而无法直接使用map join的问题。
    通过”小表 left join(小表 map join 大表)”这样的方式来实现优化。

适用场景:  
小表left outer join大表  
大表right outer join小表

**先group by 再map join 大表关联大表**

先把某个大表按照关联key做group by，来减少关联时的数据量，
    然后把group by的结果当作map join的小表再做关联。

适用场景：  
大表a left join/inner join大表b，其中大表b的join key的distinct值数量比较少

**先过滤再关联 大表有空值**

大表a left join大表b，大表b的关联key有关联不上的数据导致倾斜，
    把关联后不需要的数据在关联前先过滤掉。

适用场景：  
大表b的join key有很多空值情况，可以先将大表b的空值先做过滤在关联

**关联key随机化处理**

大表a left join大表b，小表数据量超过map join阈值，大表a的关联key有关联不上的数据导致倾斜。
    把key值关联不上，但是需要保留的那部分数据的关联key随机化处理之后再去关联，
    这样就可以避免相同key数据量太大导致数据倾斜。

适用场景：  
大表a的关联key有很多空值

**切分处理后union all**

大表a join大表b，把相同key记录数比较多的和相同key记录数比较少的记录分开来单独处理，
    然后再把结果union all 起来

适用场景：  
大表关联大表，且其它方式无法解决数据倾斜的问题的时候

**count distinct大量相同特殊值**

count distinct时，将值为空的情况单独处理，如果是计算count distinct，可以不用处理，直接过滤，在最后结果中加1。  
如果还有其他计算，需要进行group by，可以先将值为空的记录单独处理，再和其他计算结果进行union。

**使用相同的连接键**

当对3个或者更多个表进行join连接时，如果每个on子句都使用相同的连接键的话，那么只会产生一个MapReduce job。

**设置合理的reduce数**

MapReduce 程序中，reducer 个数的设定极大影响执行效率，这使得 Hive 怎样决定 reducer 个数成为一个关键问题

**mapreduce.job.reduces=10**（设置一个常量 reducetask 数量）  
**hive.exec.reducers.bytes.per.reducer=67108864（默认为 256000000）**  
**hive.exec.reducers.max=1009（默认为 1009）**

**尽量原子化操作**

尽量避免一个SQL包含复杂逻辑，可以使用中间表来完成复杂的逻辑

#### 4.本地模式运行 ####

对于小数据集，为查询触发执行任务消耗的时间>实际执行job的时间，因此可以通过本地模式，在单台机器上（或某些时候在单个进程上）处理所有的任务。

可以通过设置属性**hive.exec.mode.local.auto**的值为true，来让hve在适当的时候自动启动这个优化，也可以将这个配置写在$HOME/.hiverc文件中。

当一个job满足如下条件才能真正使用本地模式：

*  **job的输入数据大小必须小于参数：hive.exec.mode.local.auto.inputbytes.max(默认128MB)**
 *  **job的map数必须小于参数：hive.exec.mode.local.auto.tasks.max(默认4)**
 *  **job的reduce数必须为0或者1**
 *  **可用参数hive.mapred.local.mem(默认0)控制child jvm使用的最大内存数**。

#### 5.并行执行 ####

hive会将一个查询转化为一个或多个阶段，包括：MapReduce阶段、抽样阶段、合并阶段、limit阶段等。默认情况下，一次只执行一个阶段。 不过，如果某些阶段不是互相依赖，是可以并行执行的。

**set hive.exec.parallel=true**,可以开启并行执行，默认是flase。  
**set hive.exec.parallel.thread.number=16**; 同一个sql允许最大并行度，默认为8。  
会比较耗系统资源。

#### 6.严格模式 ####

**set hive.marped.mode=strict** ；防止用户执行那些可能意想不到的不好的影响的查询

*  分区表，必须选定分区范围。
 *  对于使用order by的查询，要求必须使用limit语句。因为order by为了执行排序过程会将所有的结果数据分发到同一个reducer中进行处理。
 *  限制笛卡尔积查询：两张表join时必须有on语句。

#### 7.GROUP BY优化 ####

*  Map端部分聚合  
    事实上并不是所有的聚合操作都需要在reduce部分进行，很多聚合操作都可以先在Map端进行部分聚合，然后reduce端得出最终结果。  
    设定参数为：  
    **hive.map.aggr=true**（用于设定是否在 map 端进行聚合，默认值为真）  
    **hive.groupby.mapaggr.checkinterval=100000**（用于设定 map 端进行聚合操作的条目数）
 *  有数据倾斜时进行负载均衡  
    设定参数为：  
    **hive.groupby.skewindata=true**  
    当选项设定为 true 时，生成的查询计划有两 个 MapReduce 任务。  
    在第一个 MapReduce 中，map 的输出结果集合会随机分布到 reduce 中， 每个 reduce 做部分聚合操作，并输出结果。  
    这样处理的结果是，相同的 Group By Key 有可 能分发到不同的 reduce 中，从而达到负载均衡的目的；  
    第二个 MapReduce 任务再根据预处理的数据结果按照 Group By Key 分布到 reduce 中（这个过程可以保证相同的 Group By Key 分布到同一个 reduce 中），最后完成最终的聚合操作。

#### 8.JVM重用： 解决小文件和task特别多的场景 ####

修改hadoop的mapred-site.xml文件进行设置

<property>
         <name>mapred.job.reuse.jvm.num.tasks</name>
         <value>10</value>
     </property>

使得jvm实例在同一个job中重新使用N次，减少JVM启动造成的开销。  
通过JVM重用减少JVM启动的消耗，默认每个Map或Reduce启动-个新的JVM。

Map或Reduce运行时间很短时，JVM启动过程占很大开销：  
通过共享JVM来重用JVM， 以串行方式运行MapReduce Job  
适用于同一个Job中的Map或Reduce任务  
对于不同Job的任务，总是在独立的JVM中运行

通过以下设置开启JVM重用：  
**set mapred.job.reuse.jvm.num.tasks= 5; ----默认值为1**

#### 9.优化的常用手段 ####

**主要由三个属性来决定：**

*  **hive.exec.reducers.bytes.per.reducer**  
    这个参数控制一个job会有多少个reducer来处理，依据的是输入文件的总大小。默认1GB。
 *  **hive.exec.reducers.max**  
    这个参数控制最大的reducer的数量，如果 input / bytes per reduce > max 则会启动这个参数所指定的reduce个数。  
    这个并不会影响mapre.reduce.tasks参数的设置。默认的max是999。
 *  **mapred.reduce.tasks**  
    这个参数如果指定了，hive就不会用它的estimation函数来自动计算reduce的个数，而是用这个参数来启动reducer。默认是1。

**Hive的基本优化就学习到这里，在下一篇文章中我们将学习如何解决数据进行计算时的数据倾斜**