HDFS 中常见的数据格式 1.面向行的文件格式介绍

1. 概述

hadoop 中的文件格式大致分为面向行与面向列两类

1. 面向行

• 同一行的数据存储在一起， 即连续存储， 如 SequenceFile、 MapFile、Avro DataFile 都采用面向行的方式存储

• 如果只需要访问行的一小部分数据， 也需要将整行读入到内存， 推迟序列化程度可以缓解这个问题， 但是从磁盘读取整行数据的开销却无法避免，

• 面向行的存储更适合整行数据需要同时被处理的情况

2. 面向列

• 整个文件被切割成若干列数据， 每列数据存储在一起 如 Parquet、 RCFile、 ORCFile 都采用面向列的方式存储。
• 面向列的格式可以在读取数据时跳过不需要的列， 适合用于只处理行的小部分字段的情况， 但是这种格式读取需要更多的内存， 因为需要缓存行在内存( 为了获取多行的某一列 )。
• 不适合流式写入， 因为一旦写入失败， 当前文件无法恢复， 而面向行的数据在写入失败时， 可以重新同步到最后一个同步点， 所以flume采取的是面向行的存储格式

2. 面向行的数据类型

2.1 SequenceFile

• SequenceFile 是hadoop用来存储二进制形式的key-value 对而设计的一种平面文件( Flat File )。

• 目前也有人在该文件格式的基础上提出了一些HDFS中小文件存储的解决方案， 他们的基本思路就是将小文件进行合并成一个大文件， 同时对这些小文件的位置信息构建索引， 这类解决方案还设计到hadoop的另一种文件格式 MapFile，

• SequenceFile 文件并不保证其存储的 key-value 数据是按照key的某个顺序存储的， 同时不支持 append 操作

2.1.1 压缩类型

在SequenceFile文件中， 每个key-value被看做成是一条记录( Record )， 因此基于Record的压缩策略 ， SequenceFile文件 可以支持三种压缩类型 ( SequenceFile.CompressionType )

1. NONE：对Records 不进行压缩
2. RECORD：仅压缩每个record中的value 值
3. BLOCK：将block中的所有records压缩在一起

2.1.2 压缩类型对应Writer

那么基于这三种压缩类型， hadoop 提供了对应的三种类型的Writer：

2.1.2.1 不进行压缩

SequenceFile.Write 写入时不压缩任何的key-value对( Record )

public static class Writer implements java.io.Closeable {  
  
...  
   //初始化Writer  
   void init(Path name， Configuration conf， FSDataOutputStream out， Class keyClass， Class valClass， boolean compress， CompressionCodec codec， Metadata metadata) throws IOException {  
      this.conf = conf;  
      this.out = out;  
      this.keyClass = keyClass;  
      this.valClass = valClass;  
      this.compress = compress;  
      this.codec = codec;  
      this.metadata = metadata;  
        
      //创建非压缩的对象序列化器  
      SerializationFactory serializationFactory = new SerializationFactory(conf);  
      this.keySerializer = serializationFactory.getSerializer(keyClass);  
      this.keySerializer.open(buffer);  
      this.uncompressedValSerializer = serializationFactory.getSerializer(valClass);  
      this.uncompressedValSerializer.open(buffer);  
        
      //创建可压缩的对象序列化器  
      if (this.codec != null) {  
        ReflectionUtils.setConf(this.codec， this.conf);  
        this.compressor = CodecPool.getCompressor(this.codec);  
        this.deflateFilter = this.codec.createOutputStream(buffer， compressor);  
        this.deflateOut = new DataOutputStream(new BufferedOutputStream(deflateFilter));  
        this.compressedValSerializer = serializationFactory.getSerializer(valClass);  
        this.compressedValSerializer.open(deflateOut);  
      }  
    }  
      
  
  //添加一条记录(key-value，对象值需要序列化)  
  public synchronized void append(Object key， Object val) throws IOException {  
      if (key.getClass() != keyClass)  
        throw new IOException("wrong key class: "+key.getClass().getName() +" is not "+keyClass);  
        
      if (val.getClass() != valClass)  
        throw new IOException("wrong value class: "+val.getClass().getName() +" is not "+valClass);  
  
      buffer.reset();  
  
      //序列化key(将key转化为二进制数组)，并写入缓存buffer中  
      keySerializer.serialize(key);  
      int keyLength = buffer.getLength();  
      if (keyLength < 0)  
        throw new IOException("negative length keys not allowed: " + key);  
  
      //compress在初始化是被置为false   
      if (compress) {  
        deflateFilter.resetState();  
        compressedValSerializer.serialize(val);  
        deflateOut.flush();  
        deflateFilter.finish();  
      } else {  
        //序列化value值(不压缩)，并将其写入缓存buffer中  
        uncompressedValSerializer.serialize(val);  
      }  
  
      //将这条记录写入文件流  
      checkAndWriteSync();                                // sync  
      out.writeInt(buffer.getLength());                   // total record length  
      out.writeInt(keyLength);                            // key portion length  
      out.write(buffer.getData()， 0， buffer.getLength()); // data  
    }  
  
    //添加一条记录(key-value，二进制值)  
    public synchronized void appendRaw(byte[] keyData， int keyOffset， int keyLength， ValueBytes val) throws IOException {  
      if (keyLength < 0)  
        throw new IOException("negative length keys not allowed: " + keyLength);  
  
      int valLength = val.getSize();  
  
      checkAndWriteSync();  
        
      //直接将key-value写入文件流  
      out.writeInt(keyLength+valLength);          // total record length  
      out.writeInt(keyLength);                    // key portion length  
      out.write(keyData， keyOffset， keyLength);   // key  
      val.writeUncompressedBytes(out);            // value  
    }  
  
...  
  
} 

2.1.2.2 只压缩 Record 中的 value 值

SequenceFile.RecordCompressWrite写入时只压缩key-value对( Record ) 中的value

static class RecordCompressWriter extends Writer {  
...  
  
   public synchronized void append(Object key， Object val) throws IOException {  
      if (key.getClass() != keyClass)  
        throw new IOException("wrong key class: "+key.getClass().getName() +" is not "+keyClass);  
        
      if (val.getClass() != valClass)  
        throw new IOException("wrong value class: "+val.getClass().getName() +" is not "+valClass);  
  
      buffer.reset();  
  
      //序列化key(将key转化为二进制数组)，并写入缓存buffer中  
      keySerializer.serialize(key);  
      int keyLength = buffer.getLength();  
      if (keyLength < 0)  
        throw new IOException("negative length keys not allowed: " + key);  
  
      //序列化value值(不压缩)，并将其写入缓存buffer中  
      deflateFilter.resetState();  
      compressedValSerializer.serialize(val);  
      deflateOut.flush();  
      deflateFilter.finish();  
  
      //将这条记录写入文件流  
      checkAndWriteSync();                                // sync  
      out.writeInt(buffer.getLength());                   // total record length  
      out.writeInt(keyLength);                            // key portion length  
      out.write(buffer.getData()， 0， buffer.getLength()); // data  
    }  
  
    /** 添加一条记录(key-value，二进制值，value已压缩) */  
    public synchronized void appendRaw(byte[] keyData， int keyOffset，  
        int keyLength， ValueBytes val) throws IOException {  
  
      if (keyLength < 0)  
        throw new IOException("negative length keys not allowed: " + keyLength);  
  
      int valLength = val.getSize();  
        
      checkAndWriteSync();                        // sync  
      out.writeInt(keyLength+valLength);          // total record length  
      out.writeInt(keyLength);                    // key portion length  
      out.write(keyData， keyOffset， keyLength);   // 'key' data  
      val.writeCompressedBytes(out);              // 'value' data  
    }  
      
  } // RecordCompressionWriter  
  
  
...  
}

2.1.2.3 压缩 block 中的所有 records

SequenceFile.BlockComprocessWriter 写入时将一批key-value对( Record ) 压缩成一个Block

static class BlockCompressWriter extends Writer {  
...  
  
   void init(int compressionBlockSize) throws IOException {  
      this.compressionBlockSize = compressionBlockSize;  
      keySerializer.close();  
      keySerializer.open(keyBuffer);  
      uncompressedValSerializer.close();  
      uncompressedValSerializer.open(valBuffer);  
    }  
      
    /** Workhorse to check and write out compressed data/lengths */  
    private synchronized void writeBuffer(DataOutputBuffer uncompressedDataBuffer) throws IOException {  
      deflateFilter.resetState();  
      buffer.reset();  
      deflateOut.write(uncompressedDataBuffer.getData()， 0， uncompressedDataBuffer.getLength());  
      deflateOut.flush();  
      deflateFilter.finish();  
        
      WritableUtils.writeVInt(out， buffer.getLength());  
      out.write(buffer.getData()， 0， buffer.getLength());  
    }  
      
    /** Compress and flush contents to dfs */  
    public synchronized void sync() throws IOException {  
      if (noBufferedRecords > 0) {  
        super.sync();  
          
        // No. of records  
        WritableUtils.writeVInt(out， noBufferedRecords);  
          
        // Write 'keys' and lengths  
        writeBuffer(keyLenBuffer);  
        writeBuffer(keyBuffer);  
          
        // Write 'values' and lengths  
        writeBuffer(valLenBuffer);  
        writeBuffer(valBuffer);  
          
        // Flush the file-stream  
        out.flush();  
          
        // Reset internal states  
        keyLenBuffer.reset();  
        keyBuffer.reset();  
        valLenBuffer.reset();  
        valBuffer.reset();  
        noBufferedRecords = 0;  
      }  
        
    }  
  
  
   //添加一条记录(key-value，对象值需要序列化)  
   public synchronized void append(Object key， Object val) throws IOException {  
      if (key.getClass() != keyClass)  
        throw new IOException("wrong key class: "+key+" is not "+keyClass);  
        
      if (val.getClass() != valClass)  
        throw new IOException("wrong value class: "+val+" is not "+valClass);  
  
      //序列化key(将key转化为二进制数组)(未压缩)，并写入缓存keyBuffer中  
      int oldKeyLength = keyBuffer.getLength();  
      keySerializer.serialize(key);  
      int keyLength = keyBuffer.getLength() - oldKeyLength;  
      if (keyLength < 0)  
        throw new IOException("negative length keys not allowed: " + key);  
      WritableUtils.writeVInt(keyLenBuffer， keyLength);  
  
      //序列化value(将value转化为二进制数组)(未压缩)，并写入缓存valBuffer中  
      int oldValLength = valBuffer.getLength();  
      uncompressedValSerializer.serialize(val);  
      int valLength = valBuffer.getLength() - oldValLength;  
      WritableUtils.writeVInt(valLenBuffer， valLength);  
        
      // Added another key/value pair  
      ++noBufferedRecords;  
        
      // Compress and flush?  
      int currentBlockSize = keyBuffer.getLength() + valBuffer.getLength();  
      //block已满，可将整个block进行压缩并写入文件流  
      if (currentBlockSize >= compressionBlockSize) {  
        sync();  
      }  
    }  
      
    /**添加一条记录(key-value，二进制值，value已压缩). */  
    public synchronized void appendRaw(byte[] keyData， int keyOffset， int keyLength， ValueBytes val) throws IOException {  
        
      if (keyLength < 0)  
        throw new IOException("negative length keys not allowed");  
  
      int valLength = val.getSize();  
        
      // Save key/value data in relevant buffers  
      WritableUtils.writeVInt(keyLenBuffer， keyLength);  
      keyBuffer.write(keyData， keyOffset， keyLength);  
      WritableUtils.writeVInt(valLenBuffer， valLength);  
      val.writeUncompressedBytes(valBuffer);  
  
      // Added another key/value pair  
      ++noBufferedRecords;  
  
      // Compress and flush?  
      int currentBlockSize = keyBuffer.getLength() + valBuffer.getLength();   
      if (currentBlockSize >= compressionBlockSize) {  
        sync();  
      }  
    }  
      
  } // RecordCompressionWriter  
  
  
...  
}

源码中，block的大小compressionBlockSize默认值为1000000，也可通过配置参数io.seqfile.compress.blocksize来指定。

2.1.3 SequenceFile 文件格式图示

• 三种压缩算法共有三种SequenceFile文件格式

• 根据是否压缩， 以及采用记录压缩还是块压缩， 存储格式有所不同

2.1.3.1 不进行压缩

按照记录长度、key长度、value长度、key值、 value值一次存储， 长度指的是字节数， 采用指定的Serialization进行序列化

2.1.3.2 只压缩Record中的value值

只有value被压缩， 压缩的codec 保存在Header中

2.1.3.3 压缩block中的所有records

多条记录被压缩在一起， 可以利用记录之间的相似性， 更节省空间. Block前后都加入了同步标识. Block的最小值由 io.seqfile.compress.blocksize 属性设置

2.2 MapFile

MapFile是SequenceFile的变种，在sequenceFile中加入索引并排序后就是MapFile。 索引作为一个单独文件存储， 一般是每个128个记录存储一个索引。索引可以被载入内存， 用于快速查找。存放数据的文件根据Key定义的顺序排列

MapFile的记录必须按照顺序写入，否则抛出IOException。

MapFile的衍生类型:

• SetFile: 特殊的MapFile，用于存储一序列 Writable 类型的Key， Key按照顺序写入。
• ArrayFile: Key为整数， 代表在数组中的位置， value为Writable类型.
• BloomMapFile: 针对MapFile的get()方法， 使用动态Bloom过滤器进行优化， 过滤器保存在内存中， 之后在key值存在的时候， 才会调用常规的get()方法， 真正进行读取操作.

2.3 Avro DataFile

2.3.1 datafile组成图

datafile分为文件头是数据块，如果看图还是不明白，那么看这个应该会很清楚，datafile文件头的schema：

{
    "type": "record",
    "name": "org.apache.avro.file.Header",
    "fields": [
        {
            "name": "magic",
            "type": {
                "type": "fixed",
                "name": "Magic",
                "size": 4
            }
        },
        {
            "name": "meta",
            "type": {
                "type": "map",
                "values": "bytes"
            }
        },
        {
            "name": "sync",
            "type": {
                "type": "fixed",
                "name": "Sync",
                "size": 16
            }
        }
    ]
}

数据块相对容易理解，这里就不详述了。

要注意的是16字节的同步标记，这个标记意味着datafile支持随机读，并且可以做分割，也意味着可以作为mapreduce的输入

DataFileReader可以通过同步标记去随机读datafile文件

void seek(long position)

Move to a specific, known synchronization point, one returned from DataFileWriter.sync() while writing.

void sync(long position)

Move to the next synchronization point after a position.

2.3.2 datafile写操作

以代码注释的方式进行讲解：

//首先创建一个扩展名为avro的文件（扩展名随意，这里只是为了容易分辨） 

File file = new File("data.avro");

//这行和前篇文章的代码一致，创建一个Generic Record的datum写入类 

DatumWriter<GenericRecord> writer = new GenericDatumWriter<GenericRecord>(schema);

//和Encoder不同，DataFileWriter可以将avro数据写入到文件中 

DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<GenericRecord>(writer);

//创建文件，并且写入头信息 

dataFileWriter.create(schema,file);

//写datum数据 

dataFileWriter.append(datum);

dataFileWriter.append(datum);

dataFileWriter.close(); 

2.3.3 datafile读操作

// Generic Record的datum读取类，有点不一样的就是这里不需要再传入schema，因为schema已经包含在datafile的头信息里 

DatumReader<GenericRecord> reader=new GenericDatumReader<GenericRecord>();

//datafile文件的读取类，指定文件和datumreader 

DataFileReader<GenericRecord> dataFileReader=new DataFileReader<GenericRecord>(file,reader);

//测试下读写的schema是否一致 

Assert.assertEquals(schema,dataFileReader.getSchema());

//遍历GenericRecord 

for (GenericRecord record : dataFileReader){

 System.out.println("left="+record.get("left")+",right="+record.get("right"));

}