如何在Hive中合理配置Map和Reduce任务数量以优化不同场景下的性能表现

• 通常情况下&＃xff0c;作业会通过input的目录产生一个或者多个map任务。
• 主要的决定因素有&＃xff1a; input的文件总个数&＃xff0c;input的文件大小&＃xff0c;集群设置的文件块大小(目前为128M, 可在hive中通过set dfs.block.size;命令查看到&＃xff0c;该参数不能自定义修改)&＃xff1b;
• 举例&＃xff1a;
  • a) 假设input目录下有1个文件a,大小为780M,那么hadoop会将该文件a分隔成7个块&＃xff08;6个128m的块和1个12m的块&＃xff09;&＃xff0c;从而产生7个map数。
  • b) 假设input目录下有3个文件a,b,c,大小分别为10m&＃xff0c;20m&＃xff0c;150m&＃xff0c;那么hadoop会分隔成4个块&＃xff08;10m,20m,128m,22m&＃xff09;&＃xff0c;从而产生4个map数
  • 即如果文件大于块大小(128m)&＃xff0c;那么会拆分&＃xff0c;如果小于块大小&＃xff0c;则把该文件当成一个块。

如果超过128m就一定会被拆分吗&＃xff1f;

• 不一定
• 如果是130m&＃xff0c;仍然只会产生1个map
• 允许有10%的溢出&＃xff1a;如果文件大小超过分片大小10%以内&＃xff0c;作为一个分片
  • 设计目的&＃xff1a;避免一个MapTask处理的数据太小
• 具体原因可以参考这篇文章

是不是map数越多越好&＃xff1f;

• 答案是否定的。
• 如果一个任务有很多小文件&＃xff08;远远小于块大小128m&＃xff09;&＃xff0c;则每个小文件也会被当做一个块&＃xff0c;用一个map任务来完成&＃xff0c;而一个map任务启动和初始化的时间远远大于逻辑处理的时间&＃xff0c;就会造成很大的资源浪费。而且&＃xff0c;同时可执行的map数是受限的。

那么针对这种情况&＃xff0c;如何合并小文件&＃xff0c;减少map数&＃xff1f;

• 假设一个SQL任务&＃xff1a;

Select count(1)
from xxx
where pt &＃61; ‘2020-08-11’;


• 该表的分区目录下共有194个文件&＃xff0c;其中很多是远远小于128m的小文件&＃xff0c;总大小9G&＃xff0c;正常执行会用194个map任务。
• 可以通过以下方法来在map执行前合并小文件&＃xff0c;减少map数&＃xff1a;
• 方式1&＃xff1a;输入合并&＃xff0c;即在Map前合并小文件,下面的数值需自行调整


set hive.hadoop.supports.splittable.combineinputformat&＃61;true;
-- 执行map前进行小文件合并
set hive.input.format&＃61;org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;
set mapred.min.split.size&＃61;100000000;
set mapred.max.split.size&＃61;100000000;
-- 一个节点上split的至少的大小 &＃xff0c;决定了多个data node上的文件是否需要合并
set mapred.min.split.size.per.node&＃61;50000000;
-- 一个交换机下split的至少的大小&＃xff0c;决定了多个交换机上的文件是否需要合并
set mapred.min.split.size.per.rack&＃61;50000000;


• 解释
  • 100000000表示100M
  • set hive.input.format&＃61;org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;这个参数表示执行前进行小文件合并
  • 过期的对应配置可以查询&＃xff1a;https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/DeprecatedProperties.html
  • mapred.min.split.size相当于mapreduce.input.fileinputformat.split.minsize&＃xff0c;设置哪一个都一样
  • mapred.max.split.size相当于mapreduce.input.fileinputformat.split.maxsize&＃xff0c;设置哪一个都一样
  • 所以根据求splitSize的公式&＃xff0c;return Math.max(minSize, Math.min(maxSize, blockSize));
  • 可以知道&＃xff0c;需增加mapper数&＃xff0c;则减少这些值&＃xff1b;需减少mapper数&＃xff0c;则增加这些值&＃xff0c;一般就配置这两个值就行了。
  • 一般来说
    • max.split.size >&＃61; min.split.size >&＃61; min.size.per.node >&＃61; min.size.per.node
  • 优先级
    • 当四个参数设置矛盾时&＃xff0c;系统会自动以优先级最高的参数为准&＃xff0c;进行计算
    • max.split.size <&＃61; min.split.size <&＃61; min.size.per.node <&＃61; min.size.per.node
• 方式2&＃xff1a;结果合并,下面的数值需自行调整

set hive.merge.mapfiles&＃61;true; --在Map-only的任务结束时合并小文件
set hive.merge.mapredfiles&＃61;true; --在Map-Reduce的任务结束时合并小文件
set hive.merge.size.per.task&＃61;128000000; --合并后文件的大小为128M左右
set hive.merge.smallfiles.avgsize&＃61;128000000; --当输出文件的平均大小小于128M时&＃xff0c;启动一个独立的map-reduce任务进行文件merge


• Hive在对结果文件进行合并时会执行一个额外的map-only脚本&＃xff0c;mapper的数量是文件总大小除以size.per.task参数所得的值

是不是保证每个map处理接近128m的文件块&＃xff0c;就高枕无忧了&＃xff1f;

• 答案也是不一定。
• 比如有一个127m的文件&＃xff0c;正常会用一个map去完成&＃xff0c;但这个文件只有一个或者两个小字段&＃xff0c;却有几千万的记录&＃xff0c;如果map处理的逻辑比较复杂&＃xff0c;用一个map任务去做&＃xff0c;肯定也比较耗时。

那么针对这种情况&＃xff0c;如何适当增加map数&＃xff1f;

• 当input的文件都很大&＃xff0c;任务逻辑复杂&＃xff0c;map执行非常慢的时候&＃xff0c;可以考虑增加Map数&＃xff0c;来使得每个map处理的数据量减少&＃xff0c;从而提高任务的执行效率。
• 假设有这样一个任务&＃xff1a;

Select data_desc,
count(1),
count(distinct id),
sum(case when …),
sum(case when …),
sum(…)
from a group by data_desc


• 如果表a只有一个文件&＃xff0c;大小为120M&＃xff0c;但包含几千万的记录&＃xff0c;如果用1个map去完成这个任务&＃xff0c;肯定是比较耗时的&＃xff0c;这种情况下&＃xff0c;我们要考虑将这一个文件合理的拆分成多个&＃xff0c;这样就可以用多个map任务去完成。

set mapred.reduce.tasks&＃61;10;


• map的最终个数取决于goalsize&＃61;min(totalsize/mapred.reduce.tasks,块大小),goalsize就相当于splitsize
• 但是往往这个是不生效的
  • 如果hive处理的文件是压缩模式&＃xff0c;且压缩模式不支持文件切分&＃xff0c;那么这个时候我们只能通过控制参数来减少map个数&＃xff0c;而不能通过配置参数来增加map个数&＃xff0c;所以Hive对于压缩不可切分文件的调优有限。
• 那么我们可以利用中间表

create table a_1 as
select * from a
distribute by CAST(RAND() *10 AS INT);


• 这样会将a表的记录&＃xff0c;随机的分散到包含10个文件的a_1表中&＃xff0c;再用a_1代替上面sql中的a表&＃xff0c;则会用10个map任务去完成。每个map任务处理大于12M&＃xff08;几百万记录&＃xff09;的数据&＃xff0c;效率肯定会好很多。
• 看上去&＃xff0c;貌似这两种有些矛盾&＃xff0c;一个是要合并小文件&＃xff0c;一个是要把大文件拆成小文件&＃xff0c;这点正是重点需要关注的地方&＃xff0c;根据实际情况&＃xff0c;控制map数量需要遵循两个原则&＃xff1a;使大数据量利用合适的map数&＃xff1b;使单个map任务处理合适的数据量。

• Hive自己如何确定reduce数&＃xff1a;
  • reduce个数的设定极大影响任务执行效率&＃xff0c;不指定reduce个数的情况下&＃xff0c;Hive会猜测确定一个reduce个数&＃xff0c;基于以下两个设定&＃xff1a;
  • hive.exec.reducers.bytes.per.reducer&＃xff08;每个reduce任务处理的数据量&＃xff0c;默认为1000^3&＃61;1G&＃xff09;
  • hive.exec.reducers.max&＃xff08;每个任务最大的reduce数&＃xff0c;默认为999&＃xff09;
  • 计算reducer数的公式很简单N&＃61;min(参数2&＃xff0c;总输入数据量/参数1)
    即&＃xff0c;如果reduce的输入&＃xff08;map的输出&＃xff09;总大小不超过1G,那么只会有一个reduce任务&＃xff1b;
  • 例如下图&＃xff0c;(67108864/1024/1024&＃61;64m),可以推测一个130m的输入数据量会启动reducer数&＃61;min&＃xff08;1099&＃xff0c;130m/64m&＃xff09;&＃61;3个reduce
    

reduce个数多好还是少好?

• 同map一样&＃xff0c;启动和初始化reduce也会消耗时间和资源&＃xff1b;
• 另外&＃xff0c;有多少个reduce,就会有多少个输出文件&＃xff0c;如果生成了很多个小文件&＃xff0c;那么如果这些小文件作为下一个任务的输入&＃xff0c;则也会出现小文件过多的问题&＃xff1b;

如何调整reduce个数?

• 调整reduce个数方法一&＃xff1a;
  • 调整hive.exec.reducers.bytes.per.reducer参数的值&＃xff1b;
  • 500000000即&＃xff08;500M&＃xff09;

set hive.exec.reducers.bytes.per.reducer&＃61;500000000;


• 调整reduce个数方法二&＃xff1b;

set mapred.reduce.tasks &＃61; 15;


什么情况下只有一个reduce&＃xff1f;

• 很多时候你会发现任务中不管数据量多大&＃xff0c;不管你有没有设置调整reduce个数的参数&＃xff0c;任务中一直都只有一个reduce任务&＃xff1b;其实只有一个reduce任务的情况&＃xff0c;除了数据量小于hive.exec.reducers.bytes.per.reducer参数值的情况外&＃xff0c;还有以下原因&＃xff1a;
  • a) 没有group by的汇总
    • 比如把select pt,count(1) from table_xxx where pt &＃61; &＃39;2020-08-11&＃39; group by pt; 写成 select count(1) from table_xxx where pt &＃61; &＃39;2020-08-11&＃39;;
  • b) 用了Order by
  • c) 有笛卡尔积
• 通常这些情况下&＃xff0c;除了找办法来变通和避免&＃xff0c;我暂时没有什么好的办法&＃xff0c;因为这些操作都是全局的&＃xff0c;所以hadoop不得不用一个reduce去完成&＃xff1b;
• 同样的&＃xff0c;在设置reduce个数的时候也需要考虑这两个原则&＃xff1a;使大数据量利用合适的reduce数&＃xff1b;使单个reduce任务处理合适的数据量。