你好赵伟

这个屌丝很懒，什么也没留下！

热门标签

hive 优化_hive.exec.reducers.bytes.per.reducer

作者：你好赵伟 | 2024-06-16 19:57:48

踩

hive.exec.reducers.bytes.per.reducer

Hive官方：
https://cwiki.apache.org/confluence/display/Hive/Home

1. 通常情况下，作业会通过input的目录产生一个或者多个map任务。
主要的决定因素有： input的文件总个数，input的文件大小，集群设置的文件块大小(目前为128M, 可在hive中通过set dfs.block.size;命令查看到，该参数不能自定义修改)；

1.1、Map数
Map数过大
Map阶段输出文件太大，产生大量小文件
初始化和创建Map的开销很大
Map数太小
文件处理或者查询并发度小，Job执行时间过长
大量作业时，容易堵塞集群
通常情况下，作业会通过input的目录产生一个或者多个map数
主要的决定因素有：input的文件数，input的文件大小
举例：
a) 假设input目录下有1个文件a,大小为780M,那么hadoop会将该文件a分隔成7个块（6个128m的块和1个12m的块），从而产生7个map数
b) 假设input目录下有3个文件a,b,c,大小分别为10m，20m，130m，那么hadoop会分隔成4个块（10m,20m,128m,2m）,从而产生4个map数
即，如果文件大于块大小(128m),那么会拆分，如果小于块大小，则把该文件当成一个块。
需要采取两种方式来解决：即减少map数和增加map数；
减少map数可以通过合并小文件来实现，这点是对文件源
增加map数的可以通过控制上一个job的reducer来控制（一个sql中join多个表会分解为多个mapreduce），join过程分解为至少两个mapreduce，第二个map数少，是因为第一个mapreduce的reduce数量太少造成的。

小文件进行合并：

Map阶段Hive自动（0.7版本之后）对小文件合并

对应参数和默认值：

set hive.merge.mapfiles=true #在Map-only的任务结束时合并小文件

set hive.merge.mapredfiles=true #默认是false，true时在Map-Reduce的任务结束时合并小文件

set hive.merge.size.per.task=256*1000*1000 #合并文件的大小

set mapred.max.split.size=256000000; #每个map最大分隔大小，即每个mapreduce块儿的大小

set mapred.min.split.size.per.node=100000000;#一个节点上split的最小值

set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;#执行Map前进行小文件合并

set的作用域是session级

在开启了 org.apache.hadoop.hive.ql.io.CombineHiveInputFormat后，一个data node节点上多个小文件会进行合并，合并文件数由mapred.max.split.size限制的大小决定。

mapred.min.split.size.per.node决定了多个data node上的文件是否需要合并（即，如果在多个节点上的块儿如果大小比此值小，那么将会合并）

1.1.2、参数：mapred.map.tasks

Hive中set mapred.map.tasks=100;当文件小于

案例：

dfs.block.size=256000000

hive.merge.mapredfiles=true(默认是false，可以在hive-site.xml里配置)

hive.merge.mapfiles=true

hive.merge.size.per.task=256000000

mapred.map.tasks=2

因为合并小文件默认为true，而dfs.block.size与hive.merge.size.per.task的搭配使得合并后的绝大部分文件都在256MB左右。

Case1：

现在我们假设有3个300MB大小的文件，整个JOB会有6个map，其中3个map分别处理256MB的数据，还有3个map分别处理44MB的数据。

木桶效应就来了，整个JOB的map阶段的执行时间不是看最短的1个map的执行时间，而是看最长的1个map的执行时间。所以，虽然有3个map分别只处理44MB的数据，可以很快跑完，但它们还是要等待另外3个处理256MB的map。显然，处理256MB的3个map拖了整个JOB的后腿。

Case2:

如果我们把mapred.map.tasks设置成6，再来看下有什么变化：

goalsize=min(900MB/6,256MB)=150MB

整个JOB同样会分配6个map来处理，每个map处理150MB的数据，非常均匀，谁都不会过后腿，最合理的分配了资源，执行时间大约是CASE1的59%（160/256)

1.2、reduce数

reduce数过大

生成了很多小文件（最终输出文件有reduce决定，一个reduce一个文件），那么如果这些小文件作为下一个Job输入，则会出现小文件过多需要进行合并的问题。而且创建大的reduce也需要耗费大量不必要的资源( 启动和初始化reduce也会消耗时间和资源；有多少个reduce就会有多少个输出文件 )

Reduce数太低

执行耗时

可能出现数据倾斜

Reduce数的确定：

默认下，Hive分配reducer个数基于以下：

参数1：hive.exec.reducers.bytes.per.reducer（默认是1G)

参数2：hive.exec.reducers.max（默认为999）

计算reducer数的公式

N=min(参数2，总输入数据量/参数1）

即默认一个reduce处理1G数据量

什么情况下只有一个reduce：很多时候，你会发现任务中不管数据量有多大，不管你有没有设置调整reduce个数的参数，任务中一直都只有一个reduce任务：

1、除了数据量小于hive.exec.reducers.bytes.per.reducer参数值的情况外

2、没有group by的汇总

3、用来Order by

1.2.1、设置Reduce数

参数：mapred.reduce.tasks

默认：1

set mapred.reduce.tasks=10;

作用域Session级

当某个Job的结果被Job多次引用时，设置该参数，以便增大访问的Map数。

Reduce数决定中间或者落地文件数，文件大小和Block大小无关。

二、数据倾斜

1，什么是数据倾斜？Hadoop框架的特性决定了最怕数据倾斜

由于分布不均匀，造成数据大量的集中到一点，造成数据热点。

症状：map阶段快，reduce阶段非常慢；

某些map很快，某些map很慢

某些reduce很快，某些reduce奇慢

Multi-group by

.Multi-group by是Hive的一个非常好的特性，它使得Hive中利用中间结果变得非常方便

.FROM log

. insert overwrite table test1 select log.id group by log.id

. insert overwrite table test2 select log.name group by log.name

. 上述查询语句使用了Multi-group by特性连续group by了2次数据，使用不同的group by key。这一特性可以减少一次MapReduce操作。

Bucket 与 Sampling

.Bucket是指将数据以指定列的值为key进行hash，hash到指定数目的桶中。这样就可以支持高效采样了

.Sampling可以在全体数据上进行采样，这样效率自然就低，它还是要去访问所有数据。而如果一个表已经对某一列制作了bucket，就可以采样所有桶中指定序号的某个桶，这就减少了访问量。

.如下例所示就是采样了test中32个桶中的第三个桶。

.SELECT * FROM test 、、、TABLESAMPLE(BUCKET 3 OUT OF 32);

JOIN 原则

.在使用写有 Join 操作的查询语句时有一条原则：应该将条目少的表/子查询放在 Join 操作符的左边

.原因是在 Join 操作的 Reduce 阶段，位于 Join 操作符左边的表的内容会被加载进内存，将条目少的表放在左边，可以有效减少发生 OOM 错误的几率

Map Join

.Join 操作在 Map 阶段完成，不再需要Reduce，前提条件是需要的数据在 Map 的过程中可以访问到

.例如：

.INSERT OVERWRITE TABLE phone_traffic

SELECT /*+ MAPJOIN(phone_location) */ l.phone,p.location,l.traffic from phone_location p join log l on (p.phone=l.phone)

.相关的参数为：

hive.join.emit.interval = 1000 How many rows in the right-most join operand Hive should buffer before emitting the join result.

hive.mapjoin.size.key = 10000

hive.mapjoin.cache.numrows = 10000

Group By

.Map 端部分聚合

.并不是所有的聚合操作都需要在 Reduce 端完成，很多聚合操作都可以先在 Map 端进行部分聚合，最后在 Reduce 端得出最终结果

. 基于 Hash

. 参数包括：

.hive.map.aggr = true 是否在 Map 端进行聚合，默认为 True

.hive.groupby.mapaggr.checkinterval = 100000 在 Map 端进行聚合操作的条目数目

.有数据倾斜的时候进行负载均衡

.hive.groupby.skewindata = false

.当选项设定为 true，生成的查询计划会有两个 MR Job。第一个 MR Job 中，Map 的输出结果集合会随机分布到 Reduce 中，每个 Reduce 做部分聚合操作，并输出结果，这样处理的结果是相同的 Group By Key 有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的；第二个 MR Job 再根据预处理的数据结果按照 Group By Key 分布到 Reduce 中（这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce 中），最后完成最终的聚合操作。

合并小文件

.文件数目过多，会给 HDFS 带来压力，并且会影响处理效率，可以通过合并 Map 和 Reduce 的结果文件来消除这样的影响：

.hive.merge.mapfiles = true 是否和并 Map 输出文件，默认为 True

.hive.merge.mapredfiles = false 是否合并 Reduce 输出文件，默认为 False

.hive.merge.size.per.task = 256*1000*1000 合并文件的大小

Hive自己如何确定reduce数：
reduce个数的设定极大影响任务执行效率，不指定reduce个数的情况下，Hive会猜测确定一个reduce个数，基于以下两个设定：
hive.exec.reducers.bytes.per.reducer（每个reduce任务处理的数据量，默认为1000^3=1G）
hive.exec.reducers.max（每个任务最大的reduce数，默认为999）
计算reducer数的公式很简单N=min(参数2，总输入数据量/参数1)
即，如果reduce的输入（map的输出）总大小不超过1G,那么只会有一个reduce任务；
如：select pt,count(1) from popt_tbaccountcopy_mes where pt = '2012-07-04' group by pt;
            /group/p_sdo_data/p_sdo_data_etl/pt/popt_tbaccountcopy_mes/pt=2012-07-04 总大小为9G多，因此这句有10个reduce

2.    调整reduce个数方法一：
调整hive.exec.reducers.bytes.per.reducer参数的值；
set hive.exec.reducers.bytes.per.reducer=500000000; （500M）
select pt,count(1) from popt_tbaccountcopy_mes where pt = '2012-07-04' group by pt; 这次有20个reduce

3.    调整reduce个数方法二；
set mapred.reduce.tasks = 15;
select pt,count(1) from popt_tbaccountcopy_mes where pt = '2012-07-04' group by pt;这次有15个reduce

4.    reduce个数并不是越多越好；
同map一样，启动和初始化reduce也会消耗时间和资源；
另外，有多少个reduce,就会有多少个输出文件，如果生成了很多个小文件，那么如果这些小文件作为下一个任务的输入，则也会出现小文件过多的问题；

5.    什么情况下只有一个reduce；
很多时候你会发现任务中不管数据量多大，不管你有没有设置调整reduce个数的参数，任务中一直都只有一个reduce任务；其实只有一个reduce任务的情况，除了数据量小于hive.exec.reducers.bytes.per.reducer参数值的情况外，还有以下原因：
a)    没有group by的汇总，比如把select pt,count(1) from popt_tbaccountcopy_mes where pt = '2012-07-04' group by pt; 写成 select count(1) from popt_tbaccountcopy_mes where pt = '2012-07-04';
这点非常常见，希望大家尽量改写。
b)    用了Order by
c)    有笛卡尔积
通常这些情况下，除了找办法来变通和避免，我暂时没有什么好的办法，因为这些操作都是全局的，所以hadoop不得不用一个reduce去完成；

同样的，在设置reduce个数的时候也需要考虑这两个原则：使大数据量利用合适的reduce数；使单个reduce任务处理合适的数据量；

待研究：

map的数量通常是由hadoop集群的DFS块大小确定的，也就是输入文件的总块数，正常的map数量的并行规模大致是每一个Node是 10~100个，对于CPU消耗较小的作业可以设置Map数量为300个左右，但是由于hadoop的没一个任务在初始化时需要一定的时间，因此比较合理的情况是每个map执行的时间至少超过1分钟。具体的数据分片是这样的，InputFormat在默认情况下会根据hadoop集群的DFS块大小进行分片，每一个分片会由一个map任务来进行处理，当然用户还是可以通过参数mapred.min.split.size参数在作业提交客户端进行自定义设置。还有一个重要参数就是mapred.map.tasks，这个参数设置的map数量仅仅是一个提示，只有当InputFormat 决定了map任务的个数比mapred.map.tasks值小时才起作用。同样，Map任务的个数也能通过使用JobConf 的conf.setNumMapTasks(int num)方法来手动地设置。这个方法能够用来增加map任务的个数，但是不能设定任务的个数小于Hadoop系统通过分割输入数据得到的值。当然为了提高集群的并发效率，可以设置一个默认的map数量，当用户的map数量较小或者比本身自动分割的值还小时可以使用一个相对交大的默认值，从而提高整体 hadoop集群的效率。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/你好赵伟/article/detail/728033