2. 程式人生 > >hadoop SequenceFile——大資料 儲存

hadoop SequenceFile——大資料 儲存

阿新 • • 發佈：2018-12-10

SequenceFile是一個由二進位制序列化過的key/value的位元組流組成的文字儲存檔案。

基於壓縮型別CompressType，共有三種SequenceFile Writer：

public static enum CompressionType {
    /** 不壓縮 */
    NONE, 
    /** 只壓縮value */
    RECORD,
    /** 壓縮很多記錄的key/value成一塊 */
    BLOCK
  }

There are three SequenceFile Writers based on the CompressType used to compress key/value pairs:

1、Writer : Uncompressed records.

在SequenceFile裡有Writer的實現，原始碼如下：

 public static class Writer implements java.io.Closeable, Syncable

  大資料學習去群119599574

 /** Write and flush the file header. */
    private void writeFileHeader() 
      throws IOException {
      out.write(VERSION);
      Text.writeString(out, keyClass.getName());
      Text.writeString(out, valClass.getName());
      
      out.writeBoolean(this.isCompressed());
      out.writeBoolean(this.isBlockCompressed());
      
      if (this.isCompressed()) {
        Text.writeString(out, (codec.getClass()).getName());
      }
      this.metadata.write(out);
      out.write(sync);                       // write the sync bytes
      out.flush();                           // flush header
    }

首先SequenceFile檔案有個表頭，以上是寫表頭的程式碼，我們以一個例項結合程式碼來看一下表頭的組成。

package demo;

import java.io.IOException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.Text;

public class SequenceFileWriteDemo {
	private static final String[] DATA = { "One, two, buckle my shoe",
			"Three, four, shut the door", "Five, six, pick up sticks",
			"Seven, eight, lay them straight", "Nine, ten, a big fat hen" };

	public static void main(String[] args) throws IOException {
		String uri = args[0];
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", "hdfs://xxx.xxx.xxx.xx:9000");
		FileSystem fs = FileSystem.get(URI.create(uri), conf);
		Path path = new Path(uri);
		IntWritable key = new IntWritable();
		Text value = new Text();
		SequenceFile.Writer writer = null;
		try {
			String compressType = args[1];
			System.out.println("compressType "+compressType);
			
				//	Writer : Uncompressed records. 
			if(compressType.equals("1") ){
				System.out.println("compress none");
				writer = SequenceFile.createWriter(fs, conf, path, key.getClass(),value.getClass(),CompressionType.NONE);
			}else if(compressType .equals("2") ){
				System.out.println("compress record");
				//RecordCompressWriter : Record-compressed files, only compress values. 
				writer = SequenceFile.createWriter(fs, conf, path, key.getClass(),value.getClass(),CompressionType.RECORD);	
			}else if(compressType.equals("3") ){
				System.out.println("compress block");
				//	BlockCompressWriter : Block-compressed files, both keys & values are collected in 'blocks' separately and compressed. The size of the 'block' is configurable. 
				writer = SequenceFile.createWriter(fs, conf, path, key.getClass(),value.getClass(),CompressionType.BLOCK);	
			}
			
			for (int i = 0; i < 100; i++) {
				key.set(100 - i);
				value.set(DATA[i % DATA.length]);
				System.out.printf("[%s]\t%s\t%s\n", writer.getLength(), key,value);
				writer.append(key, value);
				
			}
		} finally {
			IOUtils.closeStream(writer);
		}
	}
}

存入的文字檔案內容如下：

key     value

100     One, two, buckle my shoe

99      Three, four, shut the door

98      Five, six, pick up sticks

97      Seven, eight, lay them straight

96      Nine, ten, a big fat hen

95      One, two, buckle my shoe

94      Three, four, shut the door

93      Five, six, pick up sticks

92      Seven, eight, lay them straight

91      Nine, ten, a big fat hen

90      One, two, buckle my shoe

89      Three, four, shut the door

88      Five, six, pick up sticks

87      Seven, eight, lay them straight

86      Nine, ten, a big fat hen

85      One, two, buckle my shoe

84      Three, four, shut the door

83      Five, six, pick up sticks

82      Seven, eight, lay them straight

81      Nine, ten, a big fat hen

80      One, two, buckle my shoe

79      Three, four, shut the door

78      Five, six, pick up sticks

77      Seven, eight, lay them straight

76      Nine, ten, a big fat hen

75      One, two, buckle my shoe

74      Three, four, shut the door

73      Five, six, pick up sticks

72      Seven, eight, lay them straight

71      Nine, ten, a big fat hen

70      One, two, buckle my shoe

69      Three, four, shut the door

68      Five, six, pick up sticks

67      Seven, eight, lay them straight

66      Nine, ten, a big fat hen

65      One, two, buckle my shoe

64      Three, four, shut the door

63      Five, six, pick up sticks

62      Seven, eight, lay them straight

61      Nine, ten, a big fat hen

60      One, two, buckle my shoe

59      Three, four, shut the door

58      Five, six, pick up sticks

57      Seven, eight, lay them straight

56      Nine, ten, a big fat hen

55      One, two, buckle my shoe

54      Three, four, shut the door

53      Five, six, pick up sticks

52      Seven, eight, lay them straight

51      Nine, ten, a big fat hen

50      One, two, buckle my shoe

49      Three, four, shut the door

48      Five, six, pick up sticks

47      Seven, eight, lay them straight

46      Nine, ten, a big fat hen

45      One, two, buckle my shoe

44      Three, four, shut the door

43      Five, six, pick up sticks

42      Seven, eight, lay them straight

41      Nine, ten, a big fat hen

40      One, two, buckle my shoe

39      Three, four, shut the door

38      Five, six, pick up sticks

37      Seven, eight, lay them straight

36      Nine, ten, a big fat hen

35      One, two, buckle my shoe

34      Three, four, shut the door

33      Five, six, pick up sticks

32      Seven, eight, lay them straight

31      Nine, ten, a big fat hen

30      One, two, buckle my shoe

29      Three, four, shut the door

28      Five, six, pick up sticks

27      Seven, eight, lay them straight

26      Nine, ten, a big fat hen

25      One, two, buckle my shoe

24      Three, four, shut the door

23      Five, six, pick up sticks

22      Seven, eight, lay them straight

21      Nine, ten, a big fat hen

20      One, two, buckle my shoe

19      Three, four, shut the door

18      Five, six, pick up sticks

17      Seven, eight, lay them straight

16      Nine, ten, a big fat hen

15      One, two, buckle my shoe

14      Three, four, shut the door

13      Five, six, pick up sticks

12      Seven, eight, lay them straight

11      Nine, ten, a big fat hen

10      One, two, buckle my shoe

9       Three, four, shut the door

8       Five, six, pick up sticks

7       Seven, eight, lay them straight

6       Nine, ten, a big fat hen

5       One, two, buckle my shoe

4       Three, four, shut the door

3       Five, six, pick up sticks

2       Seven, eight, lay them straight

1       Nine, ten, a big fat hen

把以上java 打成jar,執行hadoop jar sfile.jar  /test/numbers.seq  1  1代表不壓縮

hadoop fs -get hdfs:///test/numbers1.seq   /usr/test/  取出剛剛生成的檔案，我們用UE開啟看一下

0x53 0x45 0x51

這是SequenceFile Format的magic header「SEQ」，用來區別文字是否是「SequenceFile Format」。

0x06 版本編號，目前最新版為「SEQ6」

以上程式碼原始碼如下：

 private static byte[] VERSION = new byte[] {
    (byte)'S', (byte)'E', (byte)'Q', VERSION_WITH_METADATA
  };
out.write(VERSION);

0x20 0x65........0x65

這部份是keyClassName(Key的類名)，而第1個Byte(0x20)用來表示此字串的長度，此範例為org.apache.hadoop.io.IntWritable

以上程式碼原始碼如下：

 Text.writeString(out, keyClass.getName());

0x19 0x6F  ..... 0x74 這部份是valueClassName(Value的類名)，第1個Byte(0x19)也是用來表示此字串的長度，此範例為org.apache.hadoop.io.BytesWritable

 Text.writeString(out, valClass.getName());

大資料學習去群119599574

0x00 是否compression？「0x00」=否 (此為Boolean所以佔1個Byte)0x00 是否blockCompression？「0x00」=否(此為Boolean所以佔1個Byte)

   out.writeBoolean(this.isCompressed());
    out.writeBoolean(this.isBlockCompressed());

如果是壓縮的話接下去還會寫壓縮類，此範例沒有壓縮所以沒有此類名的寫入，原始碼如下：

 if (this.isCompressed()) {
        Text.writeString(out, (codec.getClass()).getName());
      }

0x00 0x00 0x00 0x00 metadata，此範例沒有包含任何metadata， 所以輸出「0x00 0x00 0x00 0x00」

  this.metadata.write(out);

0x76 0x61  ..... 0xAF sync標記，用來表示一個「Header」的結束。

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);
      }
    }
    
out.write(sync);

至此頭部檔案寫入完畢，可見頭部的格式可以歸納如下：

SequenceFile Header

  • version - 3 bytes of magic header SEQ, followed by 1 byte of actual version number (e.g. SEQ4 or SEQ6)

  • keyClassName -key class

  • valueClassName - value class

  • compression - A boolean which specifies if compression is turned on for keys/values in this file.

  • blockCompression - A boolean which specifies if block-compression is turned on for keys/values in this file.

  • compression codec - CompressionCodec class which is used for compression of keys and/or values (if compression is enabled).

  • metadata -  for this file.

  • sync - A sync marker to denote end of the header. 

接下去我們看一下資料儲存格式：

0x00 0x00 0x00 0x1D

整個Record的size ，一個Record包含「Key、Value」。此處為29個位元組，因為 key=100 佔4個位元組，value=One, two, buckle my shoe 佔24位元組  還有一個位元組存了value的長度 所以 29位元組總共。

0x00 0x00 0x00 0x04

Key內容的size~ (此為Int所以佔4個位元組)

0x00 0x00 0x00 0x64

Key的內容，此處為100，那十六進位制就是64

大資料學習去群119599574

0x18

value內容的size，此處是One, two, buckle my shoe，長度24 所以十六進位制就是18

0x4F 0X6E....0x65

value的內容One, two, buckle my shoe

以上程式碼原始碼如下：

 public synchronized void append(Object key, Object val){
 .......
 // Write the record out
      checkAndWriteSync();                                // sync
      out.writeInt(buffer.getLength());                   // total record length
      out.writeInt(keyLength);                            // key portion length
      out.write(buffer.getData(), 0, buffer.getLength()); // data
 }

 當資料達到一個閥值，會寫sync，寫sync就是呼叫checkAndWriteSync();   原始碼如下：

 synchronized void checkAndWriteSync() throws IOException {
      if (sync != null &&
          out.getPos() >= lastSyncPos+SYNC_INTERVAL) { // time to emit sync
        sync();
      }
    }

SYNC_INTERVAL定義如下：

  private static final int SYNC_HASH_SIZE = 16;   // number of bytes in hash 
  private static final int SYNC_SIZE = 4+SYNC_HASH_SIZE; // escape + hash

  /** The number of bytes between sync points.*/
  public static final int SYNC_INTERVAL = 100*SYNC_SIZE;

可見每2000byte會寫一個sync

SequeceFile 無壓縮圖示如下：

可見格式如下：

Uncompressed SequenceFile Format

  • Record

    • Record length

    • Key length

    • Key

    • Value 

接下去我們看一下RecordCompressWriter

2、RecordCompressWriter : Record-compressed files, only compress values.

執行hadoop jar sfile.jar  /test/numbers2.seq  2  2代表使用RecordCompressWriter壓縮

同樣我們用UE看一下生成的檔案：

可見表頭和不壓縮的基本一致，有些小區別如下：

0x01

代表使用了壓縮

0x2A 0x6F ....0x63

使用的壓縮類org.apache.hadoop.io.compress.DefaultCodec

0x00 ...0x25

整個Record的size ,為37，為啥比不壓縮佔用的位元組更多？

0x00 0x00 0x00 0x04

Key內容的size~ (此為Int所以佔4個位元組)

0x00 0x00 0x00 0x64

Key的內容，此處為100，那十六進位制就是64 可見key沒有壓縮

以上原始碼如下：

 @Override
    @SuppressWarnings("unchecked")
    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();

      // Append the 'key'
      keySerializer.serialize(key);
      int keyLength = buffer.getLength();
      if (keyLength < 0)
        throw new IOException("negative length keys not allowed: " + key);

      // Compress 'value' and append it
      deflateFilter.resetState();
      compressedValSerializer.serialize(val);
      deflateOut.flush();
      deflateFilter.finish();

      // Write the record out
      checkAndWriteSync();                                // sync
      out.writeInt(buffer.getLength());                   // total record length
      out.writeInt(keyLength);                            // key portion length
      out.write(buffer.getData(), 0, buffer.getLength()); // data
    }

SequeceFile 壓縮value圖示如下：

可見格式如下：

Record-Compressed SequenceFile Format

  • Record

    • Record length

    • Key length

    • Key

    • Compressed Value 

接下去我們看一下BlockCompressWriter

3、BlockCompressWriter : Block-compressed files, both keys & values are collected in 'blocks' separately and compressed. The size of the 'block' is configurable. 

執行hadoop jar sfile.jar  /test/numbers3.seq  3  3代表使用BlockCompressWriter壓縮

同樣我們用UE看一下生成的檔案：

可見表頭和不壓縮的基本一致，有些小區別如下：

0x01

代表使用了壓縮

0x01

代表使用了block壓縮

對於block壓縮的原始碼如下：

 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);

      // Save key/value into respective buffers 
      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);

      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();
      if (currentBlockSize >= compressionBlockSize) {
        sync();
      }
    }

 其中，可見沒呼叫一次append，就會相應累加keyLenBuffer和valLenBuffer的長度

 WritableUtils.writeVInt(keyLenBuffer, keyLength);

 WritableUtils.writeVInt(valLenBuffer, valLength);

 int currentBlockSize = keyBuffer.getLength() + valBuffer.getLength();

      if (currentBlockSize >= compressionBlockSize) {

        sync();

      }

compressionBlockSize =  conf.getInt("io.seqfile.compress.blocksize", 1000000);

當超過compressionBlockSize是會呼叫sync，我們來看看sync的原始碼，如下：

  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;
      }

Sequenc File ，塊壓縮圖示如下：

可見格式如下：

Block-Compressed SequenceFile Format

  • Record Block

    • Uncompressed number of records in the block

    • Compressed key-lengths block-size

    • Compressed key-lengths block

    • Compressed keys block-size

    • Compressed keys block

    • Compressed value-lengths block-size

    • Compressed value-lengths block

    • Compressed values block-size

    • Compressed values block

  • A sync-marker every block. 

相關推薦

hadoop SequenceFile——大資料 儲存

SequenceFile是一個由二進位制序列化過的key/value的位元組流組成的文字儲存檔案。 基於壓縮型別CompressType，共有三種SequenceFile Writer： public static enum CompressionType { /

hadoop SequenceFile介紹 大資料 儲存

SequenceFile是一個由二進位制序列化過的key/value的位元組流組成的文字儲存檔案。 基於壓縮型別CompressType，共有三種SequenceFileWriter： 1 2 3 4 5 6 7 8 public static enum Co

大資料之hadoop對比spark------資料儲存

1.Hadoop的資料都是來自於哪裡： 答案：磁碟。 2.map與reduce可以同時執行嗎？ 答案：不能，由什麼決定的，shuffle過程決定的。 3.spark為什麼比hadoop要快，sprak儘量的避免從磁碟中進行讀取，以及配置資訊和計算資料，對比這些特性，極

【Hadoop & Hbase】大資料儲存系統程式設計_讀出寫入

0x01 編譯環境 jdk 1.7 hbase 0.98 hadoop 2.6 Ubuntu Linux 14.04.2 0x02 目標 其中我的任務編號為5，即Sort-based Distinct Sort-based Distinct

雲時代的大資料儲存-雲HBase

為什麼 縱觀資料庫發展的幾十年，從網狀資料庫、層次資料庫到RDBMS資料庫，在最近幾年的NewSQL的興起，加上開源的運動，再加上雲的特性，可以說是日新月異。在20世紀80年代後，大部分的業務確定使用RDBMS資料為儲存基礎。新世紀開始，隨著網際網路的發展，資料量的增大，慢慢RDBMS資料庫撐不住，就出

0462-Hadoop和大資料要完蛋了嗎？

Fayson的github： https://github.com/fayson/cdhproject 推薦關注微信公眾號：“Hadoop實操”，ID：gh_c4c535955d0f，或者掃描文末二維碼。 作者：Alex Robbio，Belatrix

玩轉大資料！Hadoop在大資料的地位如此之高？

馬雲曾說過：“未來最大的資源就是資料，不參與大資料十年後一定會後悔！” 馬雲 Hadoop 最早誕生於 2006 年，並在 2008 年成為 Apache 頂級專案，起初，Hadoop在中國基本沒有人注意到它，直到2011年開始，中國進入大資料風起雲湧的時代，因

如何使用HBase？大資料儲存的兩個實戰場景

現如今各種資料儲存方案層出不窮，本文僅僅是結合兩個實戰場景就基於HBase的大資料儲存做了簡單的分析，並對HBase的原理做了簡單的闡述。如何使用好HBase，甚至於如何選擇一個最優的資料儲存方案，還需要我們根據場景需要具體分析和設計 HBase是一個高可靠性、高效能、面向

大資料儲存、計算、應用、視覺化，資料的基本概述都在這裡了

未來的時代，一定是資料的時代，在未來，一切被記錄，一切被分析，資料將以資產的方式存在，相關知識如下： 一、先說各種資料儲存 資料是個很泛的概念，但是我們腦海裡第一反應的就是關係型資料庫和EXCEL這種二維表是資料。 而現在資料各種各樣特色，有文件、有圖片、有流式

大資料儲存---HBase介紹（上）

本次主要介紹三部分： HBase簡介 HBase整體架構 HBase安裝和啟動 Hbase基本操作 HBase簡介 hbase是bigtable的開源java版本，是建立在hdfs之上。 提供高可靠性、高效能、列儲存、可伸縮、實時讀寫nosql的資料庫系統

大資料儲存---HBase常用介紹（中）

我們這裡主要介紹HBase的API 基礎API 封裝工具類 基礎API 建立表 新增資料 查詢資料的三種方式 掃描查詢 get方式執行查詢 過濾查詢 PS:刪除表請通過shell命令進入客戶端刪除。 package com.hbase; imp

Hadoop和大資料最炫目的60款頂級開源工具

說到處理大資料的工具，普通的開源解決方案（尤其是Apache Hadoop）堪稱中流砥柱。弗雷斯特調研公司的分析師Mike Gualtieri最近預測，在接下來幾年，“100%的大公司”會採用Hadoop。Market Research的一份報告預測，到2011年，Hadoop市場會以58%的年複合

GIS大資料儲存預研

1. 背景 在實際專案執行中，時常會出現希望搜尋周邊所有資料的需求。但是以常規的儲存方案，每種資源均為一個圖層或一個表，比如人員軌跡表、車輛軌跡表、各類空間圖層表等。在進行全文空間收索時，基於傳統空間關係庫或後臺圖層服務的遍歷查詢則過於耗時。這裡，我們研究基於ElasticSearch來進行所有資料的整合，

大資料儲存的進化史 --從 RAID 到 Hdfs

　我們都知道現在大資料儲存用的基本都是 Hdfs ，但在 Hadoop 誕生之前，我們都是如何儲存大量資料的呢?這次我們不聊技術架構什麼的，而是從技術演化的角度來看看 Hadoop Hdfs。 　　我們先來思考兩個問題。 　　在 Hdfs 出現以前，計算機是通過什麼手段來儲存“大資料”

超越Hadoop的大資料分析之圖形處理尺寸

另一個來自Google的重要工具，看起來超越了Hadoop MR——Pregel框架實現了圖形計算（Malewicez et al.2010）。在Pregel中的計算是由一系列迭代組成的，被稱為supersteps。圖上的每個頂點都與一個使用者定義的計算函式相關聯；Pregel確保每個sup

超越Hadoop的大資料分析之致謝

首先，我要衷心感謝Vineet Tyagi、AVP和Impetus的創新實驗室主管。Vineet對我幫助很大，並促使我寫這本書。在6、7月份，每個工作日他給我3個小時的時間寫作本書，這是幫助我完成本書的關鍵。任何學術活動都要專門花費很多時間——這時候就得加倍努力，因為我必須在工作之餘寫作。V

超越Hadoop的大資料分析之第一章介紹：為什麼超越Hadoop Map-Reduce

本文翻譯自《BIG DATA ANALYTICS BEYOND HADOOP》譯者：吳京潤 譯者注：本文是本書第一章的開頭，第一章其它部分由其他人翻譯。 你可能是一個視訊服務提供商，而你想基於網路環境動態的選擇合適的內容分發網路來優化終端使用者的體驗。或者你是一個政府監管機構，需要為網際網路頁

超越Hadoop的大資料分析之前言

我試圖給人們學習大資料留下的一點深刻印象：儘管Apache Hadoop很有用，而且是一項非常成功的技術，但是這一觀點的前提已經有些過時了。考慮一下這樣一條時間線：由谷歌實現的MapReduce投入使用的時間可追溯到2002年，發表於2004年。Yahoo!於2006年發起Hadoop專案。

大資料儲存基石HDFS

這篇文章主要介紹HDFS的概述、讀寫流程，常用的shell操作以及一些HDFS 2.X的新特性 HDFS(Hadoop distributed file system)，通過目錄樹來定位檔案，檔案實際以塊分散式存在各個節點 優點：通過副本容錯，在廉價機上儲存海量資料。 缺點：不能高效儲存小檔

大資料儲存之分散式檔案系統（一）

1.Google檔案系統（GFS） 使用一堆廉價的商用計算機支撐大規模資料處理。 GFSClient： 應用程式的訪問介面 Master（主控伺服器）：管理節點，在邏輯上只有一個（還有一臺“影子伺服器“，在主控伺服器失效時提供元資料，但並不是完整的熱備伺服器），儲