parquet 簡介

 

原文

 

Parquet 列式存儲格式

面向分析型業務的列式存儲格式

由 Twitter 和 Cloudera 合作開發，2015 年 5 月從 Apache 的孵化器里畢業成為 Apache 頂級項目

 

 

列式存儲

列式存儲和行式存儲相比有哪些優勢呢？

1. 可以跳過不符合條件的數據，只讀取需要的數據，降低 IO 數據量。
2. 壓縮編碼可以降低磁盤存儲空間。由於同一列的數據類型是一樣的，可以使用更高效的壓縮編碼（例如 Run Length Encoding 和 Delta Encoding）進一步節約存儲空間。
3. 只讀取需要的列，支持向量運算，能夠獲取更好的掃描性能。

 當時 Twitter 的日增數據量達到壓縮之后的 100TB+，存儲在 HDFS 上，工程師會使用多種計算框架（例如 MapReduce, Hive, Pig 等）對這些數據做分析和挖掘；

日志結構是復雜的嵌套數據類型，例如一個典型的日志的 schema 有 87 列，嵌套了 7 層。所以需要設計一種列式存儲格式，既能支持關系型數據（簡單數據類型），又能支持復雜的嵌套類型的數據，同時能夠適配多種數據處理框架。

 

 

關系型數據的列式存儲，可以將每一列的值直接排列下來，不用引入其他的概念，也不會丟失數據。

 

關系型數據的列式存儲比較好理解，而嵌套類型數據的列存儲則會遇到一些麻煩。

如圖 1 所示，我們把嵌套數據類型的一行叫做一個記錄（record)，嵌套數據類型的特點是一個 record 中的 column 除了可以是 Int, Long, String 這樣的原語（primitive）類型以外，還可以是 List, Map, Set 這樣的復雜類型。

在行式存儲中一行的多列是連續的寫在一起的，在列式存儲中數據按列分開存儲，例如可以只讀取 A.B.C 這一列的數據而不去讀 A.E 和 A.B.D，那么如何根據讀取出來的各個列的數據重構出一行記錄呢？

 

 

Google 的Dremel系統解決了這個問題，核心思想是使用“record shredding and assembly algorithm”來表示復雜的嵌套數據類型，同時輔以按列的高效壓縮和編碼技術，實現降低存儲空間，提高 IO 效率，降低上層應用延遲。

 

Parquet 就是基於 Dremel 的數據模型和算法實現的。

 

 

 

 

Parquet 適配多種計算框架

Parquet 是語言無關的，

而且不與任何一種數據處理框架綁定在一起，

適配多種語言和組件，能夠與 Parquet 配合的組件有：

• 查詢引擎: Hive, Impala, Pig, Presto, Drill, Tajo, HAWQ, IBM Big SQL
• 計算框架: MapReduce, Spark, Cascading, Crunch, Scalding, Kite
• 數據模型: Avro, Thrift, Protocol Buffers, POJOs

 

那么 Parquet 是如何與這些組件協作的呢？

這個可以通過圖 2 來說明。

數據從內存到 Parquet 文件或者反過來的過程主要由以下三個部分組成：

•  存儲格式 (storage format)

parquet-format項目定義了 Parquet 內部的數據類型、存儲格式等。

• 對象模型轉換器 (object model converters)

這部分功能由parquet-mr項目來實現，主要完成外部對象模型與 Parquet 內部數據類型的映射。

• 對象模型 (object models)

對象模型可以簡單理解為內存中的數據表示，Avro, Thrift, Protocol Buffers, Hive SerDe, Pig Tuple, Spark SQL InternalRow 等這些都是對象模型。Parquet 也提供了一個example object model 幫助大家理解。

例如parquet-mr項目里的 parquet-pig 項目就是負責把內存中的 Pig Tuple 序列化並按列存儲成 Parquet 格式，以及反過來把 Parquet 文件的數據反序列化成 Pig Tuple。

這里需要注意的是 Avro, Thrift, Protocol Buffers 都有他們自己的存儲格式，但是 Parquet 並沒有使用他們，而是使用了自己在parquet-format項目里定義的存儲格式。所以如果你的應用使用了 Avro 等對象模型，這些數據序列化到磁盤還是使用的parquet-mr定義的轉換器把他們轉換成 Parquet 自己的存儲格式。

 

 

Parquet 數據模型

理解 Parquet 首先要理解這個列存儲格式的數據模型。我們以一個下面這樣的 schema 和數據為例來說明這個問題。

message AddressBook {
 required string owner;
 repeated string ownerPhoneNumbers;
 repeated group contacts {
   required string name;
   optional string phoneNumber;
 }
}

　　

 這個 schema 中每條記錄表示一個人的 AddressBook。

　　有且只有一個 owner，

　　owner 可以有 0 個或者多個 ownerPhoneNumbers，

　　owner 可以有 0 個或者多個 contacts。

　　　　每個 contact 有且只有一個 name，

　　　　這個 contact 的 phoneNumber 可有可無。

這個 schema 可以用圖 3 的樹結構來表示。

 

 

 每個 schema 的結構是這樣的：

　　根叫做 message，message 包含多個 fields。

　　　　每個 field 包含三個屬性：repetition, type, name。

　　　　　　repetition 可以是以下三種：required（出現 1 次），optional（出現 0 次或者 1 次），repeated（出現 0 次或者多次）。

　　　　　　type 可以是一個 group 或者一個 primitive 類型。

 

 

Parquet 格式的數據類型沒有復雜的 Map, List, Set 等，而是使用 repeated fields 和 groups 來表示。

例如 List 和 Set 可以被表示成一個 repeated field，Map 可以表示成一個包含有 key-value 對的 repeated field，而且 key 是 required 的。

 

 

 

 

Parquet 文件的存儲格式

那么如何把內存中每個 AddressBook 對象按照列式存儲格式存儲下來呢？

在 Parquet 格式的存儲中，一個 schema 的樹結構有幾個葉子節點，實際的存儲中就會有多少 column。

例如上面這個 schema 的數據存儲實際上有四個 column，如圖 4 所示。

 

 

Parquet 文件在磁盤上的分布情況如圖 5 所示。

所有的數據被水平切分成 Row group，一個 Row group 包含這個 Row group 對應的區間內的所有列的 column chunk。

　　一個 column chunk 負責存儲某一列的數據，這些數據是這一列的 Repetition levels, Definition levels 和 values（詳見后文）。

　　　　一個 column chunk 是由 Page 組成的，Page 是壓縮和編碼的單元，對數據模型來說是透明的。

　　Row group 是數據讀寫時候的緩存單元，所以推薦設置較大的 Row group 從而帶來較大的並行度，當然也需要較大的內存空間作為代價。

　　　　一般情況下推薦配置一個 Row group 大小 1G，一個 HDFS 塊大小 1G，一個 HDFS 文件只含有一個塊。

一個 Parquet 文件最后是 Footer，存儲了文件的元數據信息和統計信息。

 

 

拿我們的這個 schema 為例，

　　在任何一個 Row group 內，會順序存儲四個 column chunk。

　　　　這四個 column 都是 string 類型。

　　這個時候 Parquet 就需要把內存中的 AddressBook 對象映射到四個 string 類型的 column 中。

　　如果讀取磁盤上的 4 個 column 要能夠恢復出 AddressBook 對象。這就用到了我們前面提到的 “record shredding and assembly algorithm”。

 

 

Striping/Assembly 算法

對於嵌套數據類型，我們除了存儲數據的 value 之外還需要兩個變量 Repetition Level(R), Definition Level(D) 才能存儲其完整的信息用於序列化和反序列化嵌套數據類型。

Repetition Level 和 Definition Level 可以說是為了支持嵌套類型而設計的，但是它同樣適用於簡單數據類型。

在 Parquet 中我們只需定義和存儲 schema 的葉子節點所在列的 Repetition Level 和 Definition Level。

 

Definition Level

嵌套數據類型的特點是有些 field 可以是空的，也就是沒有定義。

　　如果一個 field 是定義的，那么它的所有的父節點都是被定義的。

　　從根節點開始遍歷，當某一個 field 的路徑上的節點開始是空的時候我們記錄下當前的深度作為這個 field 的 Definition Level。

　　　　如果一個 field 的 Definition Level 等於這個 field 的最大 Definition Level 就說明這個 field 是有數據的。

　　對於 required 類型的 field 必須是有定義的，所以這個 Definition Level 是不需要的。

　　　　在關系型數據中，optional 類型的 field 被編碼成 0 表示空和 1 表示非空（或者反之）。

 

Repetition Level

記錄該 field 的值是在哪一個深度上重復的。

　　只有 repeated 類型的 field 需要 Repetition Level，optional 和 required 類型的不需要。

　　Repetition Level = 0 表示開始一個新的 record。

　　　　在關系型數據中，repetion level 總是 0。

 

 

下面用 AddressBook 的例子來說明 Striping 和 assembly 的過程。

　　對於每個 column 的最大的 Repetion Level 和 Definition Level 如圖 6 所示。

　　

下面這樣兩條 record：

AddressBook {
 owner: "Julien Le Dem",
 ownerPhoneNumbers: "555 123 4567",
 ownerPhoneNumbers: "555 666 1337",
 contacts: {
   name: "Dmitriy Ryaboy",
   phoneNumber: "555 987 6543",
 },
 contacts: {
   name: "Chris Aniszczyk"
 }
}
AddressBook {
 owner: "A. Nonymous"
}

　　以 contacts.phoneNumber 這一列為例，

　　　　"555 987 6543"這個 contacts.phoneNumber 的 Definition Level 是最大 Definition Level=2。

　　　　而如果一個 contact 沒有 phoneNumber，那么它的 Definition Level 就是 1。

　　　　如果連 contact 都沒有，那么它的 Definition Level 就是 0。

 

 

下面我們拿掉其他三個 column 只看 contacts.phoneNumber 這個 column，把上面的兩條 record 簡化成下面的樣子：

AddressBook {
 contacts: {
   phoneNumber: "555 987 6543"
 }
 contacts: {
 }
}
AddressBook {
}

　　

這兩條記錄的序列化過程如圖 7 所示：

 

如果我們要把這個 column 寫到磁盤上，磁盤上會寫入這樣的數據（圖 8）：

 

注意：NULL 實際上不會被存儲，如果一個 column value 的 Definition Level 小於該 column 最大 Definition Level 的話，那么就表示這是一個空值。

 

下面是從磁盤上讀取數據並反序列化成 AddressBook 對象的過程：

• 讀取第一個三元組 R=0, D=2, Value=”555 987 6543”

R=0 表示是一個新的 record，要根據 schema 創建一個新的 nested record 直到 Definition Level=2。

D=2 說明 Definition Level=Max Definition Level，那么這個 Value 就是 contacts.phoneNumber 這一列的值，賦值操作 contacts.phoneNumber=”555 987 6543”。

• 讀取第二個三元組 R=1, D=1

R=1 表示不是一個新的 record，是上一個 record 中一個新的 contacts。

D=1 表示 contacts 定義了，但是 contacts 的下一個級別也就是 phoneNumber 沒有被定義，所以創建一個空的 contacts。

• 讀取第三個三元組 R=0, D=0

R=0 表示一個新的 record，根據 schema 創建一個新的 nested record 直到 Definition Level=0，也就是創建一個 AddressBook 根節點。

 

 

可以看出在 Parquet 列式存儲中，

　　對於一個 schema 的所有葉子節點會被當成 column 存儲，而且葉子節點一定是 primitive 類型的數據。

　　　　對於這樣一個 primitive 類型的數據會衍生出三個 sub columns (R, D, Value)，也就是從邏輯上看除了數據本身以外會存儲大量的 Definition Level 和 Repetition Level。

　　　　　　那么這些 Definition Level 和 Repetition Level 是否會帶來額外的存儲開銷呢？

　　　　　　　　實際上這部分額外的存儲開銷是可以忽略的。

　　　　　　　　因為對於一個 schema 來說 level 都是有上限的，而且非 repeated 類型的 field 不需要 Repetition Level，required 類型的 field 不需要 Definition Level，也可以縮短這個上限。

　　　　　　　　例如對於 Twitter 的 7 層嵌套的 schema 來說，只需要 3 個 bits 就可以表示這兩個 Level 了。

　　　　對於存儲關系型的 record，record 中的元素都是非空的（NOT NULL in SQL）。

　　　　　　Repetion Level 和 Definition Level 都是 0，所以這兩個 sub column 就完全不需要存儲了。

　　　　　　所以在存儲非嵌套類型的時候，Parquet 格式也是一樣高效的。

 

 

上面演示了一個 column 的寫入和重構，那么在不同 column 之間是怎么跳轉的呢，

　　這里用到了有限狀態機的知識，詳細介紹可以參考Dremel。

 

 

 

數據壓縮算法

列式存儲給數據壓縮也提供了更大的發揮空間，除了我們常見的 snappy, gzip 等壓縮方法以外，由於列式存儲同一列的數據類型是一致的，所以可以使用更多的壓縮算法

壓縮算法	使用場景
Run Length Encoding	重復數據
Delta Encoding	有序數據集，例如 timestamp，自動生成的 ID，以及監控的各種 metrics
Dictionary Encoding	小規模的數據集合，例如 IP 地址
Prefix Encoding	Delta Encoding for strings

 

性能

Parquet 列式存儲帶來的性能上的提高在業內已經得到了充分的認可，特別是當你們的表非常寬（column 非常多）的時候，Parquet 無論在資源利用率還是性能上都優勢明顯。具體的性能指標詳見參考文檔。

Spark 已經將 Parquet 設為默認的文件存儲格式，Cloudera 投入了很多工程師到 Impala+Parquet 相關開發中，Hive/Pig 都原生支持 Parquet。

Parquet 現在為 Twitter 至少節省了 1/3 的存儲空間，同時節省了大量的表掃描和反序列化的時間。這兩方面直接反應就是節約成本和提高性能。

如果說 HDFS 是大數據時代文件系統的事實標准的話，Parquet 就是大數據時代存儲格式的事實標准。

 

 

 

參考文檔

1. http://parquet.apache.org/
2. https://blog.twitter.com/2013/dremel-made-simple-with-parquet
3. http://blog.cloudera.com/blog/2015/04/using-apache-parquet-at-appnexus/
4. http://blog.cloudera.com/blog/2014/05/using-impala-at-scale-at-allstate/