Hadoop之分块、分片与shuffle机制详解；设置map任务数目

待我称王封你为后i 2021-09-25 01:30 364阅读 0赞

![70][]

# 一  分块（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>

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

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

# 二  分片（splits） #

由InputFormat这个接口来定义的，其中有个getSplits方法。这里有一个新的概念：fileSplit。**每个map处理一个fileSplit，所以有多少个fileSplit就有多少个map（map数并不是单纯的由用户设置决定的）。**

我们来看一下hadoop分配splits的源码：

long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);
    
    long minSize = Math.max(job.getLong("mapred.min.split.size", 1), minSplitSize);
    
    
    for (FileStatus file: files) {
    
    Path path = file.getPath();
    
    FileSystem fs = path.getFileSystem(job);
    
    if ((length != 0) && isSplitable(fs, path)) {
    
    long blockSize = file.getBlockSize();
    
    long splitSize = computeSplitSize(goalSize, minSize, blockSize);
    
    
    long bytesRemaining = length;
    
    while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
    
    String[] splitHosts = getSplitHosts(blkLocations,length-bytesRemaining, splitSize, clusterMap);
    
    splits.add(new FileSplit(path, length-bytesRemaining, splitSize, splitHosts));
    
    bytesRemaining -= splitSize;
    
    }
    
    
    if (bytesRemaining != 0) {
    
    splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining, blkLocations[blkLocations.length-1].getHosts()));
    
    }
    
    } else if (length != 0) {
    
    String[] splitHosts = getSplitHosts(blkLocations,0,length,clusterMap);
    
    splits.add(new FileSplit(path, 0, length, splitHosts));
    
    } else {
    
    //Create empty hosts array for zero length files
    
    splits.add(new FileSplit(path, 0, length, new String[0]));
    
    }
    
    }
    
    
    return splits.toArray(new FileSplit[splits.size()]);
    
    
    protected long computeSplitSize(long goalSize, long minSize, long blockSize) {
    
    return Math.max(minSize, Math.min(goalSize, blockSize));
    
    }

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）期望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,min(mapred.max.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 为一个较大的值。

整理一下 ：

split\_num 肯定会小于或等于默认的分块大小，即：split\_num <=default\_num

而当期望的map个数goal\_num > default\_num时才会有用

个人认为compute\_num该这么取 compute\_num = max（min(split\_num,default\_num),goal\_num）

除了这些配置以外，mapreduce还要遵循一些原则。 mapreduce的每一个map处理的数据是不能跨越文件的，也就是说min\_map\_num >= input\_file\_num。 所以，最终的map个数应该为：

final\_map\_num = max(compute\_map\_num, input\_file\_num)

[https://blog.csdn.net/senwen616/article/details/82753693][https_blog.csdn.net_senwen616_article_details_82753693]

##  ##

###  ###

### Map数量的调整 ###

有了上述分析，如何调整map的数量就显而易见了。

**1 减小Map-Reduce job 启动时创建的Mapper数量**

当处理大批量的大数据时，一种常见的情况是job启动的mapper数量太多而超出了系统限制，导致Hadoop抛出异常终止执行。解决这种异常的思路是减少mapper的数量。具体如下：

**　　1.1输入文件size巨大，但不是小文件**

这种情况可以通过增大每个mapper的input size，即增大minSize或者增大blockSize来减少所需的mapper的数量。增大blockSize通常不可行，因为当HDFS被hadoop namenode -format之后，blockSize就已经确定了（由格式化时dfs.block.size决定），如果要更改blockSize，需要重新格式化HDFS，这样当然会丢失已有的数据。**所以通常情况下只能通过增大minSize，即增大mapred.min.split.size的值。**

**　　1.2输入文件数量巨大，且都是小文件（每个小文件会生成一个map任务）**

所谓小文件，就是单个文件的size小于blockSize。这种情况通过增大mapred.min.split.size不可行，**需要使用FileInputFormat衍生的CombineFileInputFormat将多个input path合并成一个InputSplit送给mapper处理，从而减少mapper的数量。**

**2 增加Map-Reduce job 启动时创建的Mapper数量**

增加mapper的数量，可以通过减小每个mapper的输入做到，即减小blockSize或者减小mapred.min.split.size的值。通常情况下都是通过增大minSize，即增大mapred.min.split.size的值。

# 三  Shuffle机制 #

Shuffle过程是MapReduce的核心，描述着数据从map task输出到reduce task输入的这段过程。

Hadoop的集群环境，大部分的map task和reduce task是执行在不同的节点上的，那么reduce就要取map的输出结果。那么集群中运行多个Job时，task的正常执行会对集群内部的网络资源消耗严重。虽说这种消耗是正常的，是不可避免的，但是，我们可以采取措施尽可能的减少不必要的网络资源消耗。另一方面，每个节点的内部，相比于内存，磁盘IO对Job完成时间的影响相当的大。

所以：从以上分析，shuffle过程的基本要求：

1.完整地从map task端拉取数据到reduce task端

2.在拉取数据的过程中，尽可能地减少网络资源的消耗

3.尽可能地减少磁盘IO对task执行效率的影响

那么，Shuffle的设计目的就要满足以下条件：

1.保证拉取数据的完整性

2.尽可能地减少拉取数据的数据量

3.尽可能地使用节点的内存而不是磁盘

**一、map阶段**  
map节点执行map task任务生成map的输出结果  
shuffle的工作内容：  
从运算效率的出发点，map输出结果优先存储在map节点的内存中。每个map task都有一个内存缓冲区，存储着map的输出结果，默认大小100M（由io.sort.mb属性控制），一旦内存缓冲达到阈值0.8（io.sort.spill.percent），一个后台线程就会将将缓冲区中的数据以一个临时文件的方式存（spill)到磁盘的指定目录（mapred.local.dir）。同时，在写磁盘前，会进行partition、sort操作, 如果有combiner, combine排序后数据。 当整个map task结束后再对磁盘中这个map task所产生的所有临时文件做合并，生成最终的输出文件。最后，等待reduce task来拉取数据。当然，如果map task的结果不大，能够完全存储到内存缓冲区，且未达到内存缓冲区的阀值，那么就不会有写临时文件到磁盘的操作，也不会有后面的合并。

图解如下：

![70 1][]

**环形缓冲区：**是使用指针机制把内存中的地址首尾相接形成一个存储中间数据的缓存区域，默认100MB;80M阈值,20M缓冲区,是为了解决写入环形缓冲区数据的速度大于写出到spill文件的速度是数据的不丢失;

**Spill文件：**spill文件是环形缓冲区到达阈值后写入到磁盘的单个文件.这些文件在map阶段计算结束时,会合成分好区的一个merge文件供给给reduce任务抓取;spill文件过小的时候,就不会浪费io资源合并merge;默认情况下3个以下spill文件不合并;对于在环形缓冲区中的数据,最终达不到80M但是数据已经计算完毕的情况,map任务将会调用flush将缓冲区中的数据强行写出spill文件。

**二、reduce阶段**

当mapreduce任务提交后，reduce task就不断通过RPC从JobTracker那里获取map task是否完成的信息，如果获知某台TaskTracker上的map task执行完成，Shuffle的后半段过程就开始启动。其实呢，reduce task在执行之前的工作就是：不断地拉取当前job里每个map task的最终结果，并对不同地方拉取过来的数据不断地做merge，过程如下：

![70 2][]

reduce阶段分三个步骤：  
抓取，合并，排序  
1 reduce 任务会创建并行的抓取线程（fetcher）负责从完成的map任务中获取结果文件，是否完成是通过rpc心跳监听，通过http协议抓取；默认是5个抓取线程，可调，为了使整体并行，在map任务量大，分区多的时候，抓取线程调大；  
2 抓取过来的数据会先保存在内存中，如果内存过大也溢出，不可见，不可调，但是单位是每个merge文件，不会切分数据；每个merge文件都会被封装成一个segment的对象，这个对象控制着这个merge文件的读取记录操作，有两种情况出现：  
      在内存中有merge数据  
      在溢写之后存到磁盘上的数据  
通过构造函数的区分，来分别创建对应的segment对象  
3 这种segment对象会放到一个内存队列中MergerQueue，对内存和磁盘上的数据分别进行合并，内存中的merge对应的segment直接合并，磁盘中的合并与一个叫做合并因子的factor有关（默认是10）  
4 排序问题

MergerQueue继承轮换排序的接口，每一个segment 是排好序的，而且按照key的值大小逻辑（和真的大小没关系）；每一个segment的第一个key都是逻辑最小，而所有的segment的排序是按照第一个key大小排序的，最小的在前面，这种逻辑总能保证第一个segment的第一个key值是所有key的逻辑最小文件，合并之后，最终交给reduce函数计算的，是MergeQueue队列，每次计算的提取数据逻辑都是提取第一个segment的第一个key和value数据，一旦segment被调用了提取key的方法，MergeQueue队列将会整体重新按照最小key对segment排序，最终形成整体有序的计算结果；

# 四  运行流程 #

![70 3][]

转自：[https://blog.csdn.net/ZG\_24/article/details/80635056?utm\_medium=distribute.pc\_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.control&depth\_1-utm\_source=distribute.pc\_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.control][https_blog.csdn.net_ZG_24_article_details_80635056_utm_medium_distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.control_depth_1-utm_source_distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.control]

[70]: /images/20210923/728c3edbecc041009c75e229a007bad4.png
[https_blog.csdn.net_senwen616_article_details_82753693]: https://blog.csdn.net/senwen616/article/details/82753693
[70 1]: /images/20210923/d23a45b647834defb1b26393b3893677.png
[70 2]: /images/20210923/212868b5338e42c5b9015bcf17149df6.png
[70 3]: /images/20210923/36a982c669914074951e213e2ac08b27.png
[https_blog.csdn.net_ZG_24_article_details_80635056_utm_medium_distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.control_depth_1-utm_source_distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.control]: https://blog.csdn.net/ZG_24/article/details/80635056?utm_medium=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.control&depth_1-utm_source=distribute.pc_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.control