Hadoop-MapReduce-详细解析InputFormat工作机制，源码分析，切片机制-连载中

浅浅的花香味﹌ 2023-03-01 09:27 153阅读 0赞

# 详细解析InputFormat工作机制 #

### 文章目录 ###

*  详细解析InputFormat工作机制
 *   *  1 InputFormat数据输入
     *   *  1.1 切片与MapTask并行度决定机制
         *  1.2 FileInputFormat切片源码解析
         *  1.3 FileInputFormat切片机制
         *  1.4 TextInputFormat
         *  1.5 CombineTextInputFormat切片机制
         *  1.6 CombineTextInputFormat案例实操

## 1 InputFormat数据输入 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70]

### 1.1 切片与MapTask并行度决定机制 ###

**1）问题引出**

MapTask的并行度决定Map阶段的任务处理并发度，进而影响到整个Job的处理速度。

**思考：** 1G的数据，启动8个MapTask，可以提高集群的并发处理能力。那么1K的数据，也启动8个MapTask，会提高集群性能吗？MapTask并行任务是否越多越好呢？哪些因素影响了MapTask并行度？

**2）MapTask并行度决定机制**

**数据块：**Block是HDFS物理上把数据分成一块一块，在磁盘中真实存在。

**数据切片：**数据切片只是在逻辑上对输入进行分片，并不会在磁盘上将其切分成片进行存储。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 1]

### 1.2 FileInputFormat切片源码解析 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 2]  
InputFormat顶层抽象类: 负责Map端数据的输入  
`getSplits()`: 生成切片信息。  
`createRecordReader()`: 负责输入数据的读取处理

*  **子抽象类**: FileInputFormat
    
    `getSplits()`: 做了具体的实现. Hadoop默认的切片规则。
    
    **具体实现类**:
    
     *  TextInputFormat : Hadoop默认使用的InputFormat实现类  
        `createRecordReader()`: 做了具体的实现.  
        使用LineRecordReader来读取数据，就是一行一行读取.
     *  CombineFileInputFormat：用于小文件过多的场景，它可以将多个小文件从逻辑上规划到一个切片中

**1）FileInputFormat切片源码解析(input.getSplits(job))**

public List<InputSplit> getSplits(JobContext job) throws IOException {
        
        StopWatch sw = new StopWatch().start();
        long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));  // 默认是1 
    	 // minSize ==> mapreduce.input.fileinputformat.split.minsize
        long maxSize = getMaxSplitSize(job);  // 默认是Long.MAX_VALUE
    	 // maxSize ==> mapreduce.input.fileinputformat.split.maxsize
    
        // generate splits
        List<InputSplit> splits = new ArrayList<InputSplit>();
        List<FileStatus> files = listStatus(job);
    
        boolean ignoreDirs = !getInputDirRecursive(job)
          && job.getConfiguration().getBoolean(INPUT_DIR_NONRECURSIVE_IGNORE_SUBDIRS, false);
        
        // 循环每个文件，为每个文件单独生成切片.
        // 可以看出，是针对单个文件进行切片的.
        for (FileStatus file: files) {
        
          if (ignoreDirs && file.isDirectory()) {
        
            continue;
          }
          Path path = file.getPath(); // 获取文件的路径
          long length = file.getLen(); // 获取文件的长度（文件大小byte）
          if (length != 0) {
        
            BlockLocation[] blkLocations;
            if (file instanceof LocatedFileStatus) {
        
              blkLocations = ((LocatedFileStatus) file).getBlockLocations();
            } else {
        
              FileSystem fs = path.getFileSystem(job.getConfiguration());
              blkLocations = fs.getFileBlockLocations(file, 0, length);
            }
            // 判断文件是否可以切片，有些压缩文件是不可以切片的
            if (isSplitable(job, path)) {
        
              long blockSize = file.getBlockSize(); //获取块大小
    	  	  //获取切片大小，通过 Math.max(minSize, Math.min(maxSize, blockSize)) 计算出来
              long splitSize = computeSplitSize(blockSize, minSize, maxSize);
    		  // 文件剩余长度，刚开始 等于 文件长度
              long bytesRemaining = length;
    	  // 当前文件剩余的大小 除以  切片大小 > 1.1 ，继续切片，否则剩余的大小生成一个切片。
    	  // 避免数据倾斜问题 //SPLIT_SLOP = 1.1;
              while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
         
                int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
                  // 生成一个切片并添加到集合
                  // file:/D:/input/inputflow/phone_data.txt:0+1178（切片的范围（大小）） 
                splits.add(makeSplit(path, length-bytesRemaining, splitSize,
                            blkLocations[blkIndex].getHosts(),
                            blkLocations[blkIndex].getCachedHosts()));
                bytesRemaining -= splitSize; // 剩余文件大小 减去 切片大小，然后继续判断
              }
    
              if (bytesRemaining != 0) {
        
                int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
                splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
                           blkLocations[blkIndex].getHosts(),
                           blkLocations[blkIndex].getCachedHosts()));
              }
            } else {
         // not splitable
              if (LOG.isDebugEnabled()) {
        
                // Log only if the file is big enough to be splitted
                if (length > Math.min(file.getBlockSize(), minSize)) {
        
                  LOG.debug("File is not splittable so no parallelization "
                      + "is possible: " + file.getPath());
                }
              }
              splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(),
                          blkLocations[0].getCachedHosts()));
            }
          } else {
         
            //Create empty hosts array for zero length files
            splits.add(makeSplit(path, 0, length, new String[0]));
          }
        }
        // Save the number of input files for metrics/loadgen
        job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
        sw.stop();
        if (LOG.isDebugEnabled()) {
        
          LOG.debug("Total # of splits generated by getSplits: " + splits.size()
              + ", TimeTaken: " + sw.now(TimeUnit.MILLISECONDS));
        }
    
        // 返回切片信息的集合
        return splits;
      }
    提取到的信息:
    1. 本地的块大小默认是32M
    2. long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));  // 1 
    	 // minSize ==> mapreduce.input.fileinputformat.split.minsize  通过该参数改变minSize
        long maxSize = getMaxSplitSize(job);  // Long.MAX_VALUE
    	 // maxSize ==> mapreduce.input.fileinputformat.split.maxsize  通过该参数改变maxSize
    3. long splitSize = computeSplitSize(blockSize, minSize, maxSize);
       // return Math.max(minSize, Math.min(maxSize, blockSize));  计算切片大小
    
    4. 一个切片信息: file:/D:/input/inputflow/phone_data.txt:0+1178   
        意思就是读取/D:/input/inputflow/phone_data.txt的 0~1178 范围的数据.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 3]

### 1.3 FileInputFormat切片机制 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 4]  
**FileInputFormat切片大小的参数配置**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 5]

### 1.4 TextInputFormat ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 6]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 7]

### 1.5 CombineTextInputFormat切片机制 ###

框架默认的 **TextInputFormat** 切片机制是对任务按文件规划切片，不管文件多小，都会是一个单独的切片，都会交给一个MapTask，这样如果有大量小文件，就会产生大量的MapTask，处理效率极其低下。

**1）应用场景：**

**CombineTextInputFormat**用于小文件过多的场景，它可以将多个小文件从逻辑上规划到一个切片中，这样，多个小文件就可以交给一个MapTask处理。

**2）虚拟存储切片最大值设置**

CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m

*  注意：虚拟存储切片最大值设置最好根据实际的小文件大小情况来设置具体的值。

**3）切片机制**

生成切片过程包括：虚拟存储过程和切片过程二部分。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 8]  
（1）虚拟存储过程：

> 将输入目录下所有文件大小，依次和设置的`setMaxInputSplitSize`值比较，如果不大于设置的最大值，逻辑上划分一个块。如果输入文件大于设置的最大值且大于两倍，那么以最大值切割一块；当剩余数据大小超过设置的最大值且不大于最大值2倍，此时将文件均分成2个虚拟存储块（防止出现太小切片）。
> 
> 例如`setMaxInputSplitSize`值为**4M**，输入文件大小为8.02M，则先逻辑上分成一个4M。剩余的大小为4.02M，如果按照4M逻辑划分，就会出现0.02M的小的虚拟存储文件，所以将剩余的4.02M文件切分成（2.01M和2.01M）两个文件。

（2）切片过程：

> （a）判断虚拟存储的文件大小是否大于`setMaxInputSplitSize`值，大于等于则单独形成一个切片。  
> （b）如果不大于则跟下一个虚拟存储文件进行合并，然后再重复（a）步骤。  
> （c）测试举例：有4个小文件大小分别为1.7M、5.1M、3.4M以及6.8M这四个小文件，则虚拟存储之后形成6个文件块，大小分别为：
> 
> 1.7M，（2.55M、2.55M），3.4M,（3.4M、3.4M）  
> 最终会形成3个切片，大小分别为：（1.7+2.55）M，（2.55+3.4）M，（3.4+3.4）M

### 1.6 CombineTextInputFormat案例实操 ###

**1）需求**

将输入的大量小文件合并成一个切片统一处理。

（1）输入数据

准备4个小文件

（2）期望

期望一个切片处理4个文件

**2）实现过程**

**（1）** 不做任何处理，运行WordCount案例程序，观察切片个数为4。  
![在这里插入图片描述][20200723142645296.png]  
**（2）** 在WordcountDriver中增加如下代码，运行程序，并观察运行的切片个数为3。

（a）驱动类中添加代码如下：

// 如果不设置InputFormat，它默认用的是TextInputFormat.class
    job.setInputFormatClass(CombineTextInputFormat.class);
    //虚拟存储切片最大值设置4m
    CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);

（b）运行如果为3个切片。  
![在这里插入图片描述][20200723142710306.png]  
**（3）** 在WordcountDriver中增加如下代码，运行程序，并观察运行的切片个数为1。

（a）驱动中添加代码如下：

// 如果不设置InputFormat，它默认用的是TextInputFormat.class
    job.setInputFormatClass(CombineTextInputFormat.class);
    //虚拟存储切片最大值设置20m
    CombineTextInputFormat.setMaxInputSplitSize(job, 20971520);

（b）运行如果为1个切片。  
![在这里插入图片描述][20200723142747773.png]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70]: /images/20230209/e181c3545dc74e748af8ffceb84775ab.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200723141955661.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200726141404931.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200723142248122.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/2020072314230866.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/20200723142336827.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/20200723142357527.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/20200723142412702.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/20200723142551902.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNzI3MDk1,size_16,color_FFFFFF,t_70
[20200723142645296.png]: https://img-blog.csdnimg.cn/20200723142645296.png
[20200723142710306.png]: https://img-blog.csdnimg.cn/20200723142710306.png
[20200723142747773.png]: https://img-blog.csdnimg.cn/20200723142747773.png