/**

    * Generate the list of files and make them into FileSplits.

    */

   public List<InputSplit> getSplits(JobContext job

                                     ) throws IOException {

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

     long maxSize = getMaxSplitSize(job);    //Long.MAX_VALUE

     // generate splits

     List<InputSplit> splits = new ArrayList<InputSplit>();

     List<FileStatus>files = listStatus(job);

     for (FileStatus file: files) {

       Path path = file.getPath();

       FileSystem fs = path.getFileSystem(job.getConfiguration());

       long length = file.getLen();    //整个文件的长度

       BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);

       if ((length != 0) && isSplitable(job, path)) {    //默认是true，但是如果是压缩的，则是false

         long blockSize = file.getBlockSize();        //64M，67108864B

         long splitSize = computeSplitSize(blockSize, minSize, maxSize);    //计算split大小  Math.max(minSize, Math.min(maxSize, blockSize))

         long bytesRemaining = length;

         while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {

           int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);

           splits.add(new FileSplit(path, length-bytesRemaining, splitSize,

                                    blkLocations[blkIndex].getHosts()));        //hosts是主机名,name是IP

           bytesRemaining -= splitSize;        //剩余块的大小

         }

         if (bytesRemaining != 0) {    //最后一个

           splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining,

                      blkLocations[blkLocations.length-1].getHosts()));

         }

       } else if (length != 0) {    //isSplitable(job, path)等于false

         splits.add(new FileSplit(path, 0, length, blkLocations[0].getHosts()));

       } else {

         //Create empty hosts array for zero length files

         splits.add(new FileSplit(path, 0, length, new String[0]));

       }

     }

     // Save the number of input files in the job-conf

     job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());

     LOG.debug("Total # of splits: " + splits.size());

     return splits;

   }

  protected int getBlockIndex(BlockLocation[] blkLocations,

                               long offset) {

     for (int i = 0 ; i < blkLocations.length; i++) {

       // is the offset inside this block?

       if ((blkLocations[i].getOffset() <= offset) &&

           (offset < blkLocations[i].getOffset() + blkLocations[i].getLength())){

         return i;

       }

     }

     BlockLocation last = blkLocations[blkLocations.length -1];

     long fileLength = last.getOffset() + last.getLength() -1;

     throw new IllegalArgumentException("Offset " + offset +

                                        " is outside of file (0.." +

                                        fileLength + ")");

   }

  /**

    * Read one line from the InputStream into the given Text.  A line

    * can be terminated by one of the following: '\n' (LF) , '\r' (CR),

    * or '\r\n' (CR+LF).  EOF also terminates an otherwise unterminated

    * line.

    *

    * @param str the object to store the given line (without newline)

    * @param maxLineLength the maximum number of bytes to store into str;

    *  the rest of the line is silently discarded.

    * @param maxBytesToConsume the maximum number of bytes to consume

    *  in this call.  This is only a hint, because if the line cross

    *  this threshold, we allow it to happen.  It can overshoot

    *  potentially by as much as one buffer length.

    *

    * @return the number of bytes read including the (longest) newline

    * found.

    *

    * @throws IOException if the underlying stream throws

    */

   public int readLine(Text str, int maxLineLength,

                       int maxBytesToConsume) throws IOException {

     /* We're reading data from in, but the head of the stream may be

      * already buffered in buffer, so we have several cases:

      * 1. No newline characters are in the buffer, so we need to copy

      *    everything and read another buffer from the stream.

      * 2. An unambiguously terminated line is in buffer, so we just

      *    copy to str.

      * 3. Ambiguously terminated line is in buffer, i.e. buffer ends

      *    in CR.  In this case we copy everything up to CR to str, but

      *    we also need to see what follows CR: if it's LF, then we

      *    need consume LF as well, so next call to readLine will read

      *    from after that.

      * We use a flag prevCharCR to signal if previous character was CR

      * and, if it happens to be at the end of the buffer, delay

      * consuming it until we have a chance to look at the char that

      * follows.

      */

     str.clear();

     int txtLength = 0; //tracks str.getLength(), as an optimization

     int newlineLength = 0; //length of terminating newline

     boolean prevCharCR = false; //true of prev char was CR

     long bytesConsumed = 0;

     do {

       int startPosn = bufferPosn; //starting from where we left off the last time

       if (bufferPosn >= bufferLength) {

         startPosn = bufferPosn = 0;

         if (prevCharCR)

           ++bytesConsumed; //account for CR from previous read

         bufferLength = in.read(buffer);    //从输入流中读取一定数量的字节，并将其存储在缓冲区数组 b 中。以整数形式返回实际读取的字节数。

         if (bufferLength <= 0)        //结束了，没数据了

           break; // EOF

       }

       //'\n'，ASCII码：10，意义：换行NL；；；；'\r' ，ASCII码：13，意义： 回车CR

       for (; bufferPosn < bufferLength; ++bufferPosn) { //search for newline

         if (buffer[bufferPosn] == LF) {            //如果是换行字符\n

           newlineLength = (prevCharCR) ? 2 : 1;

           ++bufferPosn; // at next invocation proceed from following byte，越过换行字符

           break;

         }

         if (prevCharCR) { //CR + notLF, we are at notLF，如果是回车字符\r

           newlineLength = 1;

           break;

         }

         prevCharCR = (buffer[bufferPosn] == CR);

       }

       int readLength = bufferPosn - startPosn;

       if (prevCharCR && newlineLength == 0)    //表示还没遇到换行，有回车字符，且缓存最后一个是\r

         --readLength; //CR at the end of the buffer

       bytesConsumed += readLength;

       int appendLength = readLength - newlineLength;    //newlineLength换行符个数

       if (appendLength > maxLineLength - txtLength) {

         appendLength = maxLineLength - txtLength;

       }

       if (appendLength > 0) {

         str.append(buffer, startPosn, appendLength);    //将数据加入str

         txtLength += appendLength;

       }

     } while (newlineLength == 0 && bytesConsumed < maxBytesToConsume);//循环条件没有换行并且没超过上限

     if (bytesConsumed > (long)Integer.MAX_VALUE)

       throw new IOException("Too many bytes before newline: " + bytesConsumed);

     return (int)bytesConsumed;

   }

 public boolean nextKeyValue() throws IOException {

     if (key == null) {

       key = new LongWritable();

     }

     key.set(pos);

     if (value == null) {

       value = new Text();

     }

     int newSize = 0;

     while (pos < end) {

       newSize = in.readLine(value, maxLineLength,

                             Math.max((int)Math.min(Integer.MAX_VALUE, end-pos),

                                      maxLineLength));

       if (newSize == 0) {

         break;

       }

       pos += newSize;

       if (newSize < maxLineLength) {

         break;

       }

       // line too long. try again

       LOG.info("Skipped line of size " + newSize + " at pos " +

                (pos - newSize));

     }

     if (newSize == 0) {

       key = null;

       value = null;

       return false;

     } else {

       return true;

     }

   }