码迷，mamicode.com
首页 > 其他好文 > 详细

SequenceFile文件

时间：2017-06-24 09:56:53      阅读：198      评论：0      收藏：0      [点我收藏+]

标签：类型   tao   records   一个   解决方案   lte   null   sdn   void   

    SequenceFile文件是Hadoop用来存储二进制形式的key-value对而设计的一种平面文件(Flat File)。目前，也有不少人在该文件的基础之上提出了一些HDFS中小文件存储的解决方案，他们的基本思路就是将小文件进行合并成一个大文件，同时对这些小文件的位置信息构建索引。不过，这类解决方案还涉及到Hadoop的另一种文件格式——MapFile文件。SequenceFile文件并不保证其存储的key-value数据是按照key的某个顺序存储的，同时不支持append操作。

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

NONE: 对records不进行压缩;

RECORD: 仅压缩每一个record中的value值;

BLOCK: 将一个block中的所有records压缩在一起;

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

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

 

[java] view plain copy
 
  1. public static class Writer implements java.io.Closeable {  
  2.   
  3. ...  
  4.    //初始化Writer  
  5.    void init(Path name, Configuration conf, FSDataOutputStream out, Class keyClass, Class valClass, boolean compress, CompressionCodec codec, Metadata metadata) throws IOException {  
  6.       this.conf = conf;  
  7.       this.out = out;  
  8.       this.keyClass = keyClass;  
  9.       this.valClass = valClass;  
  10.       this.compress = compress;  
  11.       this.codec = codec;  
  12.       this.metadata = metadata;  
  13.         
  14.       //创建非压缩的对象序列化器  
  15.       SerializationFactory serializationFactory = new SerializationFactory(conf);  
  16.       this.keySerializer = serializationFactory.getSerializer(keyClass);  
  17.       this.keySerializer.open(buffer);  
  18.       this.uncompressedValSerializer = serializationFactory.getSerializer(valClass);  
  19.       this.uncompressedValSerializer.open(buffer);  
  20.         
  21.       //创建可压缩的对象序列化器  
  22.       if (this.codec != null) {  
  23.         ReflectionUtils.setConf(this.codec, this.conf);  
  24.         this.compressor = CodecPool.getCompressor(this.codec);  
  25.         this.deflateFilter = this.codec.createOutputStream(buffer, compressor);  
  26.         this.deflateOut = new DataOutputStream(new BufferedOutputStream(deflateFilter));  
  27.         this.compressedValSerializer = serializationFactory.getSerializer(valClass);  
  28.         this.compressedValSerializer.open(deflateOut);  
  29.       }  
  30.     }  
  31.       
  32.   
  33.   //添加一条记录(key-value,对象值需要序列化)  
  34.   public synchronized void append(Object key, Object val) throws IOException {  
  35.       if (key.getClass() != keyClass)  
  36.         throw new IOException("wrong key class: "+key.getClass().getName() +" is not "+keyClass);  
  37.         
  38.       if (val.getClass() != valClass)  
  39.         throw new IOException("wrong value class: "+val.getClass().getName() +" is not "+valClass);  
  40.   
  41.       buffer.reset();  
  42.   
  43.       //序列化key(将key转化为二进制数组)，并写入缓存buffer中  
  44.       keySerializer.serialize(key);  
  45.       int keyLength = buffer.getLength();  
  46.       if (keyLength < 0)  
  47.         throw new IOException("negative length keys not allowed: " + key);  
  48.   
  49.       //compress在初始化是被置为false   
  50.       if (compress) {  
  51.         deflateFilter.resetState();  
  52.         compressedValSerializer.serialize(val);  
  53.         deflateOut.flush();  
  54.         deflateFilter.finish();  
  55.       } else {  
  56.         //序列化value值(不压缩)，并将其写入缓存buffer中  
  57.         uncompressedValSerializer.serialize(val);  
  58.       }  
  59.   
  60.       //将这条记录写入文件流  
  61.       checkAndWriteSync();                                // sync  
  62.       out.writeInt(buffer.getLength());                   // total record length  
  63.       out.writeInt(keyLength);                            // key portion length  
  64.       out.write(buffer.getData(), 0, buffer.getLength()); // data  
  65.     }  
  66.   
  67.     //添加一条记录(key-value,二进制值)  
  68.     public synchronized void appendRaw(byte[] keyData, int keyOffset, int keyLength, ValueBytes val) throws IOException {  
  69.       if (keyLength < 0)  
  70.         throw new IOException("negative length keys not allowed: " + keyLength);  
  71.   
  72.       int valLength = val.getSize();  
  73.   
  74.       checkAndWriteSync();  
  75.         
  76.       //直接将key-value写入文件流  
  77.       out.writeInt(keyLength+valLength);          // total record length  
  78.       out.writeInt(keyLength);                    // key portion length  
  79.       out.write(keyData, keyOffset, keyLength);   // key  
  80.       val.writeUncompressedBytes(out);            // value  
  81.     }  
  82.   
  83. ...  
  84.   
  85. }  

 

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

 

[java] view plain copy
 
  1. static class RecordCompressWriter extends Writer {  
  2. ...  
  3.   
  4.    public synchronized void append(Object key, Object val) throws IOException {  
  5.       if (key.getClass() != keyClass)  
  6.         throw new IOException("wrong key class: "+key.getClass().getName() +" is not "+keyClass);  
  7.         
  8.       if (val.getClass() != valClass)  
  9.         throw new IOException("wrong value class: "+val.getClass().getName() +" is not "+valClass);  
  10.   
  11.       buffer.reset();  
  12.   
  13.       //序列化key(将key转化为二进制数组)，并写入缓存buffer中  
  14.       keySerializer.serialize(key);  
  15.       int keyLength = buffer.getLength();  
  16.       if (keyLength < 0)  
  17.         throw new IOException("negative length keys not allowed: " + key);  
  18.   
  19.       //序列化value值(不压缩)，并将其写入缓存buffer中  
  20.       deflateFilter.resetState();  
  21.       compressedValSerializer.serialize(val);  
  22.       deflateOut.flush();  
  23.       deflateFilter.finish();  
  24.   
  25.       //将这条记录写入文件流  
  26.       checkAndWriteSync();                                // sync  
  27.       out.writeInt(buffer.getLength());                   // total record length  
  28.       out.writeInt(keyLength);                            // key portion length  
  29.       out.write(buffer.getData(), 0, buffer.getLength()); // data  
  30.     }  
  31.   
  32.     /** 添加一条记录(key-value,二进制值,value已压缩) */  
  33.     public synchronized void appendRaw(byte[] keyData, int keyOffset,  
  34.         int keyLength, ValueBytes val) throws IOException {  
  35.   
  36.       if (keyLength < 0)  
  37.         throw new IOException("negative length keys not allowed: " + keyLength);  
  38.   
  39.       int valLength = val.getSize();  
  40.         
  41.       checkAndWriteSync();                        // sync  
  42.       out.writeInt(keyLength+valLength);          // total record length  
  43.       out.writeInt(keyLength);                    // key portion length  
  44.       out.write(keyData, keyOffset, keyLength);   // ‘key‘ data  
  45.       val.writeCompressedBytes(out);              // ‘value‘ data  
  46.     }  
  47.       
  48.   } // RecordCompressionWriter  
  49.   
  50.   
  51. ...  
  52. }  

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

 

[java] view plain copy
 
  1. static class BlockCompressWriter extends Writer {  
  2. ...  
  3.   
  4.    void init(int compressionBlockSize) throws IOException {  
  5.       this.compressionBlockSize = compressionBlockSize;  
  6.       keySerializer.close();  
  7.       keySerializer.open(keyBuffer);  
  8.       uncompressedValSerializer.close();  
  9.       uncompressedValSerializer.open(valBuffer);  
  10.     }  
  11.       
  12.     /** Workhorse to check and write out compressed data/lengths */  
  13.     private synchronized void writeBuffer(DataOutputBuffer uncompressedDataBuffer) throws IOException {  
  14.       deflateFilter.resetState();  
  15.       buffer.reset();  
  16.       deflateOut.write(uncompressedDataBuffer.getData(), 0, uncompressedDataBuffer.getLength());  
  17.       deflateOut.flush();  
  18.       deflateFilter.finish();  
  19.         
  20.       WritableUtils.writeVInt(out, buffer.getLength());  
  21.       out.write(buffer.getData(), 0, buffer.getLength());  
  22.     }  
  23.       
  24.     /** Compress and flush contents to dfs */  
  25.     public synchronized void sync() throws IOException {  
  26.       if (noBufferedRecords > 0) {  
  27.         super.sync();  
  28.           
  29.         // No. of records  
  30.         WritableUtils.writeVInt(out, noBufferedRecords);  
  31.           
  32.         // Write ‘keys‘ and lengths  
  33.         writeBuffer(keyLenBuffer);  
  34.         writeBuffer(keyBuffer);  
  35.           
  36.         // Write ‘values‘ and lengths  
  37.         writeBuffer(valLenBuffer);  
  38.         writeBuffer(valBuffer);  
  39.           
  40.         // Flush the file-stream  
  41.         out.flush();  
  42.           
  43.         // Reset internal states  
  44.         keyLenBuffer.reset();  
  45.         keyBuffer.reset();  
  46.         valLenBuffer.reset();  
  47.         valBuffer.reset();  
  48.         noBufferedRecords = 0;  
  49.       }  
  50.         
  51.     }  
  52.   
  53.   
  54.    //添加一条记录(key-value,对象值需要序列化)  
  55.    public synchronized void append(Object key, Object val) throws IOException {  
  56.       if (key.getClass() != keyClass)  
  57.         throw new IOException("wrong key class: "+key+" is not "+keyClass);  
  58.         
  59.       if (val.getClass() != valClass)  
  60.         throw new IOException("wrong value class: "+val+" is not "+valClass);  
  61.   
  62.       //序列化key(将key转化为二进制数组)(未压缩)，并写入缓存keyBuffer中  
  63.       int oldKeyLength = keyBuffer.getLength();  
  64.       keySerializer.serialize(key);  
  65.       int keyLength = keyBuffer.getLength() - oldKeyLength;  
  66.       if (keyLength < 0)  
  67.         throw new IOException("negative length keys not allowed: " + key);  
  68.       WritableUtils.writeVInt(keyLenBuffer, keyLength);  
  69.   
  70.       //序列化value(将value转化为二进制数组)(未压缩)，并写入缓存valBuffer中  
  71.       int oldValLength = valBuffer.getLength();  
  72.       uncompressedValSerializer.serialize(val);  
  73.       int valLength = valBuffer.getLength() - oldValLength;  
  74.       WritableUtils.writeVInt(valLenBuffer, valLength);  
  75.         
  76.       // Added another key/value pair  
  77.       ++noBufferedRecords;  
  78.         
  79.       // Compress and flush?  
  80.       int currentBlockSize = keyBuffer.getLength() + valBuffer.getLength();  
  81.       //block已满，可将整个block进行压缩并写入文件流  
  82.       if (currentBlockSize >= compressionBlockSize) {  
  83.         sync();  
  84.       }  
  85.     }  
  86.       
  87.     /**添加一条记录(key-value,二进制值,value已压缩). */  
  88.     public synchronized void appendRaw(byte[] keyData, int keyOffset, int keyLength, ValueBytes val) throws IOException {  
  89.         
  90.       if (keyLength < 0)  
  91.         throw new IOException("negative length keys not allowed");  
  92.   
  93.       int valLength = val.getSize();  
  94.         
  95.       // Save key/value data in relevant buffers  
  96.       WritableUtils.writeVInt(keyLenBuffer, keyLength);  
  97.       keyBuffer.write(keyData, keyOffset, keyLength);  
  98.       WritableUtils.writeVInt(valLenBuffer, valLength);  
  99.       val.writeUncompressedBytes(valBuffer);  
  100.   
  101.       // Added another key/value pair  
  102.       ++noBufferedRecords;  
  103.   
  104.       // Compress and flush?  
  105.       int currentBlockSize = keyBuffer.getLength() + valBuffer.getLength();   
  106.       if (currentBlockSize >= compressionBlockSize) {  
  107.         sync();  
  108.       }  
  109.     }  
  110.       
  111.   } // RecordCompressionWriter  
  112.   
  113.   
  114. ...  
  115. }  

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

 

   根据三种压缩算法，共有三种类型的SequenceFile文件格式：

1). Uncompressed SequenceFile

    技术分享

 

 

2). Record-Compressed SequenceFile

技术分享

3). Block-Compressed SequenceFile

技术分享

 

SequenceFile文件

标签：类型   tao   records   一个   解决方案   lte   null   sdn   void   

原文地址：http://www.cnblogs.com/mfryf/p/7072446.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！