Hadoop SequenceFile 详解

Hadoop中有两种基本的文件类型, 一个是SequenceFile, 一个MapFile

他们通过底层 I/O接口 FSDataOutputStream 来进行写入, 对外操作起来就像普通的文件操作.

两者的相同之处在于都存放了很多键值对

两者的不同在于MapFile是SequenceFile的升级版, 它总是根据key来排序, 支持通过key来查讯value

而SequenceFile的读取则是从第一个键值对一个一个往下读, 直到返回空值, 有点象读数据库的感觉.

SequenceFile的一个特点在在于有一个sync_maker, 这个sync_maker帮助SequenceFile型文件支持随机读取.

在未压缩的SequenceFile的文件结构是

• Header

• Record
  • Record length
  • Key length
  • Key
  • (Compressed?) Value
• A sync-marker every few k bytes or so.

sync_maker的结构和生成如下:

    byte[] sync;                          // 16 random bytes{try {                                       MessageDigest digester = MessageDigest.getInstance("MD5");long time = Time.now();digester.update((new UID()+"@"+time).getBytes());sync = digester.digest();} catch (Exception e) {throw new RuntimeException(e);}}

    public void sync() throws IOException {if (sync != null && lastSyncPos != out.getPos()) {out.writeInt(SYNC_ESCAPE);                // mark the start of the syncout.write(sync);                          // write synclastSyncPos = out.getPos();               // update lastSyncPos}}

在SequenceFile的header里会存放一个sync_maker的值, 指的就是head最后会有一个sync_marker

SequenceFile文件支持从文件的任意位置开始随机读取, 就是通过在文件的字节流里找这个全局唯一的sync_marker

所以SequenceFileRead里的sync函数是这样定义的:

sync

public void sync(long position)throws IOException

Seek to the next sync mark past a given position.

Throws:

IOException

实现逻辑是从任意位置找起,然后寻找和sync_marker一样的bytes[]

    public synchronized void sync(long position) throws IOException {if (position+SYNC_SIZE >= end) {seek(end);return;}if (position < headerEnd) {// seek directly to first recordin.seek(headerEnd);// note the sync marker "seen" in the headersyncSeen = true;return;}try {seek(position+4);                         // skip escapein.readFully(syncCheck);int syncLen = sync.length;for (int i = 0; in.getPos() < end; i++) {int j = 0;for (; j < syncLen; j++) {if (sync[j] != syncCheck[(i+j)%syncLen])break;}if (j == syncLen) {in.seek(in.getPos() - SYNC_SIZE);     // position before syncreturn;}syncCheck[i%syncLen] = in.readByte();}} catch (ChecksumException e) {             // checksum failurehandleChecksumException(e);}}