1.分片(splits)相关概念

由InputFormat这个接口来定义的,其中有个getSplits方法。这里有一个新的概念:fileSplit。每个map处理一个fileSplit,所以有多少个fileSplit就有多少个map(map数并不是单纯的由用户设置决定的)。
我们来看一下hadoop分片splits的源码:

  1. long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);
  2. long minSize = Math.max(job.getLong("mapred.min.split.size", 1), minSplitSize);
  3. for (FileStatus file: files) {
  4. Path path = file.getPath();
  5. FileSystem fs = path.getFileSystem(job);
  6. if ((length != 0) && isSplitable(fs, path)) {
  7. long blockSize = file.getBlockSize();
  8. long splitSize = computeSplitSize(goalSize, minSize, blockSize);
  9. long bytesRemaining = length;
  10. while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
  11. String[] splitHosts = getSplitHosts(blkLocations,length-bytesRemaining, splitSize, clusterMap);
  12. splits.add(new FileSplit(path, length-bytesRemaining, splitSize, splitHosts));
  13. bytesRemaining -= splitSize;
  14. }
  15. if (bytesRemaining != 0) {
  16. splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining, blkLocations[blkLocations.length-1].getHosts()));
  17. }
  18. } else if (length != 0) {
  19. String[] splitHosts = getSplitHosts(blkLocations,0,length,clusterMap);
  20. splits.add(new FileSplit(path, 0, length, splitHosts));
  21. } else {
  22. //Create empty hosts array for zero length files
  23. splits.add(new FileSplit(path, 0, length, new String[0]));
  24. }
  25. }
  26. return splits.toArray(new FileSplit[splits.size()]);
  27. protected long computeSplitSize(long goalSize, long minSize, long blockSize) {
  28. return Math.max(minSize, Math.min(goalSize, blockSize));
  29. }

totalSize:是整个Map-Reduce job所有输入的总大小。

numSplits:来自job.getNumMapTasks(),即在job启动时用org.apache.hadoop.mapred.JobConf.setNumMapTasks(int n)设置的值,给M-R框架的Map数量的提示。

goalSize:是输入总大小与提示Map task数量的比值,即期望每个Mapper处理多少的数据,仅仅是期望,具体处理的数据数由下面的computeSplitSize决定。

minSplitSize:默认为1,可由子类复写函数protected void setMinSplitSize(long minSplitSize) 重新设置。一般情况下,都为1,特殊情况除外。

minSize:取的1和mapred.min.split.size中较大的一个。

blockSize:HDFS的块大小,默认为64M,一般大的HDFS都设置成128M。

splitSize:就是最终每个Split的大小,那么Map的数量基本上就是totalSize/splitSize。

接下来看看computeSplitSize的逻辑:首先在goalSize(期望每个Mapper处理的数据量)和HDFS的block size中取较小的,然后与mapred.min.split.size相比取较大的。

2.分片的逻辑策略

一个片为一个splits,即一个map,只要搞清楚片的大小,就能计算出运行时的map数。而一个split的大小是由goalSize, minSize, blockSize这三个值决定的。computeSplitSize的逻辑是,先从goalSize和blockSize两个值中选出最小的那个(比如一般不设置map数,这时blockSize为当前文件的块size,而goalSize是文件大小除以用户设置的map数得到的,如果没设置的话,默认是1),在默认的大多数情况下,blockSize比较小。然后再取blockSize和minSize中最大的那个。而minSize如果不通过”mapred.min.split.size”设置的话(”mapred.min.split.size”默认为0),minSize为1,这样得出的一个splits的size就是blockSize,即一个块一个map,有多少块就有多少map。

input_file_num : 输入文件的个数
(1)默认map个数
如果不进行任何设置,默认的map个数是和blcok_size相关的。
default_num = total_size / block_size;
(2)期望map数量
可以通过参数mapred.map.tasks来设置程序员期望的map个数,但是这个个数只有在大于default_num的时候,才会生效。
goal_num =mapred.map.tasks;
(3)设置处理的文件大小
可以通过mapred.min.split.size 设置每个task处理的文件大小,但是这个大小只有在大于
block_size的时候才会生效。
split_size = max(
mapred.min.split.size,
block_size);split_num = total_size / split_size;
(4)计算的map个数
compute_map_num = min(split_num, max(default_num, goal_num))
除了这些配置以外,mapreduce还要遵循一些原则。 mapreduce的每一个map处理的数据是不能跨越文件的,也就是说max_map_num <= input_file_num。 所以,最终的map个数应该为:
final_map_num = min(compute_map_num, input_file_num)

经过以上的分析,在设置map个数的时候,可以简单的总结为以下几点:
i)如果想增加map个数,则设置mapred.map.tasks 为一个较大的值。
ii)如果想减小map个数,则设置mapred.min.split.size 为一个较大的值。

3.Map数量优化调整方法

有了上述分析,如何调整map的数量就显而易见了。
减小Map-Reduce job 启动时创建的Mapper数量
当处理大批量的大数据时,一种常见的情况是job启动的mapper数量太多而超出了系统限制,导致Hadoop抛出异常终止执行。解决这种异常的思路是减少mapper的数量。具体如下:
输入文件size巨大,但不是小文件
这种情况可以通过增大每个mapper的input size,即增大minSize或者增大blockSize来减少所需的mapper的数量。增大blockSize通常不可行,因为当HDFS被hadoop namenode -format之后,blockSize就已经确定了(由格式化时dfs.block.size决定),如果要更改blockSize,需要重新格式化HDFS,这样当然会丢失已有的数据。所以通常情况下只能通过增大minSize,即增大mapred.min.split.size的值。
输入文件数量巨大,且都是小文件
所谓小文件,就是单个文件的size小于blockSize。这种情况通过增大mapred.min.split.size不可行,需要使用FileInputFormat衍生的CombineFileInputFormat将多个input path合并成一个InputSplit送给mapper处理,从而减少mapper的数量。
  
增加Map-Reduce job 启动时创建的Mapper数量
增加mapper的数量,可以通过减小每个mapper的输入做到,即减小blockSize或者减小mapred.min.split.size的值。通常情况下都是通过减小minSize,即减小mapred.min.split.size的值。

注意:
1、若文件是压缩文件且压缩的格式并不支持文件分割(无法从文件任意地方读取文件),则该文件不管多大都是一个分片
2、map job数最终是由分片数决定,程序员只能给一个期望map数。