解读MapReduce中逻辑切片的实现方法getSplits源码

心已赠人 2022-03-27 08:40 95阅读 0赞

在书写任务的run方法时，定义输入输出格式中，我们使用了两段代码：

//2 设置数据读入与写出格式
    job.setInputFormatClass(TextInputFormat.class);
    job.setOutputFormatClass(TextOutputFormat.class);

其中**TextInputFormat**和**TextOutputFormat****被我们用来设置输入和输出格式**

**TextInputFormat中并没有**getSplits切片源码，getSplits切片源码在他的父类FileInputFormat中，它直接用就行了。

那我们来看看它父类中getSplits方法的源码实现。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70][]

###  getSplits源码解析。 ###

黄色数字修饰的代码部分，代表会有更多源码解析。

public List<InputSplit> getSplits(JobContext job) throws IOException {
    
            //秒表，用来计时，start方法表示计时开始。计时功能用于日志记录此代码运行时间。
    1       Stopwatch sw = (new Stopwatch()).start();
    
            //两参数用来计算切片大小。
    2       long minSize = Math.max(this.getFormatMinSplitSize(), getMinSplitSize(job));
    3       long maxSize = getMaxSplitSize(job);
    
            //使用ArrayList来存放切片对象InputSplit
            List<InputSplit> splits = new ArrayList();
            //拿到所有路径下的文件
    5       List<FileStatus> files = this.listStatus(job);
    
            //获取迭代器
            Iterator i$ = files.iterator();
    
            while(true) {
                while(true) {
                    //当迭代器发现后面还有文件时，继续迭代遍历。
                    while(i$.hasNext()) {
    
                        //返回每个文件的状态
                        FileStatus file = (FileStatus)i$.next();
    
                        //从状态中获取路径
                        Path path = file.getPath();
    
                        //及在文件的长度
                        long length = file.getLen();
    
                        if (length != 0L) {
    
                            //获取块位置
                            BlockLocation[] blkLocations;
                            if (file instanceof LocatedFileStatus) {
                                blkLocations = ((LocatedFileStatus)file).getBlockLocations();
                            } else {
                                FileSystem fs = path.getFileSystem(job.getConfiguration());
                                blkLocations = fs.getFileBlockLocations(file, 0L, length);
                            }
                            
                            //判断文件是否逻辑可切，比如有些音视频文件并不能逻辑切割。他们不支持分布式数据处理
                            if (this.isSplitable(job, path)) {
                                long blockSize = file.getBlockSize();
                                //计算切片大小
    4                           long splitSize = this.computeSplitSize(blockSize, minSize, maxSize);
    
                                long bytesRemaining;
                                int blkIndex;
    5 核心                      for(bytesRemaining = length; (double)bytesRemaining / (double)splitSize > 1.1D; bytesRemaining -= splitSize) {
                                    blkIndex = this.getBlockIndex(blkLocations, length - bytesRemaining);
                                    splits.add(this.makeSplit(path, length - bytesRemaining, splitSize, blkLocations[blkIndex].getHosts(), blkLocations[blkIndex].getCachedHosts()));
                                }
    
                                if (bytesRemaining != 0L) {
                                    blkIndex = this.getBlockIndex(blkLocations, length - bytesRemaining);
                                    splits.add(this.makeSplit(path, length - bytesRemaining, bytesRemaining, blkLocations[blkIndex].getHosts(), blkLocations[blkIndex].getCachedHosts()));
                                }
                            //不可切的时候，直接分配对应文件大小的逻辑切片。
                            } else {
                                splits.add(this.makeSplit(path, 0L, length, blkLocations[0].getHosts(), blkLocations[0].getCachedHosts()));
                            }
                        } else {
                            //当判断文件为0长度时，分配0长度的切片
                            splits.add(this.makeSplit(path, 0L, length, new String[0]));
                        }
                    }
    
                    job.getConfiguration().setLong("mapreduce.input.fileinputformat.numinputfiles", (long)files.size());
                     //秒表，stop方法表示计时结束。
                    sw.stop();
                    //日志记录了切片大小，和花费时间。之前会判断是否开启了日志记录。
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("Total # of splits generated by getSplits: " + splits.size() + ", TimeTaken: " + sw.elapsedMillis());
                    }
                    //返回切片数组
                    return splits;
                }
            }
        }

### **数字1：** ###

此处代码挖到底层是一个本地Native方法，会返回系统当前纳秒时间，并计算start到stop方法调用间隔的时间：![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 1][]

--------------------

### 数字 2： ###

会取两个方法产生的值的较大的一个。第一个方法默认返回**1**.

**long minSize = Math.max(this.getFormatMinSplitSize(), getMinSplitSize(job));**

protected long getFormatMinSplitSize() {
            return 1L;
        }

第二个方法的意思是，当我们配置**mapreduce.input.fileinputformat.split.minsize**参数时，返回此参数，否者返回**1**

public static long getMinSplitSize(JobContext job) {
            return job.getConfiguration().getLong("mapreduce.input.fileinputformat.split.minsize", 1L);
        }

### **结论1：**所以当我们不做配置时，最小切片大小为**1.** ###

--------------------

### **数字3：** ###

与取最小值同理，当我们配置**mapreduce.input.fileinputformat.split.maxsize**参数时，返回此参数

否者返回：**9223372036854775807  能看出这是long整型类型的最大值**

public static long getMaxSplitSize(JobContext context) {
        return context.getConfiguration().getLong("mapreduce.input.fileinputformat.split.maxsize", 9223372036854775807L);
    }

## ##

--------------------

### **数字4：** ###

计算切片大小时，这方法会取maxSize与吧blockSize中的小值去与minSize比，取大的那个。

这里因为blockSize为128MB，minSize和maxSize从上面得出分别为**1**和**9223372036854775807**

因此得出切片大小为128MB。

protected long computeSplitSize(long blockSize, long minSize, long maxSize) {
        return Math.max(minSize, Math.min(maxSize, blockSize));
    }

### **结论2：**最大切片大小为**128MB** ###

--------------------

### 数字5：核心代码 ###

for(
        bytesRemaining = length; 
        (double)bytesRemaining / (double)splitSize > 1.1D; 
        bytesRemaining -= splitSize
    ) {
        blkIndex = this.getBlockIndex(blkLocations, length - bytesRemaining);
        splits.add(
            this.makeSplit(
                path, length - bytesRemaining, 
                splitSize, 
                blkLocations[blkIndex].getHosts(),
                blkLocations[blkIndex].getCachedHosts()
            )
        );
      }
    if (bytesRemaining != 0L) {
        blkIndex = this.getBlockIndex(blkLocations, length - bytesRemaining);
        splits.add(
            this.makeSplit(path, 
                           length - bytesRemaining, 
                           bytesRemaining,
                           blkLocations[blkIndex].getHosts(),                              
                           blkLocations[blkIndex].getCachedHosts()
            )
        );
    }

makeSplit方法：

protected FileSplit makeSplit(
                            Path file, 
                            long start, 
                            long length, 
                            String[] hosts, 
                            String[] inMemoryHosts) {
                    return new FileSplit(file, start, length, hosts, inMemoryHosts);
    }

第一个参数path文件的位置。

第二个参数start为文件被切入的开始位置。

第三个参数length为切入长度。

比如来了一个150MB的切片

那么先判断150MB/128MB是否大于1.1，也相当于，判断进入的切片是否大于140.8MB（128\*1.1=140.8）

**大于时进入for循环中**

> 这150MB的数据会被从0位置开始放，因为第一次length - bytesRemaining=150-150=0
> 
> 然后放入128MB的数据进入逻辑切片中，因为splitSize=128MB
> 
>  
> 
> 第二次呢因为 bytesRemaining -= splitSize 这段代码 150-128=22
> 
> 22<140.8MB
> 
> 所以进入if判断中
> 
>

**小于则进入if判断中**

> 进入if判断后，如果刚好切完 剩余数据为0，这不切了，否者继续切完剩下部分。
> 
> 会使用文件长度减去剩余数据，150-22=128       //length - bytesRemaining
> 
> 因此切刀会从文件的128MB的位置开始切，切到完，也就是把文件剩余部分全切出来。//bytesRemaining
> 
> 放入逻辑切片中

数字5这段核心代码的意思大概就是

>  *  **从路径中获取一个 X 大小的文件**
>  *  **把他们切成许多块**
>  *  **每块大小为一个 140.8MB 大小的块**
>  *  **文件末尾数据不足140.8MB大小时，也会切成一个块放入逻辑切块区。**

这就是**getSplits(JobContext job)**代码精髓吧。

**逻辑切片的意思是，这个切片中并不是真正的切片数据，这个逻辑切片只记录了真实切片的存放路径，逻辑切片在真实切片的切入位置，以及切入长度。三个参数。**

看懂的给个赞吧\_(:з」∠)\_

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70]: /images/20220327/f06e3ca08d1240688a21374d73164aee.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI5NTE5MDQx_size_16_color_FFFFFF_t_70 1]: /images/20220327/a683a127ea4b462cb16eac4e5a608f71.png