this.bytes = createByteArray(row, roffset, rlength,

        family, foffset, flength, qualifier, qoffset, qlength,

        timestamp, type, value, voffset, vlength);

KeyValue维护了一个byte数组，里面保存了这个KeyValue要写入HFile的全部信息。

/**

   * Write KeyValue format into a byte array.

   *

   * @param row row key

   * @param roffset row offset

   * @param rlength row length

   * @param family family name

   * @param foffset family offset

   * @param flength family length

   * @param qualifier column qualifier

   * @param qoffset qualifier offset

   * @param qlength qualifier length

   * @param timestamp version timestamp

   * @param type key type

   * @param value column value

   * @param voffset value offset

   * @param vlength value length

   * @return The newly created byte array.

   */

  private static byte [] createByteArray(final byte [] row, final int roffset,

      final int rlength, final byte [] family, final int foffset, int flength,

      final byte [] qualifier, final int qoffset, int qlength,

      final long timestamp, final Type type,

      final byte [] value, final int voffset, int vlength) {

    checkParameters(row, rlength, family, flength, qualifier, qlength, value, vlength);

    // Allocate right-sized byte array.

    int keyLength = (int) getKeyDataStructureSize(rlength, flength, qlength);

    byte [] bytes =

        new byte[(int) getKeyValueDataStructureSize(rlength, flength, qlength, vlength)];

    // Write key, value and key row length.

    int pos = 0;

    pos = Bytes.putInt(bytes, pos, keyLength);//4B的KeyLength

    pos = Bytes.putInt(bytes, pos, vlength);//4B valueLength

    pos = Bytes.putShort(bytes, pos, (short)(rlength & 0x0000ffff));//2B RowLength

    pos = Bytes.putBytes(bytes, pos, row, roffset, rlength);//Row内容

    pos = Bytes.putByte(bytes, pos, (byte)(flength & 0x0000ff));//1B FamilyLength

    if(flength != 0) {

      pos = Bytes.putBytes(bytes, pos, family, foffset, flength);

    }//将famliy内容加载到bytes

    if(qlength != 0) {

      pos = Bytes.putBytes(bytes, pos, qualifier, qoffset, qlength);

    }//将qualifer内容加载到bytes

    pos = Bytes.putLong(bytes, pos, timestamp);//timestamp

    pos = Bytes.putByte(bytes, pos, type.getCode());//type

    if (value != null && value.length > 0) {

      pos = Bytes.putBytes(bytes, pos, value, voffset, vlength);

    }//value

    return bytes;

  }

private void append(final long memstoreTS, final byte[] key,

            final int koffset, final int klength, final byte[] value,

            final int voffset, final int vlength) throws IOException {

       boolean dupKey = checkKey(key, koffset, klength);//校验是否符合规则

        if (!dupKey) {

            checkBlockBoundary();

        }//检查边界

        if (!fsBlockWriter.isWriting())

            newBlock();//第一次写入，会进行这个操作，之后在checkBlockBoundary()内执行此操作

        {

            DataOutputStream out = fsBlockWriter.getUserDataStream();

            out.writeInt(klength);

            totalKeyLength += klength;

            out.writeInt(vlength);

            totalValueLength += vlength;

            out.write(key, koffset, klength);

            out.write(value, voffset, vlength);

            if (this.includeMemstoreTS) {

                WritableUtils.writeVLong(out, memstoreTS);

            }

        }

        // Are we the first key in this block?

        if (firstKeyInBlock == null) {

            // Copy the key.

            firstKeyInBlock = new byte[klength];

            System.arraycopy(key, koffset, firstKeyInBlock, 0, klength);

        }

        lastKeyBuffer = key;//记录最后一个key

        lastKeyOffset = koffset;

        lastKeyLength = klength;

        entryCount++;

    }

/**

   * Put the header into the given byte array at the given offset.

   * @param onDiskSize size of the block on disk header + data + checksum

   * @param uncompressedSize size of the block after decompression (but

   *          before optional data block decoding) including header

   * @param onDiskDataSize size of the block on disk with header

   *        and data but not including the checksums

   */

  private void putHeader(byte[] dest, int offset, int onDiskSize,

      int uncompressedSize, int onDiskDataSize) {

    offset = blockType.put(dest, offset);//DATABLK*(8B)

    offset = Bytes.putInt(dest, offset, onDiskSize - HConstants.HFILEBLOCK_HEADER_SIZE);//在磁盘上block的大小（不包含头，主要用来处理压缩文件）

    offset = Bytes.putInt(dest, offset, uncompressedSize - HConstants.HFILEBLOCK_HEADER_SIZE);//未压缩时block的大小（不包含头）

    offset = Bytes.putLong(dest, offset, prevOffset);//前一个块的标识

    offset = Bytes.putByte(dest, offset, checksumType.getCode());//校验和类型

    offset = Bytes.putInt(dest, offset, bytesPerChecksum);

    Bytes.putInt(dest, offset, onDiskDataSize);//

  }

/**

* Add one index entry to the current leaf-level block. When the leaf-level

* block gets large enough, it will be flushed to disk as an inline block.

*

* @param firstKey the first key of the data block

* @param blockOffset the offset of the data block

* @param blockDataSize the on-disk size of the data block ({@link HFile}

* format version 2), or the uncompressed size of the data block (

* {@link HFile} format version 1).

*/

public void addEntry(byte[] firstKey, long blockOffset, int blockDataSize)

{

curInlineChunk.add(firstKey, blockOffset, blockDataSize);

++totalNumEntries;

}

key=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaZI/info:city/LATEST_TIMESTAMP/Put

  offset=2155038, dataSize=38135

key=aaaaaaaaaaaaaaaaabbbabbaaaabbbbaTU/info:city/LATEST_TIMESTAMP/Put

  offset=4343199, dataSize=38358

key=aaaaaaaaaaaaaaaabbbababbbbababaaxe1T70L5R6F2/info:city/LATEST_TIMESTAMP/Put

  offset=6539008, dataSize=38081

.....

while (rootChunk.getRootSize() > maxChunkSize) {

    rootChunk = writeIntermediateLevel(out, rootChunk);

    numLevels += 1;

}

public static FixedFileTrailer readFromStream(FSDataInputStream istream,

long fileSize) throws IOException {

    int bufferSize = MAX_TRAILER_SIZE;

    long seekPoint = fileSize - bufferSize; //trailer 位置

    if (seekPoint < 0) {

    // It is hard to imagine such a small HFile.

        seekPoint = 0;

        bufferSize = (int) fileSize;

    }

    //加载trailer

    istream.seek(seekPoint);

    ByteBuffer buf = ByteBuffer.allocate(bufferSize);

    istream.readFully(buf.array(), buf.arrayOffset(),

    buf.arrayOffset() + buf.limit());

    .....

    FixedFileTrailer fft = new FixedFileTrailer(majorVersion, minorVersion);

    fft.deserialize(new DataInputStream(new ByteArrayInputStream(buf.array(),

    buf.arrayOffset() + bufferSize - trailerSize, trailerSize)));

    return fft;

}

key=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaZI/info:city/LATEST_TIMESTAMP/Put

  offset=3944715, dataSize=38949

key=aaaaaaaaaaaaaaaabbbaaabbaabaabbaiSje/info:city/LATEST_TIMESTAMP/Put

  offset=6825203, dataSize=27845

public static int binarySearch(byte [][]arr, byte []key, int offset,

      int length, RawComparator<?> comparator) {

    int low = 0;

    int high = arr.length - 1;

    while (low <= high) {

      int mid = (low+high) >>> 1;

      // we have to compare in this order, because the comparator order

      // has special logic when the 'left side' is a special key.

      int cmp = comparator.compare(key, offset, length,

          arr[mid], 0, arr[mid].length);

      // key lives above the midpoint

      if (cmp > 0)

        low = mid + 1;

      // key lives below the midpoint

      else if (cmp < 0)

        high = mid - 1;

      // BAM. how often does this really happen?

      else

        return mid;

    }

    return - (low+1);

  }