Hadoop - HDFS的數據流剖析

剖析文件的讀取

為了了解客戶端及與之交互的HDFS、namenode 和 datanode之間的數據流是什么樣的，我們可以參考下圖，該圖顯示了在讀取文件時事件的發生順序。

客戶端通過調用FileSystem對象的open()方法來打開希望讀取的文件，對于HDFS來說，這個對象是分布式文件系統（圖中步驟1）的一個實例。DistributedFileSystem 通過使用RPC來調用namenode，以確定文件起始塊的位置（步驟2）。對于每一個塊，namenode返回存有該塊副本的datanode地址。此外，這些datanode根據它們與客戶端的距離來排序。如果該客戶端本身就是一個datanode（比如，在一個MapReduce任務中），并保存有相應數據塊的一個副本時，該節點就會從本地datanode讀取數據。

DistributedFileSystem類返回一個FSDataInputStream對象（一個支持文件定位的數據流）給客戶端并讀取數據。FSDataInputStream類轉而封裝DFSInputStream對象，該對象管理著datanode和namenode的I/O。

接著，客戶端對這個輸入流調用 read()方法（步驟3）。存儲著文件起始幾個塊的datanode地址的DFSInputStream隨即連接距離最近的datanode。通過對數據反復調用read()方法，可以將數據從datanode傳輸到客戶端（步驟4）。到達塊的末端時，DFSInputStream關閉與該datanode的連接，然后尋找下一個塊的最佳datanode（步驟5）。客戶端只需要讀取連續的流，并且對客戶端都是透明的。

客戶端從流中讀取數據時，塊是按照打開DFSInputStream與datanode新建連接的順序讀取的。它也會根據需要詢問namenode來檢索下一批數據塊的datanode的位置。一旦客戶端完成讀取，就對FSDataInputStream調用close()方法（步驟6）。

在讀取數據的時候，如果DFSInputStream在與datanode通信時遇到錯誤，會嘗試從這個塊的另一個最鄰近datanode讀取數據。它也會記住那個故障datanode，以保證以后不會反復讀取該節點上后續的塊。DFSInputStream也會通過校驗和確認從datanode發來的數據是否完整。 如果發現有損壞的塊，就在DFSInputStream試圖從其他datanode讀取其副本之前通知namenode。

這個設計的一個重點是，namenode告知客戶端每個塊中最佳的datanode，并讓客戶端直接連接到該datanode檢索數據。由于數據流分散在急群眾的所有datanode，所以這種設計能使HDFS可擴展到大量的并發客戶端。同時，namenode只需要響應塊位置的請求（這些信息存儲在內存中，因而非常高效），無需響應數據請求，否則隨著客戶端數量的增長，namenode會很快成為瓶頸。

網絡拓撲與Hadoop

剖析文件的寫入
接下來我們看看文件時如何寫入HDFS的，盡管比較詳細，但對于理解數據流還是很有用的，因為它清楚地說明了HDFS的一致模型。

我們要考慮的情況是如何新建一個文件，把數據寫入該文件，最后關閉該文件。見下圖。

客戶端通過對DistributedFileSystem對象調用create()函數來新建文件（步驟1）。DistributedFileSystem對namenode創建一個RPC調用，在文件系統的命名空間中新建一個文件，此時該文件中還沒有相應的數據塊（步驟2）。namenode執行各種不同的檢查以確保這個文件不存在以及客戶端有新建文件的權限。如果這些檢查均通過，namenode就會為創建新文件記錄一條記錄；否則，文件創建失敗并向客戶端拋出一個IOException異常。DistributedFileSystem向客戶端返回一個FSDataOutputStream對象，由此客戶端可以開始寫入數據。就像讀取事件一樣，FSDataOutputStream封裝一個DFSoutPutstream對象，該對象負責處理datanode和namenode之間的通信。

在客戶端寫入數據時（步驟3），DFSOutputStream將它分成一個個的數據包，并寫入內部隊列，稱為“數據隊列”（data queue）。DataStreamer處理數據隊列，它的責任是根據datanode列表來要求namenode分配適合的新塊來存儲數據復本。這一組datanode構成一個管線——我們假設復本數為3，所以管線中有3個節點。DataStreamer將數據包流式傳輸到管線中第一個datanode，該datanode存儲數據包并將它發送到管線中的第二個datanode。同樣，第二個datanode存儲該數據包并且發送給管線中的第三個（也是最后一個）datanote（步驟4）。

DFSOutputStream也維護著一個內部數據包隊列來等待datanode的收到確認回執，稱為“確認隊列”（ack queue）。收到管道中所有datanode確認信息后，該數據包才會從確認隊列刪除（步驟5）。

如果數據在寫入期間datanode發生故障，則執行以下操作（對寫入數據的客戶端是透明的）。首先關閉管線，確認把隊列中的所有數據包都添加回數據隊列的最前端，以確保故障節點下游的datanode不會漏掉任何一個數據包。為存儲在另一正常datanode的當前數據結塊指定一個新的標識，并將該標識傳送給namenode，以便故障datanode在恢復后可以刪除存儲的部分數據塊。從管線中刪除故障數據節點并且把余下的數據塊寫入管線中另外兩個正常的datanode。namenode注意到塊復本量不足時，會在另一個節點上創建一個新的復本。后續的數據塊繼續正常接受處理。

在一個塊被寫入期間可能會有多個datanode同時發生故障，但非常少見。只要寫入了dfs.replication.min的復本數（默認為1），寫操作就會成功，并且這個塊可以在集群中異步復制，直到達到其目標復本數（dfs.replication的默認值為3）。

客戶端完成數據的寫入后，對數據流調用close()方法（步驟6）。該操作將剩余的所有數據包寫入datanode管線，并在聯系到namenode且發送文件寫入完成信號之前，等待確認（步驟7）。namenode已經知道文件由哪些塊組成（通過Datastreamer請求分配數據塊），所以它在返回成功之前只需要等待數據塊進行最小量的復制。

復本怎么放