Hive运行Mapreduce 中 Map 生成数量分析_set hive.input.format

背景

通常在执行HIVE sql，都会控制map 、reduce 个数在一个合理的范围，避免过多的资源浪费。由于hadoop 的每个任务在初始化是都需要一定时间，因此理想状态每个map 执行时间至少超过一分钟。

---

一、Hive input.format

查看集群hive 默认input.format

hive> set hive.input.format;
hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat
1
2

一般常见的inputformat 方法

set hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat; #默认
set hive.input.format=org.apache.hadoop.mapred.TextInputFormat
set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
1
2
3

二、Mapreduce 中Map 个数决定因素

MapReduce框架根据用户指定的InputFormat切割数据集，读取数据，并提供给map任务多条键值对进行处理，决定并行启动的map任务数目。MapReduce框架根据用户指定的OutputFormat，把生成的键值对输出为特定格式的数据。

1、TextInputFormat

在 Map 阶段读取数据前，根据InputFormat 方法将输入数据且分成若干个 split，以便确定 Map Task 的个数即 Mapper 的个数，在 MapReduce 框架中，一个 split 就意味着需要一个 Map Task;

Split个数取决于使用了哪种InputFormat，以及待处理的数据文件是否可分割。默认的FileInputFormat将根据block的个数来分割。
默认主要影响因素有

1、HDFS块的大小，即HDFS中dfs.block.size的值
2、文件大小
3、文件个数
4、splitsiez 大小。 分片是按照splitszie的大小进行分割的，一个split的大小在没有设置的情况下，默认等于hdfs block的大小。
在应用程序中可以通过两个参数来对splitsize进行调节，
splitsize=max（minimumsize，min（maximumsize，blocksize））
1
2
3
4
5
6

2、CombineHiveInputFormat

CombineInputFormat的功能，是将一个目录（可能包括多个小文件，不包括子目录）作为一个map的输入，而不是通常使用一个文件作为输入。

1、使用CombineHiveInputFormat，以下三个参数可以控制map 个数

确认set hive.hadoop.supports.splittable.combineinputformat 是否为true ,当为false 时，下面参数设置是没效果的。

hive.hadoop.supports.splittable.combineinputformat ：是否支持可切分的CombieInputFormat ，true是支持

最小split 大小
set mapred.min.split.size 或者 set mapreduce.input.fileinputformat.split.minsize

最大split大小
set mapred.max.split.size 或者 set mapreduce.input.fileinputformat.split.maxsize 

一个节点上最小的split大小
set mapred.min.split.size.per.node 或者 set mapreduce.input.fileinputformat.split.minsize.per.node

一个机架上最小的split大小
set mapred.min.split.size.per.rack 或者 set mapreduce.input.fileinputformat.split.minsize.per.rack  


  
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2、分片过程

I、 遍历node节点，在同一个node上的 input 文件进行合并，若合并超过max.split.size就生成一个新分片。小于时，若大于minsize.per.node，那也创建一个split 。小于时就暂存input文件, 继续合并下一个node节点中的文件。

遍历过程每个node最多生成一个split，为提高并发度，让split尽量分布到不同的node上。

II、 遍历 rack 机架，那么会把相同机架上的 node 中暂存的input文件block，进行合并，当合并文件的block 超过max.split.size就生成一个新分片。小于时，根据size.per.rack 参数，如果大于size.per.rack 则生成一个split ,小于则暂存block.

III、 对跨机架最终剩余的block 进行合并处理，大于max.split.size就生成一个新分片，循环处理，最后剩的数据创建一个split。

总结

注意：

如果只设置split.maxsize那么基本每个分片大小都需凑满split.maxsize

如果split.maxsize，minsize.per.node，minsize.per.rack 三个都没有设置，那是所有输入整合成一个分片！

切片大小默认等于Block大小，避免MR程序读取数据时，导致一个map任务需要大量访问非本地的Block造成网络开销，反而比正常的非合并方式更慢。
1
2
3
4
5

参考文章：
https://developer.aliyun.com/article/26262
https://www.jianshu.com/p/f0fe3b644a47
https://blog.csdn.net/hummingbird0/article/details/115084843