HDFS Block Replica Placement實現原理

時間 2019-12-13

標籤 hdfs block replica placement 實現原理欄目 Hadoop 简体版

原文原文鏈接

1. 背景

Block Replica Placement——數據塊複本存儲策略，HDFS Namenode以此爲依據選取數據塊複本應存儲至哪些HDFS Datanodes，策略的設計須要權衡如下三個因素：

可靠性
寫帶寬
讀帶寬

注：本文均以數據塊複本因子爲3來討論。

咱們以兩個比較極端的例子來講明上述三個因素之間的關係。

（1）數據塊的三個複本集中存儲至一臺HDFS Datanode；

若是Client（數據寫入客戶端）與Datanode不是同一臺機器，以下圖：

數據塊的第一個複本須要Client經過網絡傳輸將數據寫入Datanode，其他兩個複本爲本地寫入，寫帶寬的開銷爲數據塊的大小；

若是Client與Datanode是同一臺機器，以下圖：

數據塊的三個複本均爲本地寫入，沒有任何的網絡數據傳輸，寫帶寬的開銷爲零。

這種策略的寫帶寬或爲數據塊大小，或爲零，能夠說是全部策略中寫帶寬開銷最小的。但缺陷也比較明顯，數據塊的三個複本所有存儲至一臺Datanode，其它Datanodes中沒有任何的冗餘複本，若是這臺Datanode出現故障，整個數據塊的數據會所有丟失；數據的集中存儲，也不利於「本地性計算」，若是這個數據塊涉及的計算任務沒法調度至這臺Datanode實例所處的機器上運行，則數據須要遠程讀取，即須要跨機架（交換機）讀取，讀帶寬開銷較大。

（2）數據塊的所有複本分散存儲至不一樣的數據中心（Data Center）；

數據塊的三個複本分別被存儲至不一樣的三個數據中心的Datanode，數據可靠性最高，但數據的寫入和讀取大多須要經過互聯網，寫帶寬和讀帶寬的開銷較大。

能夠看出，數據塊複本存儲策略的設計須要綜合考慮可靠性、寫帶寬、讀帶寬三者之間的相互影響。

爲何說跨機架（交換機）或跨數據中心的帶寬開銷較大？

跨機架的服務器之間的通訊須要經過交換機，跨數據中心的服務器之間的通訊須要經過專線鏈接，對於公司而言，交換機和專線的帶寬資源都是有限的，跨機架或數據中心的服務器之間大量的數據傳輸一般都會帶來比較高昂的帶寬成本。

2. Hadoop默認數據塊複本存儲策略

Hadoop默認數據塊複本存儲策略是以「一個數據中心、多個機架」爲基礎設計的，以下圖所示：

（1）從HDFS Cluster中隨機選取一個Datanode（Rack r1/Datanode d12）用於存儲第一個複本Replica1；

（2）從其它機架（非Rack r1）中選取一個Datanode（Rack r2/Datanode d21）用於存儲第二個複本Replica2；

（3）從機架Rack r2中隨機選取一個Datanode（Rack r2/Datanode d22）用於存儲第三個複本Replica3；

（4）若是複製因子大於3，則繼續從HDFS Cluster中隨機選取Datanode用於存儲第n（n >= 4）個複本Replica；

說明：

（1）選取的Datanode須要知足：磁盤空間使用不是不少，系統負載不是很高（Datanode不是很繁忙）；

（2）同一個機架中不要存儲同一個數據塊太多的複本；

3. 核心源碼剖析

HDFS提供的數據塊複本存儲策略類結構以下：

HDFS 默認數據塊複本存儲策略實現類：org.apache.hadoop.hdfs.server.blockmanagement.BlockPlacementPolicyDefault，它有三個很是重要的組成部分：

（1）NetworkTopology clusterMap；

（2）chooseTarget；

（3）getPipeline；

3.1 NetworkTopology

NetworkTopology（org.apache.hadoop.net.NetworkTopology）使用樹形層級結構表示集羣內部的網絡拓撲結構，以下圖：

HDFS默認數據塊複本存儲策略只考慮目前比較常見的一個數據中心（Data Center）的場景，以下圖：

網絡拓撲結構（NetworkTopology）的創建過程當中涉及到一個很是重要的問題：機架感知，即在一個集羣（數據中心）中，HDFS Namenode如何知道一個HDFS Datanode歸屬於哪一個機架位?

在HDFS的實現中，機架感知有一個接口DNSToSwitchMapping（org.apache.hadoop.net.DNSToSwitchMapping）：

其中，方法resolve用於解析機器主機名（域名）或IP地址（後續討論統一使用IP地址）對應的機架ID。也就是說，所謂的機架感知，實際是經過必定的方式根據IP地址解析出對應的機架ID。DNSToSwitchMapping有不少的實現類（機架ID的解析方式能夠有不少種），HDFS具體使用哪種實現，能夠經過屬性「net.topology.node.switch.mapping.impl」進行指定，這裏咱們僅僅介紹默認實現：ScriptBasedMapping（org.apache.hadoop.net.ScriptBasedMapping）。

ScriptBasedMapping是DNSToSwitchMapping的一種實現，容許咱們提供一個自定義的腳本，用於完成IP地址與機架ID之間的解析。自定義的腳本使用時須要咱們在配置文件core-default.xml中進行聲明，與之相關的有兩個重要的屬性：

net.topology.script.file.name：自定義腳本的絕對路徑；

net.topology.script.number.args：自定義腳本最多可接受的參數個數，默認值爲100；

ScriptBasedMapping每次解析IP地址（可能有多個）對應的機架ID時，均須要調用resolve方法，工做過程以下：

（1）將須要解析的IP地址以列表（List）的參數形式傳入resolve方法，即resolve方法能夠一次性解析多個IP地址對應的機架ID；

（2）resolve方法內部將IP地址列表轉換爲自定義腳本可接受的「多個參數」，執行自定義腳本；

（3）獲取自定義腳本的輸出，並從中解析出與IP地址列表一一對應的機架ID，並以列表（List）的形式返回；

注：受限於自定義腳本最多可接受的參數個數（net.topology.script.number.args），resolve方法內部可能須要將IP地址列表分批屢次調用自定義腳本，完成整個解析過程。

假設自定義腳本使用Python語言實現，且僅能夠接受一個參數（net.topology.script.number.args=1），腳本示例以下：

若是IP列表爲：

ip1

ip3

ip6

解析出的機架ID：

/rack1

/rack3

/rack1

自定義腳本的執行是由ShellCommandExecutor（org.apache.hadoop.util.Shell.ShellCommandExecutor）驅動的，它的內部使用java.lang.ProcessBuilder、java.lang.Process，將腳本以子進程的方式執行，而後經過java.lang.Process.getInputStream()獲取腳本的輸出內容（一行字符串），最後經過java.util.StringTokenizer從輸出內容中獲取機架ID。

ShellCommandExecutor執行自定義腳本的過程當中，須要注意如下幾個問題：

（1）腳本的執行過程是以子進程的形式進行的，高併發調用下可能帶來比較大的性能開銷；

（2）腳本須要支持接收多個參數，這裏特指多個以空格分隔的IP地址字符串；

（3）由於腳本可能接收多個IP地址（2），所以腳本輸出可能包含多個機架ID，與輸入的IP地址一一對應；如上所述，腳本的輸出內容爲一行字符串，爲了可以從中解析出多個機架ID，該字符串必須以空格、\t、\n、\r、\f之一做爲分隔符，不然java.util.StringTokenizer沒法正確解析。

綜上所述，HDFS默認使用的機架感知策略是經過咱們自定義的一個腳本實現的，這給了咱們很大的自由度，能夠根據自身的實際狀況任意調整HDFS Datanode歸屬的機架ID。

那麼，網絡拓撲結構，即NetworkTopology實例的內部數據結構，是在何時被創建的？

衆所周知，HDFS使用的是Master-Slave模式，Namenode至關於Master，Datanode至關於Slave，Datanode啓動以後須要向Namenode進行註冊，一個個Datanode向Namenode註冊的過程當中，就包含着機架感知、網絡拓撲結構的創建過程，相關代碼能夠參考：org.apache.hadoop.hdfs.server.blockmanagement.DatanodeManager.registerDatanode。

NetworkTopology實例的內部數據結構大體以下圖所示：

InnerNode與Node的代碼能夠參考：org.apache.hadoop.net.NetworkTopology.InnerNode、org.apache.hadoop.net.Node。

3.2 chooseTarget

如上文2.1中所述，BlockPlacementPolicyDefault chooseTarget的工做過程能夠簡要概括爲四步，具體實現時這四步能夠對應到四個方法：

（1）從HDFS Cluster中隨機選取一個Datanode（Rack r1/Datanode d12）用於存儲第一個複本Replica1；

chooseLocalStorage（org.apache.hadoop.hdfs.server.blockmanagement.BlockPlacementPolicyDefault.chooseLocalStorage）；

（2）從其它機架（非Rack r1）中選取一個Datanode（Rack r2/Datanode d21）用於存儲第二個複本Replica2；

chooseRemoteRack（org.apache.hadoop.hdfs.server.blockmanagement.BlockPlacementPolicyDefault.chooseRemoteRack）；

（3）從機架Rack r2中隨機選取一個Datanode（Rack r2/Datanode d22）用於存儲第三個複本Replica3；

chooseLocalRack（org.apache.hadoop.hdfs.server.blockmanagement.BlockPlacementPolicyDefault.chooseLocalRack）；

（4）若是複製因子大於3，則依次從HDFS Cluster中隨機選取Datanode用於存儲第n（n >= 4）個複本Replica；

chooseRandom（org.apache.hadoop.hdfs.server.blockmanagement.BlockPlacementPolicyDefault.chooseRandom）；

實際上，chooseLocalStorage、chooseRemoteRack、chooseLocalRack最終都會將調用請求轉發給chooseRandom進行處理，這是爲何呢？

chooseLocalStorage：從整個集羣中「隨機」選取一個Node或直接選取本機（忽略這種狀況）；

chooseRemoteRack：從整個集羣中的其它（即排除指定機架）機架中「隨機」選取一個Node；

chooseLocalRack：從集羣中的指定機架中「隨機」選取一個Node；

這三個方法的工做過程當中都涉及到「隨機」的選取過程，而chooseRandom的方法參數設計過程當中已經考慮到了上述三個狀況：

chooseLocalStorage調用chooseRandom時：

numOfReplicas：1，表示咱們須要選取多少個Node；

scope：網絡拓撲結構的根節點（root），即""；表示從整個集羣中隨機選取；

excludedNodes：空值或已經被選取的Nodes，表示選取出的Node不能出如今這些excludedNodes中；

chooseRemoteRack調用chooseRandom時：

numOfReplicas：1，表示咱們須要選取多少個Node；

scope：~rack，表示從整個集羣中非rack機架中隨機選取；

excludedNodes：空值或已經被選取的Nodes，表示選取出的Node不能出如今這些excludedNodes中；

chooseLocalRack調用chooseRandom時：

numOfReplicas：1，表示咱們須要選取多少個Node；

scope：rack，表示從集羣機架rack中隨機選取；

excludedNodes：空值或已經被選取的Nodes，表示選取出的Node不能出如今這些excludedNodes中；

chooseRandom核心源碼以下：

能夠看出，chooseRandom整個工做流程能夠理解爲一個循環選取的過程，循環條件即爲：「numOfReplicas > 0 && numOfAvailableNodes > 0」，它表示着兩個含義：

（1）numOfReplicas > 0，表示仍須要繼續選取Node用於存在數據塊複本；在咱們的討論中，它的值恆爲1；

（2）numOfAvailableNodes > 0，集羣中尚有可用的Node用於選取；它的值是經過NetworkTopology countNumOfAvailableNodes（org.apache.hadoop.net.NetworkTopology.countNumOfAvailableNodes）計算而得的；根據計算上下文（chooseLocalStorage、chooseRemoteRack、chooseLocalRack）的不一樣，它能夠計算整個集羣、整個集羣排除某機架、集羣中指定機架的可用Node數目；

每一次的選取能夠大體分爲如下幾步：

（1）從集羣網絡拓撲結構（NetworkTopology）中隨機選取一個Node firstChosen（DatanodeDescriptor，實現Node接口，用於描述HDFS Datanode信息）；

（2）若是firstChosen沒有出如今excludedNodes中，則繼續；不然，結束本次選取；

（3）獲取firstChosen中存儲信息storages（每個HDFS Datanode能夠指定多個存儲位置，每個存儲位置使用DatanodeStorageInfo表示），並對這些storages進行隨機排序（shuffle）；

（4）依次從這些storages中尋找「適合」（addIfIsGoodTarget）的存儲位置（DatanodeStorageInfo）；

也就是說，最終返回的不僅是某一個HDFS Datanode，還包括具體的存儲位置，即DatanodeStorageInfo。

NetworkTopology chooseRandom

咱們以一個具體的示例來講明集羣網絡拓撲結構的「隨機選取」的過程。

假設咱們的集羣網絡拓撲結構以下：

「隨機選取」僅僅選取「Leaf Node」（葉子節點，表示Datanode），以「深度優先」的方式依次輸出這些「Leaf Node」：

（1）整個集羣中選取；

整個集羣中可供選取的Node數目爲9，取1~9之間的隨機數，假設爲4，則選取Node的方式爲：從頭部開始，選取第4個Node便可。

（2）整個集羣中選取，且排除機架rack2；

整個集羣須要排除機架rack2包含的Nodes，可供選取的Node數目爲6，取1~6之間的隨機數，假設爲6，則選取Node的方式爲：從頭部開始，選取第6個Node便可。

（3）集羣機架rack1中選取；

集羣機架rack1中可供選取的Node數目爲4，取1~4之間的隨機數，假設爲3，則選取Node的方式爲：從頭部開始，選取第3個Node便可。

BlockPlacementPolicyDefault addIfIsGoodTarget

如前所述，每個Datanode能夠包含多個存儲位置（Datanode指定多個目錄用於存儲數據塊複本，每個目錄掛載一塊磁盤），選取Datanode完成以後，還須要從中尋找出「合適」的存儲位置（DatanodeStorageInfo），若是從這個Datanode中沒法找到「合適」的存儲位置，則須要繼續選取Node的過程。

「合適」包括七個方面，源碼以下：

（1）存儲類型是否匹配，SSD/DISK；

（2）存儲是否爲只讀；

（3）存儲所處的Datanode是否處於下線狀態；

（4）存儲所處的Datanode狀態是否異常；

（5）存儲空間是否足夠；

（6）存儲所處的Datanode負載是否太高；

（7）存儲所處的Datanode歸屬的機架中存儲同一數據塊的複本的Datanodes數目是否超過設置maxNodesPerRack；

注：（7）中maxNodesPerRack的計算公式：int maxNodesPerRack = (totalNumOfReplicas-1)/clusterMap.getNumOfRacks()+2;

3.3 getPipeline

用於存儲數據塊複本的Nodes選取完畢以後，還須要將Client與這些Nodes造成一個「管道」，並且這個「管道」的「節點距離之和」（數據傳輸距離）必須最短，getPipeline就是用來實現這個「管道」的造成過程的，實際就是一個排序的過程。

注：關於「節點距離」的概念可參考<<Hadoop The Definitive Guide>>章節：「Anatomy of a File Read」。