HDFS學習筆記（6）AVRO
一、引言
1、 簡介
Avro是Hadoop中的一個子專案，也是Apache中一個獨立的專案，Avro是一個基於二進位制資料傳輸高效能的中介軟體。在Hadoop的其他專案中例如HBase(Ref)和Hive(Ref)的Client端與服務端的資料傳輸也採用了這個工具。Avro是一個數據序列化的系統。Avro 可以將資料結構或物件轉化成便於儲存或傳輸的格式。Avro設計之初就用來支援資料密集型應用，適合於遠端或本地大規模資料的儲存和交換。
2、 特點
Ø 豐富的資料結構型別；
Ø 快速可壓縮的二進位制資料形式，對資料二進位制序列化後可以節約資料儲存空間和網路傳輸頻寬；
Ø 儲存持久資料的檔案容器；
Ø 可以實現遠端過程呼叫RPC；
Ø 簡單的動態語言結合功能。
avro支援跨程式語言實現（C, C++, C#，Java, Python, Ruby, PHP），類似於Thrift，但是avro的顯著特徵是：avro依賴於模式，動態載入相關資料的模式，Avro資料的讀寫操作很頻繁，而這些操作使用的都是模式，這樣就減少寫入每個資料檔案的開銷，使得序列化快速而又輕巧。這種資料及其模式的自我描述方便了動態指令碼語言的使用。當Avro資料儲存到檔案中時，它的模式也隨之儲存，這樣任何程式都可以對檔案進行處理。如果讀取資料時使用的模式與寫入資料時使用的模式不同，也很容易解決，因為讀取和寫入的模式都是已知的。
|
New schema |
Writer |
Reader |
Action |
|
Added field |
Old |
New |
The reader uses the default value of the new field, since it is not written by the writer. |
|
New |
Old |
The reader does not know about the new field written by the writer, so it is ignored (projection). |
|
|
Removed field |
Old |
New |
The reader ignores the removed field (projection). |
|
New |
Old |
The removed field is not written by the writer. If the old schema had a default defined for the field, the reader uses this; otherwise, it gets an error. In this case, it is best to update the reader’s schema, either at the same time as or before the writer’s. |
Avro和動態語言結合後，讀/寫資料檔案和使用RPC協議都不需要生成程式碼，而程式碼生成作為一種可選的優化只需要在靜態型別語言中實現。
Avro依賴於模式(Schema)。通過模式定義各種資料結構，只有確定了模式才能對資料進行解釋，所以在資料的序列化和反序列化之前，必須先確定模式的結構。正是模式的引入，使得資料具有了自描述的功能，同時能夠實現動態載入，另外與其他的資料序列化系統如Thrift相比，資料之間不存在其他的任何標識，有利於提高資料處理的效率。
二、技術要領
1、 型別
資料型別標準化的意義：一方面使不同系統對相同的資料能夠正確解析，另一方面，資料型別的標準定義有利於資料序列化/反序列化。
簡單的資料型別：Avro定義了幾種簡單資料型別，下表是其簡單說明：
|
型別 |
說明 |
|
null |
no value |
|
boolean |
a binary value |
|
int |
32-bit signed integer |
|
long |
64-bit signed integer |
|
float |
single precision (32-bit) IEEE 754 floating-point number |
|
double |
double precision (64-bit) IEEE 754 floating-point number |
|
bytes |
sequence of 8-bit unsigned bytes |
|
string |
unicode character sequence |
簡單資料型別由型別名稱定義，不包含屬性資訊，例如字串定義如下：
{"type": "string"}
複雜資料型別：Avro定義了六種複雜資料型別，每一種複雜資料型別都具有獨特的屬性，下表就每一種複雜資料型別進行說明。
|
型別 |
屬性 |
說明 |
|
Records |
type name |
record |
|
name |
a JSON string providing the name of the record (required). |
|
|
namespace |
a JSON string that qualifies the name(optional). |
|
|
doc |
a JSON string providing documentation to the user of this schema (optional). |
|
|
aliases |
a JSON array of strings, providing alternate names for this record (optional). |
|
|
fields |
a JSON array, listing fields (required). |
|
|
name |
a JSON string. |
|
|
type |
a schema/a string of defined record. |
|
|
default |
a default value for field when lack. |
|
|
order |
ordering of this field. |
|
|
Enums |
type name |
enum |
|
name |
a JSON string providing the name of the enum (required). |
|
|
namespace |
a JSON string that qualifies the name. |
|
|
doc |
a JSON string providing documentation to the user of this schema (optional). |
|
|
aliases |
a JSON array of strings, providing alternate names for this enum (optional) |
|
|
symbols |
a JSON array, listing symbols, as JSON strings (required). All symbols in an enum must be unique. |
|
|
Arrays |
type name |
array |
|
items |
the schema of the array’s items. |
|
|
Maps |
type name |
map |
|
values |
the schema of the map’s values. |
|
|
Fixed |
type name |
fixed |
|
name |
a string naming this fixed (required). |
|
|
namespace |
a string that qualifies the name. |
|
|
aliases |
a JSON array of strings, providing alternate names for this enum (optional). |
|
|
size |
an integer, specifying the number of bytes per value (required). |
|
|
Unions |
a JSON arrays |
每一種複雜資料型別都含有各自的一些屬性，其中部分屬性是必需的，部分是可選的。
這裡需要說明Record型別中field屬性的預設值，當Record Schema例項資料中某個field屬性沒有提供例項資料時，則由預設值提供，具體值見下表。Union的field預設值由Union定義中的第一個Schema決定。
|
avro type |
json type |
example |
|
null |
null |
null |
|
boolean |
boolean |
true |
|
int,long |
integer |
1 |
|
float,double |
number |
1.1 |
|
bytes |
string |
"\u00FF" |
|
string |
string |
"foo" |
|
record |
object |
{"a": 1} |
|
enum |
string |
"FOO" |
|
array |
array |
[1] |
|
map |
object |
{"a": 1} |
|
fixed |
string |
"\u00ff" |
2、 序列化/反序列化
Avro指定兩種資料序列化編碼方式：binary encoding 和Json encoding。使用二進位制編碼會高效序列化，並且序列化後得到的結果會比較小；而JSON一般用於除錯系統或是基於WEB的應用。
binary encoding規則如下：
1、 簡單資料型別
|
Type |
Encoding |
Example |
|
null |
Zero bytes |
Null |
|
boolean |
A single byte |
{true:1, false:0} |
|
int/long |
variable-length zig-zag coding |
|
|
float |
4 bytes |
Java's floatToIntBits |
|
double |
8 bytes |
Java's doubleToLongBits |
|
bytes |
a long followed by that many bytes of data |
|
|
string |
a long followed by that many bytes of UTF-8 encoded character data |
“foo”:{3,f,o,o} 06 66 6f 6f |
2、 複雜資料型別
|
Type |
encoding |
|
Records |
encoded just the concatenation of the encodings of its fields |
|
Enums |
a int representing the zero-based position of the symbol in the schema |
|
Arrays |
encoded as series of blocks. A block with count 0 indicates the end of the array. block:{long,items} |
|
Maps |
encoded as series of blocks. A block with count 0 indicates the end of the map. block:{long,key/value pairs}. |
|
Unions |
encoded by first writing a long value indicating the zero-based position within the union of the schema of its value. The value is then encoded per the indicated schema within the union. |
|
fixed |
encoded using number of bytes declared in the schema |
例項：
Ø records
{
"type":"record",
"name":"test",
"fields" : [
{"name": "a","type": "long"},
{"name": "b","type": "string"}
]
}
假設：a=27b=”foo” (encoding：36(27), 06(3), 66("f"), 6f("o"))
binary encoding：3606 66 6f 6f
Ø enums
{"type": "enum","name": "Foo", "symbols": ["A","B", "C", "D"] }
“D”(encoding: 06(3))
binary encoding: 06
Ø arrays
{"type": "array","items": "long"}
設：{3, 27 } (encoding：04(2), 06(3), 36(27) )
binary encoding：0406 36 00
Ø maps
設：{("a":1), ("b":2) } （encoding：61(“a”), 62(“b”), 02(1), 04(2)）
binary encoding：0261 02 02 62 04
Ø unions
["string","null"]
設：(1)null; (2) “a”
binary encoding：
(1) 02；說明：02代表null在union定義中的位置1；
(2) 00 02 61；說明：00為string在union定義的位置，02 61為”a”的編碼。
圖1表示的是Avro本地序列化和反序列化的例項，它將使用者定義的模式和具體的資料編碼成二進位制序列儲存在物件容器檔案中，例如使用者定義了包含學號、姓名、院系和電話的學生模式，而Avro對其進行編碼後儲存在student.db檔案中，其中儲存資料的模式放在檔案頭的元資料中，這樣讀取的模式即使與寫入的模式不同，也可以迅速地讀出資料。假如另一個程式需要獲取學生的姓名和電話，只需要定義包含姓名和電話的學生模式，然後用此模式去讀取容器檔案中的資料即可。
圖表 1
3、 模式Schema
Schema通過JSON物件表示。Schema定義了簡單資料型別和複雜資料型別，其中複雜資料型別包含不同屬性。通過各種資料型別使用者可以自定義豐富的資料結構。
Schema由下列JSON物件之一定義：
1. JSON字串：命名
2. JSON物件：{“type”: “typeName” …attributes…}
3. JSON陣列：Avro中Union的定義
舉例：
{"namespace": "example.avro",
"type":"record",
"name":"User",
"fields": [
{"name":"name", "type": "string"},
{"name":"favorite_number", "type": ["int", "null"]},
{"name":"favorite_color", "type": ["string","null"]}
]
}
4、 排序
Avro為資料定義了一個標準的排列順序。比較在很多時候是經常被使用到的物件之間的操作，標準定義可以進行方便有效的比較和排序。同時標準的定義可以方便對Avro的二進位制編碼資料直接進行排序而不需要反序列化。
只有當資料項包含相同的Schema的時候，資料之間的比較才有意義。資料的比較按照Schema深度優先，從左至右的順序遞迴的進行。找到第一個不匹配即可終止比較。
兩個擁有相同的模式的項的比較按照以下規則進行：
null：總是相等。
int,long,float：按照數值大小比較。
boolean：false在true之前。
string：按照字典序進行比較。
bytes，fixed：按照byte的字典序進行比較。
array：按照元素的字典序進行比較。
enum：按照符號在列舉中的位置比較。
record：按照域的字典序排序，如果指定了以下屬性：
“ascending”，域值的順序不變。
“descending”，域值的順序顛倒。
“ignore”，排序的時候忽略域值。
map：不可進行比較。
5、 物件容器檔案
Avro定義了一個簡單的物件容器檔案格式。一個檔案對應一個模式，所有儲存在檔案中的物件都是根據模式寫入的。物件按照塊進行儲存，塊可以採用壓縮的方式儲存。為了在進行mapreduce處理的時候有效的切分檔案，在塊之間採用了同步記號。一個檔案可以包含任意使用者定義的元資料。
一個檔案由兩部分組成：檔案頭和一個或者多個檔案資料塊。
檔案頭：
Ø 四個位元組，ASCII‘O’，‘b’，‘j’,1。
Ø 檔案元資料，用於描述Schema。
Ø 16位元組的檔案同步記號。
Ø 其中，檔案元資料的格式為：
i. 值為-1的長整型，表明這是一個元資料塊。
ii. 標識塊長度的長整型。
iii. 標識塊中key/value對數目的長整型。
iv. 每一個key/value對的string key和bytesvalue。
v. 標識塊中位元組總數的4位元組長的整數。
檔案資料塊：
資料是以塊結構進行組織的，一個檔案可以包含一個或者多個檔案資料塊。
Ø 表示檔案中塊中物件數目的長整型。
Ø 表示塊中資料序列化後的位元組數長度的長整型。
Ø 序列化的物件。
Ø 16位元組的檔案同步記號。
當資料塊的長度為0時即為檔案資料塊的最後一個數據，此後的所有資料被自動忽略。
下圖示物件容器檔案的結構分解及說明：
一個儲存檔案由兩部分組成:頭資訊(Header)和資料塊(Data Block)。而頭資訊又由三部分構成：四個位元組的字首，檔案Meta-data資訊和隨機生成的16位元組同步標記符。Avro目前支援的Meta-data有兩種：schema和codec。
codec表示對後面的檔案資料塊(File Data Block)採用何種壓縮方式。Avro的實現都需要支援下面兩種壓縮方式：null(不壓縮)和deflate(使用Deflate演算法壓縮資料塊)。除了文件中認定的兩種Meta-data，使用者還可以自定義適用於自己的Meta-data。這裡用long型來表示有多少個Meta-data資料對，也是讓使用者在實際應用中可以定義足夠的Meta-data資訊。對於每對Meta-data資訊，都有一個string型的key(需要以“avro.” 為字首)和二進位制編碼後的value。對於檔案中頭資訊之後的每個資料塊，有這樣的結構：一個long值記錄當前塊有多少個物件，一個long值用於記錄當前塊經過壓縮後的位元組數，真正的序列化物件和16位元組長度的同步標記符。由於物件可以組織成不同的塊，使用時就可以不經過反序列化而對某個資料塊進行操作。還可以由資料塊數，物件數和同步標記符來定位損壞的塊以確保資料完整性。
三、RPC實現
當在RPC中使用Avro時，伺服器和客戶端可以在握手連線時交換模式。伺服器和客戶端有彼此全部的模式，因此相同命名欄位、缺失欄位和多餘欄位等資訊之間通訊中需要處理的一致性問題就可以容易解決。如圖2所示，協議中定義了用於傳輸的訊息，訊息使用框架後放入緩衝區中進行傳輸，由於傳輸的初始就交換了各自的協議定義，因此即使傳輸雙方使用的協議不同所傳輸的資料也能夠正確解析。
圖表 2
Avro作為RPC框架來使用。客戶端希望同伺服器端互動時，就需要交換雙方通訊的協議，它類似於模式，需要雙方來定義，在Avro中被稱為訊息(Message)。通訊雙方都必須保持這種協議，以便於解析從對方傳送過來的資料，這也就是傳說中的握手階段。
訊息從客戶端傳送到伺服器端需要經過傳輸層(Transport Layer)，它傳送訊息並接收伺服器端的響應。到達傳輸層的資料就是二進位制資料。通常以HTTP作為傳輸模型，資料以POST方式傳送到對方去。在 Avro中，它的訊息被封裝成為一組緩衝區(Buffer)，類似於下圖的模型：
如上圖，每個緩衝區以四個位元組開頭，中間是多個位元組的緩衝資料，最後以一個空緩衝區結尾。這種機制的好處在於，傳送端在傳送資料時可以很方便地組裝不同資料來源的資料，接收方也可以將資料存入不同的儲存區。還有，當往緩衝區中寫資料時，大物件可以獨佔一個緩衝區，而不是與其它小物件混合存放，便於接收方方便地讀取大物件。
物件容器檔案是Avro的資料儲存的具體實現，資料交換則由RPC服務提供，與物件容器檔案類似，資料交換也完全依賴Schema，所以與Hadoop目前的RPC不同，Avro在資料交換之前需要通過握手過程先交換Schema。
1、 握手過程
握手的過程是確保Server和Client獲得對方的Schema定義，從而使Server能夠正確反序列化請求資訊，Client能夠正確反序列化響應資訊。一般的，Server/Client會快取最近使用到的一些協議格式，所以，大多數情況下，握手過程不需要交換整個Schema文字。
所有的RPC請求和響應處理都建立在已經完成握手的基礎上。對於無狀態的連線，所有的請求響應之前都附有一次握手過程；對於有狀態的連線，一次握手完成，整個連線的生命期內都有效。
具體過程：
Client發起HandshakeRequest，其中含有Client本身SchemaHash值和對應Server端的Schema Hash值(clientHash!=null,clientProtocol=null, serverHash!=null)。如果本地快取有serverHash值則直接填充，如果沒有則通過猜測填充。
Server用如下之一HandshakeResponse響應Client請求：
(match=BOTH, serverProtocol=null,serverHash=null)：當Client傳送正確的serverHash值且Server快取相應的clientHash。握手過程完成，之後的資料交換都遵守本次握手結果。
(match=CLIENT, serverProtocol!=null,serverHash!=null)：當Server快取有Client的Schema，但是Client請求中ServerHash值不正確。此時Server傳送Server端的Schema資料和相應的Hash值，此次握手完成，之後的資料交換都遵守本次握手結果。
(match=NONE)：當Client傳送的ServerHash不正確且Server端沒有Client Schema的快取。這種情況下Client需要重新提交請求資訊 (clientHash!=null,clientProtocol!=null, serverHash!=null)，Server響應 (match=BOTH, serverProtocol=null,serverHash=null)，此次握手過程完成，之後的資料交換都遵守本次握手結果。
握手過程使用的Schema結構如下示。
{
"type":"record",
"name":"HandshakeRequest","namespace":"org.apache.avro.ipc",
"fields":[
{"name":"clientHash", "type": {"type": "fixed","name": "MD5", "size": 16}},
{"name":"clientProtocol", "type": ["null","string"]},
{"name":"serverHash", "type": "MD5"},
{"name":"meta", "type": ["null", {"type":"map", "values": "bytes"}]}
]
}
{
"type":"record",
"name":"HandshakeResponse", "namespace":"org.apache.avro.ipc",
"fields":[
{"name":"match","type": {"type": "enum","name": "HandshakeMatch",
"symbols":["BOTH", "CLIENT", "NONE"]}},
{"name":"serverProtocol", "type": ["null","string"]},
{"name":"serverHash","type": ["null", {"type":"fixed", "name": "MD5", "size": 16}]},
{"name":"meta","type": ["null", {"type":"map", "values": "bytes"}]}
]
}
2、 訊息幀格式
訊息從客戶端傳送到伺服器端需要經過傳輸層，它傳送請求並接收伺服器端的響應。到達傳輸層的資料就是二進位制資料。通常以HTTP作為傳輸模型，資料以POST方式傳送到對方去。在 Avro中訊息首先分幀後被封裝成為一組緩衝區(Buffer)。
資料幀的格式如下：
Ø 一系列Buffer：
1、4位元組的Buffer長度
2、Buffer位元組資料
Ø 長度為0的Buffer結束資料幀
3、 Call格式
一個呼叫由請求訊息、結果響應訊息或者錯誤訊息組成。請求和響應包含可擴充套件的元資料，兩種訊息都按照之前提出的方法分幀。
呼叫的請求格式為：
Ø 請求元資料，一個型別值的對映。
Ø 訊息名，一個Avro字串。
Ø 訊息引數。引數根據訊息的請求定義序列化。
呼叫的響應格式為：
Ø 響應的元資料，一個型別值的對映。
Ø 一位元組的錯誤標誌位。
Ø 如果錯誤標誌為false，響應訊息，根據響應的模式序列化。
如果錯誤標誌位true，錯誤訊息，根據訊息的錯誤聯合模式序列化。
四、例項
1、 本地序列化/反序列化
user.avsc
{"namespace":"example.avro",
"type": "record",
"name": "User",
"fields": [
{"name": "name", "type":"string"},
{"name": "favorite_number", "type": ["int", "null"]},
{"name": "favorite_color", "type":["string", "null"]}
]
}
Main.java
public class Main {
public static void main(String[] args)throws Exception {
User user1 = new User();
user1.setName("Alyssa");
user1.setFavoriteNumber(256);
// Leave favorite color null
// Alternate constructor
User user2 = new User("Ben", 7,"red");
// Construct via builder
User user3 = User.newBuilder()
.setName("Charlie")
.setFavoriteColor("blue")
.setFavoriteNumber(null)
.build();
// Serialize user1 and user2to disk
File file = new File("users.avro");
DatumWriter<User> userDatumWriter = new SpecificDatumWriter<User>(User.class);
DataFileWriter<User> dataFileWriter = newDataFileWriter<User>(userDatumWriter);
dataFileWriter.create(user1.getSchema(),new File("users.avro"));
dataFileWriter.append(user1);