自定义 Hadoop 的输入格式

  public boolean nextKeyValue() throws IOException {    if (key == null) {      key = new LongWritable();    }    key.set(pos);    if (value == null) {      value = new Text();    }    int newSize = 0;            // flag为是否清空的标记，保证每隔两条清空一次    boolean flag=true;    //curValue 为读取到的最新的一行，value为这次循环读取的所有行的不断积累    Text curValue = new Text();    int cycleCount = 0;    do{        if (curValue.getLength()==0) {        cycleCount++;    }else {      cycleCount = 0;    }        while (getFilePosition() <= end || in.needAdditionalRecordAfterSplit()) {          if (pos == 0) {            newSize = skipUtfByteOrderMark();          } else {            //重写readLine方法，传入curValue和flag            newSize = in.readLine(value, maxLineLength, maxBytesToConsume(pos),curValue,flag);            pos += newSize;          }
          if ((newSize == 0) || (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;    }  }

  if(flag){    str.clear();  }  curValue.clear();  int txtLength = 0; // tracks str.getLength(), as an optimization  long bytesConsumed = 0;  int delPosn = 0;  int ambiguousByteCount=0; // To capture the ambiguous characters count  do {    int startPosn = bufferPosn; // Start from previous end position    if (bufferPosn >= bufferLength) {      startPosn = bufferPosn = 0;      bufferLength = fillBuffer(in, buffer, ambiguousByteCount > 0);      if (bufferLength <= 0) {        if (ambiguousByteCount > 0) {          str.append(recordDelimiterBytes, 0, ambiguousByteCount);          curValue.append(recordDelimiterBytes, 0, ambiguousByteCount);          bytesConsumed += ambiguousByteCount;        }        break; // EOF      }    }    for (; bufferPosn < bufferLength; ++bufferPosn) {      if (buffer[bufferPosn] == recordDelimiterBytes[delPosn]) {        delPosn++;        if (delPosn >= recordDelimiterBytes.length) {          bufferPosn++;          break;        }      } else if (delPosn != 0) {        bufferPosn--;        delPosn = 0;      }    }    int readLength = bufferPosn - startPosn;    bytesConsumed += readLength;    int appendLength = readLength - delPosn;    if (appendLength > maxLineLength - txtLength) {      appendLength = maxLineLength - txtLength;    }    bytesConsumed += ambiguousByteCount;    if (appendLength >= 0 && ambiguousByteCount > 0) {      //appending the ambiguous characters (refer case 2.2)      str.append(recordDelimiterBytes, 0, ambiguousByteCount);      curValue.append(recordDelimiterBytes, 0, ambiguousByteCount);      ambiguousByteCount = 0;      // since it is now certain that the split did not split a delimiter we      // should not read the next record: clear the flag otherwise duplicate      // records could be generated      unsetNeedAdditionalRecordAfterSplit();    }    if (appendLength > 0) {      str.append(buffer, startPosn, appendLength);      curValue.append(buffer, startPosn, appendLength);      txtLength += appendLength;    }    if (bufferPosn >= bufferLength) {      if (delPosn > 0 && delPosn < recordDelimiterBytes.length) {        ambiguousByteCount = delPosn;        bytesConsumed -= ambiguousByteCount; //to be consumed in next      }    }  } while (delPosn < recordDelimiterBytes.length       && bytesConsumed < maxBytesToConsume);  if (bytesConsumed > Integer.MAX_VALUE) {    throw new IOException("Too many bytes before delimiter: " + bytesConsumed);  }  return (int) bytesConsumed; }

/** * 重载readDefaultLine方法 */private int readDefaultLine(Text str, int maxLineLength, int maxBytesToConsume,Text curValue,boolean flag)    throws IOException {  if(flag){    str.clear();  }  curValue.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 = fillBuffer(in, buffer, prevCharCR);      if (bufferLength <= 0) {        break; // EOF      }    }    for (; bufferPosn < bufferLength; ++bufferPosn) { //search for newline      if (buffer[bufferPosn] == LF) {        newlineLength = (prevCharCR) ? 2 : 1;        ++bufferPosn; // at next invocation proceed from following byte        break;      }      if (prevCharCR) { //CR + notLF, we are at notLF        newlineLength = 1;        break;      }      prevCharCR = (buffer[bufferPosn] == CR);    }    int readLength = bufferPosn - startPosn;    if (prevCharCR && newlineLength == 0) {      --readLength; //CR at the end of the buffer    }    bytesConsumed += readLength;    int appendLength = readLength - newlineLength;    if (appendLength > maxLineLength - txtLength) {      appendLength = maxLineLength - txtLength;    }    if (appendLength > 0) {      str.append(buffer, startPosn, appendLength);      curValue.append(buffer, startPosn, appendLength);      txtLength += appendLength;    }  } while (newlineLength == 0 && bytesConsumed < maxBytesToConsume);
  if (bytesConsumed > Integer.MAX_VALUE) {    throw new IOException("Too many bytes before newline: " + bytesConsumed);  }  return (int)bytesConsumed;}