hadoop 分片与分块，map task和reduce task的理解

Posted 2020-08-06 渣渣的博客

分块：Block

　　HDFS存储系统中，引入了文件系统的分块概念（block），块是存储的最小单位，HDFS定义其大小为64MB。与单磁盘文件系统相似，存储在 HDFS上的文件均存储为多个块，不同的是，如果某文件大小没有到达64MB，该文件也不会占据整个块空间。在分布式的HDFS集群上，Hadoop系统保证一个块存储在一个datanode上。

　　把File划分成Block，这个是物理上真真实实的进行了划分，数据文件上传到HDFS里的时候，需要划分成一块一块，每块的大小由hadoop-default.xml里配置选项进行划分。一个大文件可以把划分后的所有块存储到同一个磁盘上，也可以在每个磁盘上都存在这个文件的分块。

这个就是默认的每个块64M：

<property>  
  <name>dfs.block.size</name>  
  <value>67108864</value>  
  <description>The default block size for new files.</description>  
</property>  

数据划分的时候有冗余，即进行备份，个数是由以下配置指定的。具体的物理划分步骤由Namenode决定。

1 <property>  
2   <name>dfs.replication</name>  
3   <value>3</value>  
4   <description>Default block replication.   
5   The actual number of replications can be specified when the file is created.  
6   The default is used if replication is not specified in create time.  
7   </description>  
8 </property>  

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

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相比取较大的。

 

　　一个片为一个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）期望大小
可以通过参数
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个数的时候，可以简单的总结为以下几点：
（1）如果想增加map个数，则设置mapred.map.tasks 为一个较大的值。
（2）如果想减小map个数，则设置mapred.min.split.size 为一个较大的值。

map task

如何调整map数量：

有了2的分析，下面调整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的值。

 

参考资料：

http://blog.csdn.net/strongerbit/article/details/7440111

http://blog.csdn.net/clerk0324/article/details/50887866

http://blog.csdn.net/yeruby/article/details/20068731

 http://zhidao.baidu.com/link?url=fLPxBdClbJ0R0-VGGiSbXR4bx9tlhadShKNYQ76CNrShD7Q7zsxr5b_df9gl7l5LA3VsXTkbeTvtOlj1fQY_yNx7bzopbfrW_tSbzN2J6Se