MapReduce中TextInputFormat分片和读取分片数据源码级分析

 1  /** 
 2    * Generate the list of files and make them into FileSplits.
 3    */ 
 4   public List<InputSplit> getSplits(JobContext job
 5                                     ) throws IOException {
 6     long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
 7     long maxSize = getMaxSplitSize(job);    //Long.MAX_VALUE
 8 
 9     // generate splits
10     List<InputSplit> splits = new ArrayList<InputSplit>();
11     List<FileStatus>files = listStatus(job);
12     for (FileStatus file: files) {
13       Path path = file.getPath();
14       FileSystem fs = path.getFileSystem(job.getConfiguration());
15       long length = file.getLen();    //整个文件的长度
16       BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
17       if ((length != 0) && isSplitable(job, path)) {    //默认是true，但是如果是压缩的，则是false
18         long blockSize = file.getBlockSize();        //64M，67108864B
19         long splitSize = computeSplitSize(blockSize, minSize, maxSize);    //计算split大小  Math.max(minSize, Math.min(maxSize, blockSize))
20 
21         long bytesRemaining = length;            
22         while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
23           int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
24           splits.add(new FileSplit(path, length-bytesRemaining, splitSize, 
25                                    blkLocations[blkIndex].getHosts()));        //hosts是主机名,name是IP
26           bytesRemaining -= splitSize;        //剩余块的大小
27         }
28         
29         if (bytesRemaining != 0) {    //最后一个
30           splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining, 
31                      blkLocations[blkLocations.length-1].getHosts()));
32         }
33       } else if (length != 0) {    //isSplitable(job, path)等于false
34         splits.add(new FileSplit(path, 0, length, blkLocations[0].getHosts()));
35       } else {
36         //Create empty hosts array for zero length files
37         splits.add(new FileSplit(path, 0, length, new String[0]));
38       }
39     }
40     
41     // Save the number of input files in the job-conf
42     job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
43 
44     LOG.debug("Total # of splits: " + splits.size());
45     return splits;
46   }

 1  protected int getBlockIndex(BlockLocation[] blkLocations, 
 2                               long offset) {
 3     for (int i = 0 ; i < blkLocations.length; i++) {
 4       // is the offset inside this block?
 5       if ((blkLocations[i].getOffset() <= offset) &&
 6           (offset < blkLocations[i].getOffset() + blkLocations[i].getLength())){
 7         return i;
 8       }
 9     }
10     BlockLocation last = blkLocations[blkLocations.length -1];
11     long fileLength = last.getOffset() + last.getLength() -1;
12     throw new IllegalArgumentException("Offset " + offset + 
13                                        " is outside of file (0.." +
14                                        fileLength + ")");
15   }

 1  /**
 2    * Read one line from the InputStream into the given Text.  A line
 3    * can be terminated by one of the following: ‘\n‘ (LF) , ‘\r‘ (CR),
 4    * or ‘\r\n‘ (CR+LF).  EOF also terminates an otherwise unterminated
 5    * line.
 6    *
 7    * @param str the object to store the given line (without newline)
 8    * @param maxLineLength the maximum number of bytes to store into str;
 9    *  the rest of the line is silently discarded.
10    * @param maxBytesToConsume the maximum number of bytes to consume
11    *  in this call.  This is only a hint, because if the line cross
12    *  this threshold, we allow it to happen.  It can overshoot
13    *  potentially by as much as one buffer length.
14    *
15    * @return the number of bytes read including the (longest) newline
16    * found.
17    *
18    * @throws IOException if the underlying stream throws
19    */
20   public int readLine(Text str, int maxLineLength,
21                       int maxBytesToConsume) throws IOException {
22     /* We‘re reading data from in, but the head of the stream may be
23      * already buffered in buffer, so we have several cases:
24      * 1. No newline characters are in the buffer, so we need to copy
25      *    everything and read another buffer from the stream.
26      * 2. An unambiguously terminated line is in buffer, so we just
27      *    copy to str.
28      * 3. Ambiguously terminated line is in buffer, i.e. buffer ends
29      *    in CR.  In this case we copy everything up to CR to str, but
30      *    we also need to see what follows CR: if it‘s LF, then we
31      *    need consume LF as well, so next call to readLine will read
32      *    from after that.
33      * We use a flag prevCharCR to signal if previous character was CR
34      * and, if it happens to be at the end of the buffer, delay
35      * consuming it until we have a chance to look at the char that
36      * follows.
37      */
38     str.clear();
39     int txtLength = 0; //tracks str.getLength(), as an optimization
40     int newlineLength = 0; //length of terminating newline
41     boolean prevCharCR = false; //true of prev char was CR
42     long bytesConsumed = 0;
43     do {
44       int startPosn = bufferPosn; //starting from where we left off the last time
45       if (bufferPosn >= bufferLength) {
46         startPosn = bufferPosn = 0;
47         if (prevCharCR)
48           ++bytesConsumed; //account for CR from previous read
49         bufferLength = in.read(buffer);    //从输入流中读取一定数量的字节，并将其存储在缓冲区数组 b 中。以整数形式返回实际读取的字节数。
50         if (bufferLength <= 0)        //结束了，没数据了
51           break; // EOF
52       }
53       //‘\n‘，ASCII码：10，意义：换行NL；；；；‘\r‘ ，ASCII码：13，意义： 回车CR
54       for (; bufferPosn < bufferLength; ++bufferPosn) { //search for newline
55         if (buffer[bufferPosn] == LF) {            //如果是换行字符\n
56           newlineLength = (prevCharCR) ? 2 : 1;
57           ++bufferPosn; // at next invocation proceed from following byte，越过换行字符
58           break;
59         }
60         if (prevCharCR) { //CR + notLF, we are at notLF，如果是回车字符\r
61           newlineLength = 1;
62           break;
63         }
64         prevCharCR = (buffer[bufferPosn] == CR);
65       }
66       int readLength = bufferPosn - startPosn;
67       if (prevCharCR && newlineLength == 0)    //表示还没遇到换行，有回车字符，且缓存最后一个是\r
68         --readLength; //CR at the end of the buffer
69       bytesConsumed += readLength;
70       int appendLength = readLength - newlineLength;    //newlineLength换行符个数
71       if (appendLength > maxLineLength - txtLength) {
72         appendLength = maxLineLength - txtLength;
73       }
74       if (appendLength > 0) {
75         str.append(buffer, startPosn, appendLength);    //将数据加入str
76         txtLength += appendLength;
77       }
78     } while (newlineLength == 0 && bytesConsumed < maxBytesToConsume);//循环条件没有换行并且没超过上限
79 
80     if (bytesConsumed > (long)Integer.MAX_VALUE)
81       throw new IOException("Too many bytes before newline: " + bytesConsumed);    
82     return (int)bytesConsumed;
83   }

 1 public boolean nextKeyValue() throws IOException {
 2     if (key == null) {
 3       key = new LongWritable();
 4     }
 5     key.set(pos);
 6     if (value == null) {
 7       value = new Text();
 8     }
 9     int newSize = 0;
10     while (pos < end) {
11       newSize = in.readLine(value, maxLineLength,
12                             Math.max((int)Math.min(Integer.MAX_VALUE, end-pos),
13                                      maxLineLength));
14       if (newSize == 0) {
15         break;
16       }
17       pos += newSize;
18       if (newSize < maxLineLength) {
19         break;
20       }
21 
22       // line too long. try again
23       LOG.info("Skipped line of size " + newSize + " at pos " + 
24                (pos - newSize));
25     }
26     if (newSize == 0) {
27       key = null;
28       value = null;
29       return false;
30     } else {
31       return true;
32     }
33   }