ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 大数据 > Hadoop > Mapreduce源码分析分片、处理流程

Mapreduce源码分析分片、处理流程

原创 Hadoop 作者：破棉袄 时间：2014-08-12 15:29:08 0 删除 编辑

InputFormat（代码以TextInputFormat为例）：

FileinputFormat中有三个重要方法：
            1).isSplitable
            2).getSplits
            3).createRecordReader

一.isSplitable方法：
                改方法返回值为bool类型，判断是否进行分片。
  
二.getSplits方法：
         该方法返回值为List。如果isSplitable返回值为ture，则该方法返回的为分块之后的文件，否则为没有分块的文件。
          InputSplit是一个抽象类，FileSplit继承该类，FileSplit有以下属性：
              private Path file;               //文件路径
              private long start;            //该块存储的文件的起始下标
              private long length;         //该块存储块大小
              private String[] hosts;    //存储该块的主机

     getSplits方法源码：                 
  1. /**
  2.    * Generate the list of files and make them into FileSplits.
  3.    * @param job the job context
  4.    * @throws IOException
  5.    */
  6.   public List<InputSplit> getSplits(JobContext job) throws IOException {
  7.     long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
  8.     long maxSize = getMaxSplitSize(job);

  10.     // generate splits
  11.     List<InputSplit> splits = new ArrayList<InputSplit>();
  12.     List<FileStatus> files = listStatus(job);                //该方法会遍历输入目录和目录的子目录，将文件信息保存到List中。
  13.     for (FileStatus file: files) {                           //遍历该List将文件放入到 splits 中。
  14.       Path path = file.getPath();
  15.       long length = file.getLen();
  16.       if (length != 0) {
  17.         BlockLocation[] blkLocations;
  18.         if (file instanceof LocatedFileStatus) {
  19.           blkLocations = ((LocatedFileStatus) file).getBlockLocations();
  20.         } else {
  21.           FileSystem fs = path.getFileSystem(job.getConfiguration());
  22.           blkLocations = fs.getFileBlockLocations(file, 0, length);
  23.         }
  24.         if (isSplitable(job, path)) {                                  //如果为分片
  25.           long blockSize = file.getBlockSize();
  26.           long splitSize = computeSplitSize(blockSize, minSize, maxSize); //获取分片大小

  28.           long bytesRemaining = length;
  29.           while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {  // SPLIT_SLOP值为1.1 如果 文件总大小/分片大小<1.1 即使该文件大小大于块大小，那该文件也不会分割。
  30.             int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
  31.             splits.add(makeSplit(path, length-bytesRemaining, splitSize,blkLocations[blkIndex].getHosts()));    
  32.              // makeSplit方法： protected FileSplit makeSplit(Path file, long start, long length, String[] hosts) {
  33.              //            return new FileSplit(file, start, length, hosts);
             
  34.          
    1. bytesRemaining -= splitSize;                                   //          bytesRemaining 减去 已经分片的大小splitSize
  35.           }

  37.           if (bytesRemaining != 0) {                                     
  38.             int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
  39.             splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining,
  40.                        blkLocations[blkIndex].getHosts()));
  41.           }
  42.         } else { // not splitable                                            //如果不分片，则直接将文件放入到List中，起始位置为0，大小为文件总大小。                                                                            
  43.           splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts()));
  44.         }
  45.       } else {
  46.         //Create empty hosts array for zero length files
  47.         splits.add(makeSplit(path, 0, length, new String[0]));
  48.       }
  49.     }
  50.     // Save the number of input files for metrics/loadgen
  51.     job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
  52.     LOG.debug("Total # of splits: " + splits.size());
            return splits;
      }
 三.createRecordReader方法：
       该方法返回一个RecordReader对象。以支持分片的LineRecordReader为例：
       该对象有四个重要方法：
        1).initialize
        2).nextKeyValue
        3).getCurrentKey
        4).getCurrentValue

     1.initialize方法：     
  1. public void initialize(InputSplit genericSplit,
  2.                          TaskAttemptContext context) throws IOException {
  3.     FileSplit split = (FileSplit) genericSplit;
  4.     Configuration job = context.getConfiguration();
  5.     this.maxLineLength = job.getInt(MAX_LINE_LENGTH, Integer.MAX_VALUE);
  6.     start = split.getStart();                                            //获取起始下标
  7.     end = start + split.getLength();                                     //获取大小
  8.     final Path file = split.getPath();

  10.     // open the file and seek to the start of the split
  11.     final FileSystem fs = file.getFileSystem(job);
  12.     fileIn = fs.open(file);                      
  13.     
  14.     CompressionCodec codec = new CompressionCodecFactory(job).getCodec(file);   //根据文件后缀名获取相应解码器。 如果输入文件为压缩文件则会自动获取。
  15.     if (null!=codec) {                                                         //如果是压缩文件
  16.       isCompressedInput = true;    
  17.       decompressor = CodecPool.getDecompressor(codec);
  18.       if (codec instanceof SplittableCompressionCodec) {
  19.         final SplitCompressionInputStream cIn =
  20.           ((SplittableCompressionCodec)codec).createInputStream(
  21.             fileIn, decompressor, start, end,
  22.             SplittableCompressionCodec.READ_MODE.BYBLOCK);
  23.         if (null == this.recordDelimiterBytes){                       //recordDelimiterBytes为终止符，如果读取一行时遇到该符号则结束。
  24.           in = new LineReader(cIn, job);                              //获取到LineReader   LineReader封装了一个InputStream
  25.         } else {
  26.           in = new LineReader(cIn, job, this.recordDelimiterBytes);
  27.         }

  29.         start = cIn.getAdjustedStart();
  30.         end = cIn.getAdjustedEnd();
  31.         filePosition = cIn;
  32.       } else {
  33.         if (null == this.recordDelimiterBytes) {
  34.           in = new LineReader(codec.createInputStream(fileIn, decompressor),
  35.               job);
  36.         } else {
  37.           in = new LineReader(codec.createInputStream(fileIn,
  38.               decompressor), job, this.recordDelimiterBytes);
  39.         }
  40.         filePosition = fileIn;
  41.       }
  42.     } else {
  43.       fileIn.seek(start);
  44.       if (null == this.recordDelimiterBytes){
  45.         in = new LineReader(fileIn, job);
  46.       } else {
  47.         in = new LineReader(fileIn, job, this.recordDelimiterBytes);
  48.       }

  50.       filePosition = fileIn;
  51.     }
  52.     // If this is not the first split, we always throw away first record
  53.     // because we always (except the last split) read one extra line in
  54.     // next() method.
  55.     if (start != 0) {
  56.       start += in.readLine(new Text(), 0, maxBytesToConsume(start));
  57.     }
  58.     this.pos = start;
  59.   }
  2.nextKeyValue方法：
 
  1. public boolean nextKeyValue() throws IOException {
  2.     if (key == null) {
  3.       key = new LongWritable();
  4.     }
  5.     key.set(pos);                                                       //pos默认值为FileSplit的start，也就是行号。
  6.     if (value == null) {
  7.       value = new Text();
  8.     }
  9.     int newSize = 0;
  10.     // We always read one extra line, which lies outside the upper
  11.     // split limit i.e. (end - 1)
  12.     while (getFilePosition() <= end) {
  13.       newSize = in.readLine(value, maxLineLength,                      //获取到value
  14.           Math.max(maxBytesToConsume(pos), maxLineLength));
  15.       pos += newSize;
  16.       if (newSize < maxLineLength) {
  17.         break;
  18.       }

  20.       // line too long. try again
  21.       LOG.info("Skipped line of size " + newSize + " at pos " + 
                   (pos - newSize));
        }
        if (newSize == 0) {
          key = null;
          value = null;
          return false;
        } else {
          return true;
        }
      }
 3.getCurrentKey 和 getCurrentValue 为获取 key value

看一下map的run方法：
  1. public void run(Context context) throws IOException, InterruptedException {
  2.     setup(context);
  3.     try {
  4.       while (context.nextKeyValue()) {
  5.         map(context.getCurrentKey(), context.getCurrentValue(), context);
  6.       }
  7.     } finally {
  8.       cleanup(context);
  9.     }
  10.   }



可能不是很全面，主要目的为个人备忘



       

来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/29754888/viewspace-1249907/ ，如需转载，请注明出处，否则将追究法律责任。

0
0
上一篇： yarn-RM-AM-NM之间的通讯
下一篇： HBase的两种协处理器
请登录后发表评论 登录
全部评论
破棉袄

注册时间：2014-07-16

  • 博文量
    180
  • 访问量
    1158153

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593