hbase hfile文件格式 详解_hbase数据文件存储格式

option java_package = "com.bdifn.hbase.hfile.proto";
option java_outer_classname = "HFileProtos";
option java_generic_services = true;
option java_generate_equals_and_hash = true;
option optimize_for = SPEED;

message FileTrailerProto {
  optional uint64 file_info_offset = 1;   //fileInfo起始偏移量
  optional uint64 load_on_open_data_offset = 2; //加载到内存区域起始偏移量
  optional uint64 uncompressed_data_index_size = 3; //未压缩的数据索引大小
  optional uint64 total_uncompressed_bytes = 4;  //KeyValue未压缩的总大小
  optional uint32 data_index_count = 5;          //Root DataIndex 的个数,如果只有1级索引的话，往往也是datablock个数
  optional uint32 meta_index_count = 6;          //元数据索引个数
  optional uint64 entry_count = 7;               //KeyValue总个数
  optional uint32 num_data_index_levels = 8;     //数据索引的级别,
  optional uint64 first_data_block_offset = 9;   //第一个数据块的偏移量
  optional uint64 last_data_block_offset = 10;   //最后一个数据块的偏移量
  optional string comparator_class_name = 11;    //比较器类名
  optional uint32 compression_codec = 12;        //编码
  optional bytes encryption_key = 13;            //加密相关
}

public static void main(String[] args) throws Exception {
Configuration config = new Configuration();

// 我已经将文件拷到了f盘根目录
String pathStr = "file:///f:/0a99d83b2b0a49c0adbc371d4bfe021e";
Path path = new Path(pathStr);
FileSystem fs = FileSystem.get(URI.create(pathStr), config);

FSDataInputStream input = fs.open(path);

long length = input.available();
int trailerSize = 212;

input.seek(length - trailerSize);
byte[] trailerBytes = new byte[trailerSize];
input.read(trailerBytes);

ByteBuffer trailerBuf = ByteBuffer.wrap(trailerBytes);
trailerBuf.position(trailerSize - 4);

int version = trailerBuf.getInt();
//3, 0, 0, 2//最后三位是majorVersion
int majorVersion = version & 0x00ffffff;
//高位是 minorVersion
int minorVersion = version >>> 24;

String magicCode = Bytes.toString(Arrays.copyOfRange(trailerBytes, 0, 8));

// 除去头8个字节MagicCode ,除去尾4个字节version信息。咱就是这么暴力。
FileTrailerProto hfileProtos = FileTrailerProto.PARSER
.parseDelimitedFrom(new ByteArrayInputStream(trailerBytes, 8,
trailerSize - 4));
System.out.println(String.format("MagicCode:%s,majorVersion:%d,:minorVersion:%d",magicCode,majorVersion,minorVersion));
System.out.println(hfileProtos);
    }

public class MyHFileBlock {public static class Header {private String magicCode ;
int onDiskSizeWithoutHeader;
int unCompressBlockSize;
long prevBlockOffset;
byte checkSum;
int bytesPerChecksum;
int onDiskDataSizeWithHeader;
    }

private Header header;
private ByteBuffer blockBuf;
private byte [] checkSum ;

....
   public static class BlockIndexReader {public BlockIndexReader(MyHFileBlock block) {
....
      }
   public BlockIndexReader parseMultiLevel(int numEntries, String expectedMagicCode, int level) throws Exception {
     .....
   }
   .......
   }
}

public class MyHFileBlockIterator {private ByteBuffer loadOnOpenBuffer;

    public MyHFileBlockIterator(FSDataInputStream data, long offset, int length) {

try {
    data.seek(offset);
    byte[] loadOnOpenBytes = new byte[length];
    data.read(loadOnOpenBytes);
    loadOnOpenBuffer = ByteBuffer.wrap(loadOnOpenBytes);
} catch (IOException e) {
    e.printStackTrace();
}
    }

    public MyHFileBlockIterator(byte [] data) {
loadOnOpenBuffer = ByteBuffer.wrap(data);
    }
    
    public MyHFileBlock nextBlock() {

MyHFileBlock block = new MyHFileBlock(loadOnOpenBuffer);
Header header = block.getHeader();
int currentBlockLength = block.getHeader()
.getOnDiskDataSizeWithHeader();

int dataSize = currentBlockLength - MyHFileBlock.HARDER_SIZE;

byte[] dataBlockArray = new byte[dataSize];

loadOnOpenBuffer.get(dataBlockArray);

ByteBuffer dataBlock = ByteBuffer.wrap(dataBlockArray);
     
block.setBlockBuf(dataBlock);

int checkSumChunks = header.getOnDiskSizeWithoutHeader()
/ header.getBytesPerChecksum();
if (header.getOnDiskSizeWithoutHeader() % header.getBytesPerChecksum() != 0) {
    checkSumChunks++;
}
int checkSumBytes = checkSumChunks * 4;
byte[] checkSum = new byte[checkSumBytes];

loadOnOpenBuffer.get(checkSum);

block.setCheckSum(checkSum);

return block;
    }
    
    public boolean hasNext(){
return loadOnOpenBuffer.position() < loadOnOpenBuffer.capacity();
    }
}

option java_package = "com.bdifn.hbase.hfile.proto";
option java_outer_classname = "FileInfoProtos";
option java_generic_services = true;
option java_generate_equals_and_hash = true;
option optimize_for = SPEED;

message BytesBytesPair {
  required bytes first = 1;
  required bytes second = 2;
}

message FileInfoProto {
    repeated BytesBytesPair map_entry = 1;
}

....
FileTrailerProto hfileProtos = FileTrailerProto.PARSER.parseDelimitedFrom(new ByteArrayInputStream(trailerBytes, 8,trailerBytes.length - 4));
long loadOnOpenLength = length - trailerSize - hfileProtos.getLoadOnOpenDataOffset();
MyHFileBlockIterator inter = new MyHFileBlockIterator(input,hfileProtos.getLoadOnOpenDataOffset(), (int) loadOnOpenLength);
//解析出来root data index
MyHFileBlock dataIndex = inter.nextBlock();
int dataIndexLevels = hfileProtos.getNumDataIndexLevels();
int dataIndexEntries = hfileProtos.getDataIndexCount();
//创建root data index reader
MyHFileBlock.BlockIndexReader rootDataReader = dataIndex.createBlockIndexReader().parseMultiLevel(dataIndexEntries,"IDXROOT2", dataIndexLevels);
//解析出来root meta index
MyHFileBlock metaIndex = inter.nextBlock();
.....
//获取file info
MyHFileBlock fileInfo = inter.nextBlock();
//解析读取FileInfo内容
ByteArrayInputStream in = new ByteArrayInputStream(fileInfo.getBlockBuf().array());
int pblen = ProtobufUtil.lengthOfPBMagic();
byte[] pbuf = new byte[pblen];
if (in.markSupported())
in.mark(pblen);
int read = in.read(pbuf);
FileInfoProtos.FileInfoProto fileInfoProto = FileInfoProtos.FileInfoProto.parseDelimitedFrom(in);

List<BytesBytesPair> entries = fileInfoProto.getMapEntryList();

for (BytesBytesPair entry : entries) {
System.out.println(entry.getFirst().toStringUtf8() + ":"+ entry.getSecond().toStringUtf8());
}
//剩下的BloomFileter metadata block.
while (inter.hasNext()) {
MyHFileBlock block = inter.nextBlock();
System.out.println(block.getHeader());
}

public class KeyValueIterator {public static final int KEY_LENGTH_SIZE = 4;
public static final int VALUE_LENGTH_SIZE = 4;

private byte [] data ;
private int currentOffset ;

public KeyValueIterator(byte [] data) {
this.data = data;
currentOffset = 0;
}

public KeyValue nextKeyValue(){
KeyValue kv = null;
int keyLen = Bytes.toInt(data,currentOffset,4);
incrementOffset(KEY_LENGTH_SIZE);

int valueLen = Bytes.toInt(data,currentOffset,4);
incrementOffset(VALUE_LENGTH_SIZE);

//1 is memTS
incrementOffset(keyLen,valueLen,1);

int kvSize = KEY_LENGTH_SIZE + VALUE_LENGTH_SIZE +  keyLen + valueLen ;

kv = new KeyValue(data , currentOffset - kvSize - 1, kvSize);
return kv;
}
public void incrementOffset(int ... lengths) {
for(int length : lengths)
currentOffset = currentOffset + length;
}

public boolean hasNext() {
return currentOffset < data.length;
}
}

//从rootDataReader中获取第一块的offset及数据大小
long offset = rootDataReader.getBlockOffsets()[0];
int size = rootDataReader.getBlockDataSizes()[0];

byte[] dataBlockArray = new byte[size];
input.seek(offset);
input.read(dataBlockArray);
//图方便，直接用iterator来解析出来FileBlock
MyHFileBlockIterator dataBlockIter = new MyHFileBlockIterator(dataBlockArray);
MyHFileBlock dataBlock1 = dataBlockIter.nextBlock();
//将data内容给一个keyvalue迭代器
KeyValueIterator kvIter = new KeyValueIterator(dataBlock1.getBlockBuf().array());
while (kvIter.hasNext()) {
    KeyValue kv = kvIter.nextKeyValue();
    //do some with keyvalue. like print the kv.
    System.out.println(kv);
}