Hadoop - HDFS的數(shù)據(jù)流剖析

剖析文件的讀取

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

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

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

接著，客戶端對這個輸入流調(diào)用 read()方法（步驟3）。存儲著文件起始幾個塊的datanode地址的DFSInputStream隨即連接距離最近的datanode。通過對數(shù)據(jù)反復(fù)調(diào)用read()方法，可以將數(shù)據(jù)從datanode傳輸?shù)娇蛻舳耍ú襟E4）。到達(dá)塊的末端時(shí)，DFSInputStream關(guān)閉與該datanode的連接，然后尋找下一個塊的最佳datanode（步驟5）。客戶端只需要讀取連續(xù)的流，并且對客戶端都是透明的。

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

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

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

網(wǎng)絡(luò)拓?fù)渑cHadoop

剖析文件的寫入
接下來我們看看文件時(shí)如何寫入HDFS的，盡管比較詳細(xì)，但對于理解數(shù)據(jù)流還是很有用的，因?yàn)樗宄卣f明了HDFS的一致模型。

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

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

在客戶端寫入數(shù)據(jù)時(shí)（步驟3），DFSOutputStream將它分成一個個的數(shù)據(jù)包，并寫入內(nèi)部隊(duì)列，稱為“數(shù)據(jù)隊(duì)列”（data queue）。DataStreamer處理數(shù)據(jù)隊(duì)列，它的責(zé)任是根據(jù)datanode列表來要求namenode分配適合的新塊來存儲數(shù)據(jù)復(fù)本。這一組datanode構(gòu)成一個管線——我們假設(shè)復(fù)本數(shù)為3，所以管線中有3個節(jié)點(diǎn)。DataStreamer將數(shù)據(jù)包流式傳輸?shù)焦芫€中第一個datanode，該datanode存儲數(shù)據(jù)包并將它發(fā)送到管線中的第二個datanode。同樣，第二個datanode存儲該數(shù)據(jù)包并且發(fā)送給管線中的第三個（也是最后一個）datanote（步驟4）。

DFSOutputStream也維護(hù)著一個內(nèi)部數(shù)據(jù)包隊(duì)列來等待datanode的收到確認(rèn)回執(zhí)，稱為“確認(rèn)隊(duì)列”（ack queue）。收到管道中所有datanode確認(rèn)信息后，該數(shù)據(jù)包才會從確認(rèn)隊(duì)列刪除（步驟5）。

如果數(shù)據(jù)在寫入期間datanode發(fā)生故障，則執(zhí)行以下操作（對寫入數(shù)據(jù)的客戶端是透明的）。首先關(guān)閉管線，確認(rèn)把隊(duì)列中的所有數(shù)據(jù)包都添加回?cái)?shù)據(jù)隊(duì)列的最前端，以確保故障節(jié)點(diǎn)下游的datanode不會漏掉任何一個數(shù)據(jù)包。為存儲在另一正常datanode的當(dāng)前數(shù)據(jù)結(jié)塊指定一個新的標(biāo)識，并將該標(biāo)識傳送給namenode，以便故障datanode在恢復(fù)后可以刪除存儲的部分?jǐn)?shù)據(jù)塊。從管線中刪除故障數(shù)據(jù)節(jié)點(diǎn)并且把余下的數(shù)據(jù)塊寫入管線中另外兩個正常的datanode。namenode注意到塊復(fù)本量不足時(shí)，會在另一個節(jié)點(diǎn)上創(chuàng)建一個新的復(fù)本。后續(xù)的數(shù)據(jù)塊繼續(xù)正常接受處理。

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

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

復(fù)本怎么放