hadoop2.7作业提交详解之文件分片

╰+攻爆jí腚メ 2021-11-01 04:16 253阅读 0赞

在前面一篇文章中（[hadoop2.7之作业提交详解（上）][hadoop2.7]）中涉及到文件的分片。

JobSubmitter.submitJobInternal方法中调用了  
**int maps = writeSplits(job, submitJobDir); //设置map的数量，而map的数量是根据文件的大小和分片的大小，以及文件的数量决定的**

接下来我们看一下JobSubmitter.writeSplits方法：

private int writeSplits(org.apache.hadoop.mapreduce.JobContext job,
        Path jobSubmitDir) throws IOException,
        InterruptedException, ClassNotFoundException {
      JobConf jConf = (JobConf)job.getConfiguration();
      int maps;
      if (jConf.getUseNewMapper()) {
        maps = writeNewSplits(job, jobSubmitDir); //这里我们使用新的方式
      } else {
        maps = writeOldSplits(jConf, jobSubmitDir);
      }
      return maps;
    }

接下来继续看JobSubmitter.writeNewSplits方法:

private <T extends InputSplit>
    int writeNewSplits(JobContext job, Path jobSubmitDir) throws IOException,
        InterruptedException, ClassNotFoundException {
      Configuration conf = job.getConfiguration();
      InputFormat<?, ?> input =
        ReflectionUtils.newInstance(job.getInputFormatClass(), conf);  //输入对象，InputFormat是个抽象类  
    
      List<InputSplit> splits = input.getSplits(job); //调用InputFormat实现类的getSplits方法
      T[] array = (T[]) splits.toArray(new InputSplit[splits.size()]);
    
      // sort the splits into order based on size, so that the biggest
      // go first
      Arrays.sort(array, new SplitComparator()); //对切片的大小进行排序，最大的放最前面
      JobSplitWriter.createSplitFiles(jobSubmitDir, conf, 
          jobSubmitDir.getFileSystem(conf), array);//创建Split文件 
      return array.length;
    }

接下来看一下InputFormat这个抽象类：

public abstract class InputFormat<K, V> {
        //用来返回分片结果
        public abstract 
        List<InputSplit> getSplits(JobContext context
                                   ) throws IOException, InterruptedException;
        //RecordReader是用来从一个输入分片中读取一个一个的K-V对的抽象类，我们可以将其看作是在InputSplit上的迭代器。
        //最主要的方法就是nextKeyvalue()方法，由它获取分片上的下一个K-V 对。
        public abstract 
        RecordReader<K,V> createRecordReader(InputSplit split,
                                             TaskAttemptContext context
                                            ) throws IOException, 
                                                     InterruptedException;
    
    }

接下来我们继续看这个抽象类的实现类：

public class TextInputFormat extends FileInputFormat；  
public abstract class FileInputFormat<K, V> extends InputFormat；  
public abstract class InputFormat。

由于TextInputFormat从抽象类FileInputFormat中继承，所以大部分的方法都来自于FileInputFormat类，TextInputFormat类只重写了两个方法：如下：

public class TextInputFormat extends FileInputFormat<LongWritable, Text> {
    
      @Override
      public RecordReader<LongWritable, Text> 
        createRecordReader(InputSplit split,
                           TaskAttemptContext context) {
        String delimiter = context.getConfiguration().get(
            "textinputformat.record.delimiter");
        byte[] recordDelimiterBytes = null;
        if (null != delimiter)
          recordDelimiterBytes = delimiter.getBytes(Charsets.UTF_8);
          //LineRecordReader由一个FileSplit构造出来，start是这个FileSplit的起始位置，pos是当前读取分片的位置，
          //end是分片结束位置，in是打开的一个读取这个分片的输入流，它是使用这个FileSplit对应的文件名来打开的。
          //key和value则分别是每次读取的K-V对。然后我们还看到可以利用getProgress()来跟踪读取分片的进度，
          //这个函数就是根据已经读取的K-V对占总K-V对的比例来显示进度的
        return new LineRecordReader(recordDelimiterBytes);
      }
    
      @Override
      protected boolean isSplitable(JobContext context, Path file) {
     //如果是压缩文件就不切分，非压缩文件就切分。
        final CompressionCodec codec =
          new CompressionCodecFactory(context.getConfiguration()).getCodec(file);
        if (null == codec) {
          return true;
        }
        return codec instanceof SplittableCompressionCodec;
      }
    }

我们在返回到JobSubmitter.writeNewSplits方法中，有List<InputSplit> splits = input.getSplits(job);主要是调用了**TextInputFormat.getSplits()**方法，而TextInputFormat继承了FileInputFormat类，所以调用的就是FileInputFormat.getSplits()方法：

public List<InputSplit> getSplits(JobContext job) throws IOException {
      StopWatch sw = new StopWatch().start();//用来计算纳秒级别的时间
      long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job)); //最小值默认为1
      long maxSize = getMaxSplitSize(job); //最大值为long的最大值，默认为0x7fffffffffffffffL
    
      // generate splits
      List<InputSplit> splits = new ArrayList<InputSplit>();
      List<FileStatus> files = listStatus(job); //获得所有的输入文件
      for (FileStatus file: files) {
        Path path = file.getPath(); //文件路径
        long length = file.getLen(); //文件大小
        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(); //128M
            long splitSize = computeSplitSize(blockSize, minSize, maxSize); //计算分片的大小，默认为128M 
    
            long bytesRemaining = length;
            while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) { //判断剩余文件大小是否大于128M*1.1 
              int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);//f返回每个分片起始位置
              splits.add(makeSplit(path, length-bytesRemaining, splitSize,
                          blkLocations[blkIndex].getHosts(),
                          blkLocations[blkIndex].getCachedHosts()));
              bytesRemaining -= splitSize; // 依次减去分片的大小，对剩余长度再次分片
            }
    // 多次分片后，最后的数据长度仍不为0但又不足一个分片大小
            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
            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()); //设置参数NUM_INPUT_FILES
      sw.stop();
      if (LOG.isDebugEnabled()) {
        LOG.debug("Total # of splits generated by getSplits: " + splits.size()
            + ", TimeTaken: " + sw.now(TimeUnit.MILLISECONDS));
      }
      return splits;
    }
    //public class FileSplit extends InputSplit implements Writable {
    //  private Path file;//输入文件路径 
    //  private long start;//分片在文件中的位置(起点)
    //  private long length;//分片长度
    //  private String[] hosts;//这个分片所在数据块的多个复份所在节点
    //  private SplitLocationInfo[] hostInfos;//每个数据块复份所在节点,以及是否缓存 
    //}
    //makeSplit方法存放的分片格式
    protected FileSplit makeSplit(Path file, long start, long length, 
                                  String[] hosts, String[] inMemoryHosts) {
      return new FileSplit(file, start, length, hosts, inMemoryHosts);
    }
    
    //计算分片的大小
    protected long computeSplitSize(long blockSize, long minSize,
                                    long maxSize) {
      return Math.max(minSize, Math.min(maxSize, blockSize));
    }

通过FileInputFormat.getSplits()，可以返回一个存放分片的ArraryList，接下继续回到JobSubmitter.writeNewSplits方法中：

接下来将ArrayList转换为数组，并根据分片的大小排序。然后调用JobSplitWriter.createSplitFiles()方法创建split文件。最后返回数组的长度，也就是map的个数。

转载于:https://www.cnblogs.com/zsql/p/11276584.html

[hadoop2.7]: https://www.cnblogs.com/zsql/p/11239429.html