深入理解Hadoop集群和網絡
本文側重于Hadoop集群的體系結構和方法，以及它與網絡和服務器基礎設施這件的關系。文章的素材主要來自于研究工作以及同現實生活中運行Hadoop集群客戶的討論。如果你也在你的數據中心運行產品級的Hadoop集群，那么我希望你能寫下有價值的評論。
Hadoop集群部署時有三個角色：Client machines, Master nodes和Slave nodes。
Master nodes負責Hadoop的兩個關鍵功能：數據存儲（HDFS）；以及運行在這個數據之上的并行計算，又稱為Map-Reduce。Name node負責調度數據存儲，而Job Tracker則負責并行數據處理的調度（使用Map-Reduce技術）。Slave nodes由大量的機器組成，完成數據存儲以及運行計算這樣的臟活。每個slave node都運行Data node和Task Tracker daemon，這些slave daemon和master nodes的相應daemon進行通信。Task tracker daemon由Job Tracker管理，Data node Daemon由Name node管理。
Client機器包含了Hadoop集群的所有設置，但是它既不是Master也不是Slave。Client的角色是向集群保存數據，提交 Map-Reduce jobs（描述如何處理數據），獲取查看MR jobs的計算結果。在小型集群中（40節點）你可能會發現一個物理機器扮演多個角色，比如既是Job Tracker又是Name node，在中等或者大規模集群中，一般都是用獨立的服務器負責單獨的角色。
在真正的產品集群中，不存在虛擬服務器和虛擬機平臺，因為他們僅會導致不必要的性能損耗。Hadoop最好運行在linux機器上，直接工作于底層硬件之上。換句話說，Hadoop可以工作在虛擬機之上，對于學習Hadoop是一個不錯的廉價方法，我本身就有一個6-node的Hadoop cluster運行Windows 7 laptop的VMware Workstation之上
上圖是Hadoop集群的典型架構。機架服務器（不是刀鋒服務器）分布在多個機架中，連接到一個1GB(2GB)帶寬的機架交換機，機架交換機連接到上一層的交換機，這樣所有的節點通過最上層的交換機連接到一起。大部分的服務器都是Slave nodes配置有大的磁盤存儲 中等的CPU和DRAM，少部分是Master nodes配置較少的存儲空間 但是有強大的CPU和大DRAM
本文中，我們不討論網絡設計的各種細節，而是把精力放在其他方面。首先我們看下應用的工作流程。
為什么會出現Hadoop？ 它又解決了什么問題？ 簡單的說，是由于商業和政府有大量的數據需要快速分析和處理，如果我們把這些巨大的數據切分為小的數據塊分散到多臺計算機中，并且用這些計算機并行處理分配給他們的小塊數據，可以快速的得到結果，這就是Hadoop能做的事情。
在我們的簡單例子中，我們有一個大文件保存著所有發給客戶服務部分的郵件，想要快速統計單詞"Refund"被輸入了多少次，這個結果有助于評估對退換貨部門的需求，以指定相應對策。這是一個簡單的單詞計數練習。Client上傳數據到集群(File.txt)，提交一個job來描述如何分析數據，集群存儲結果到一個新文件(Result.txt)，然后Client讀取結果文件。
沒有加載數據，Hadoop集群就沒有任何用途。所以我們首先從加載大文件File.txt到集群中以便處理。目標是能夠快速并行處理許多數據，為了達到這個目的需要盡可能多的機器能夠同時操縱文件數據。Client把數據文件切分成許多小blocks，然后把他們分散到集群的不同機器上。塊數越多，用來處理這些數據的機器就越多。同時這些機器一定會失效，所以要確保每個數據塊保存到多個機器上以防止數據丟失。所以每個數據塊在存入集群時會進行復制。Hadoop的標準設定是集群中的每個block有三份copy，這個配置可以由hdfs-site.xml的dfs.replication 參數設置
Client把文件File.txt分成3塊，對于每一塊，Client和Name node協商獲取可以保存它以及備份塊的Data nodes列表。Client然后直接寫block數據到Data node，Data node負責寫入數據并且復制copy到其他的Data nodes。重復操作，直到所有的塊寫完。Name node本身不參與數據保存，Name node僅僅提供文件數據位置以及數據可用位置（文件系統元數據）
Hadoop引入了Rack Awareness的概念。作為Hadoop系統管理員，你能夠手動定義集群中每一個slave Data node的rack號。為什么要把自己置于這種麻煩呢？有兩個關鍵的原因：數據丟失和網絡性能。記住每一個數據塊都會被復制到多臺Data node上，以防止某臺機器失效導致數據丟失。有沒有可能一份數據的所有備份恰好都在同一個機架上，而這個機架又出現了故障，比如交換機失效或者電源失效。為了防止這種情況，Name node需要知道Data nodes在網絡拓撲中的位置，并且使用這個信息決定數據應該復制到集群的什么位置。
我們還假定同一機架的兩臺機器間相比不同機架的兩臺機器間 有更高的帶寬和更低的網絡延遲，在大部分情況下是正確的。機架交換機的上行帶寬通常小于下行帶寬。此外，機架內延遲通常低于機架間延遲。如果上面的假定成立，那么采用Rack Awareness既可以通過優化位置保護數據，同時提升網絡性能，豈不是太cool了。是的，的確是這樣，非常cool，對不對。
別急，Not cool的是Rack Awareness是需要手工定義的，并且要持續的更新它，并且保證這個信息精確。如果機架交換機可以自動提供它下面的Data node列表給Name node，那就更cool了。或者如果Data nodes可以自動告知Name node他們屬于哪個交換機，一樣很cool.
此外，讓人感興趣的是OpenFlow 網絡，Name node可以請求OpenFlow控制器關于節點位置的拓撲情況。
Client準備寫文件File.txt到集群時，把文件劃分為塊，塊A為第一個塊。Client向Name node申請寫文件File.txt，從Name node獲得許可，并且接收到一個Data nodes列表（3個表項），每一個Data nodes用來寫入塊A的一個copy。Name node使用Rack Awareness來決定這個Data nodes列表。規則是：對于每一個數據塊，兩個 copy存放在一個機架上，另外一個copy存放在另外一個機架上。
#p#
在Client寫Block A之前，它要知道準備接受”Block A“ copy的Data nodes是否以及做好了準備。首先，Client選擇list中的第一個節點Data Node1，打開一個TCP50010鏈接然后請求：“hey，準備接受一個block，這是一個Data nodes列表（2個表項），Data node5和Data node6”，請確保他們兩個也準備好了“；于是Data Node1打開一個到Data node5的TCP500100連接然后說：”Hey，準備接受一個block，這是一個Data nodes列表（1個表項），Data node6”，請確保他準備好了“；Data Node5同樣會問Data Node6：“Hey, 準備好接收一個block嗎“
”準備就緒“的響應通過已經創建好的TCP pipeline傳回來，直到Data Node1發送一個"Ready"給Client。現在Client可以開始寫入數據了。
寫入數據的過程中，在涉及寫操作的Data nodes之間創建一個復制pipeline。也就是說一個數據節點接收數據的同時，同時會把這份數據通過pipeline push到下一個Node中。
從上圖可以看到，Rack Awareness起到了改善集群性能的做用。Data node5和Data node6在同一個機架上，因此pipeline的最后一步復制是發生在機架內，這就受益于機架內帶寬以及低延遲。在Data node1, Data node5, Data node6完成block A之前，block B的操作不會開始。
當三個Nodes成功的接收到Block A后，揮發送"Block received"報告給Name node，同時發送"Success"到pipeline中，然后關閉TCP事務。Client在接收到Success信息后，通知Name node數據塊已經成功的寫入。Name node更新File.txt中Block A的metadata信息（包含Name locations信息）
Client現在可以開始Block B的傳輸了
隨著File.txt的塊被寫入，越來越多的Data nodes涉及到pipeline中，散落到機架內的熱點，以及跨機架的復制
Hadoop占用了很多的網絡帶寬和存儲空間。Hadoop專為處理大文件而生，比如TB級尺寸的文件。每一個文件在網絡和存儲中都被復制了三次。如果你有一個1TB的文件，那么將消耗3TB的網絡帶寬，同時要消耗3TB的磁盤空間存貯這個文件。
隨著每塊的復制pipeline的完成，文件被成功的寫入集群。文件散落在集群內的機器上，每個機器保存文件的一小部分數據。組成文件的塊數目越多，數據散落的機器就越多，將來更多的CPU和磁盤驅動器就能參與到并行處理中來，提供更強大更快的處理能力。這也是建造巨大集群的原動力。當機器的數變多，集群則變得wide，網絡也相應的需要擴展。
擴展集群的另外一種方法是deep擴展。就是維持機器數據不便，而是增加機器的CPU處理能力和磁盤驅動器的數目。在這種情況下，需要提高網絡的I/O吞吐量以適應增大的機器處理能力，因此如何讓Hadoop集群運行10GB nodes稱為一個重要的考慮。
Name node保存集群內所有文件的metadata，監督Data nodes的健康以及協調數據的存取。Name node是HDFS的控制中心。它本身并不保存任何cluster data。Name node知道一個文件由哪些塊組成以及這些塊存放在集群內的什么地方。Name node告訴Client需要和哪些Data node交互，管理集群的存儲容量，掌握Data node的健康狀況，確保每一個數據塊都符合系統備份策略。
Data node每3秒鐘發送一個heartbeats給Name node ，二者使用TCP9000端口的TCP握手來實現heartbeats。每十個heartbeats會有一個block report，Data node告知它所保存的數據塊。block report使得Namenode能夠重建它的metadata以確保每個數據block有足夠的copy，并且分布在不同的機架上。
Name node是Hadoop Distributed File System(HDFS)的關鍵部件。沒有Name node，clients無法從HDFS讀寫數據，也無法執行Map Reduce jobs。因此Name node 最好配置為一臺高冗余的企業級服務器：雙電源，熱插拔風扇，冗余NIC連接等。
如果Name node收不到某Data node的heartbeats，那么Name node假定這個Data node死機并且Data node上的所有數據也丟失了。通過這臺dead Data node的block report，Name node知道哪些block copies需要復制到其他Data nodes。Name node參考Rack Awareness數據來選擇接收新copy的Data node，并遵守以下復制規則：一個機架保存兩份copies，另一個機架保存第三份copy。
考慮由于機架交換機或者機架電源失敗導致的整個機架Data node都失效的情況。Name node將指導集群內的剩余Data nodes開始復制失效機架上的所有數據。如果失效機架上服務器的存儲容量為12TB，那么這將導致數百TB的數據在網絡中傳輸。
Secondary Name node是Hadoop的一種服務器角色。一個很普遍的誤解是它提供了對Name node的高可用性備份，實際上不是。
Secondary Name node偶爾會連接到Name node(缺省為每小時)，同步Name node in-memory metadata以及保存metadata的文件。Secondary Name node合并這些信息到一個組新的文件中，保存到本地的同時把這些文件發送回Name Node。
當Name node宕機，保存在Secondary Name node中的文件可以用來恢復Name node。在一個繁忙的集群中，系統管理員可以配置同步時間為更小的時間間隔，比如每分鐘。
當一個Client想要從HDFS獲取一個文件時，比如job的輸出結果。Client首先從Name node查詢文件block的位置。Name node返回一個包含所有block位置的鏈表，每個block位置包含全部copies所在的Data node
Client選取一個數據塊的Data node位置，通過TCP50010端口從這個Data node讀取一塊，在讀取完當前塊之前，Client不會處理下一塊。
在某些情況下Data node daemon本身需要從HDFS讀取數據塊。比如Data Node被請求處理自身不存在的數據，因而它必須從網絡上的其他Data node獲得數據然后才能開始處理。
#p#
另外一種情況是Name node的Rack Awareness信息提供的網絡優化行為。當Data node向Name node查詢數據塊位置信息，Name node優先查看請求者所在的機架內的Data nodes包含這個數據塊。如果包含，那么Name node把這個Data node提供給請求的Data node。這樣可以保證數據僅在in-rack內流動，可以加快數據的處理速度以及job的完成速度
現在File.txt分散到集群的機器中這樣就可以提供更快更有效的并行處理速度。Hadoop并行處理框架稱為Map Reduce，名稱來自于并行處理的兩個重要步驟：Map和Reduce
第一步是Map 過程，這個步驟同時請求所有包含數據的Data node運行計算。在我們的例子中則是請求這些Data node統計存儲在他們上的File.txt數據塊包含多少此Refund
要達到這個目的，Client首先提交Map Reduce job給Job tracker，發送請求：“How many times does Refund occur in file.txt”。Job tracker向Name node查詢哪些Data nodes包含文件File.txt的數據塊。然后Job Tracker在這些Data nodes上運行Java代碼 在Data node的本地數據上執行Map計算。Task Tracker啟動一個Map task監測這些tasks的執行。Task Tracker通過heartbeats向Job Tracker匯報task的狀態。
當每一個Map task都完成后，計算結果保存在這些節點的臨時存儲區內，我們稱之為"intermediate data"。下一步是把這些中間數據通過網絡發送給運行Reduce的節點以便完成最后的計算。
Job tracker總是嘗試選擇包含待處理數據的Data node做Map task，但是有時不會這樣。比如，所有包含這塊數據的Data node已經有太多的tasks正在運行，不再接收其他的task.
這種情況下，Job Tracker將詢問Name node，Name node會根據Rack Awareness建議一個in-rack Data node。Job tracker把task分配給這個in-rack Data node。這個Data node會在Name node的指導下從包含待處理數據的in-rack Data node獲取數據。
Map Reduce 框架的第二部分叫做Reduce。Map task已經完成了計算，計算結果保存在intermediate data。現在我們需要把所有的中間數據匯集到一起作進一步處理得到最終結果。
Job Tracker可以在集群內的任意一個node上執行Reduce，它指導Reduce task從所有完成Map trasks的Data node獲取中間數據。這些Map tasks可能同時響應Reducer，這就導致了很多nodes幾乎同時向單一節點發起TCP連接。我們稱之為incast或者fan-in(微突發流)。如果這種微突發流比較多，那么就要求網絡交換機有良好的內部流量管理能力，以及相應的buffers。這種間歇性的buffers使用可能會影響其他的網絡行為。這需要另開一篇詳細討論。
Reducer task已經收集了所有intermediate data，現在可以做最后計算了。在這個例子中，我們只需簡單的把數字相加就得到了最終結果，寫入result.txt
我們的這個例子并沒有導致很多的intermediate data在網絡間傳輸。然而其他的jobs可能會產生大量的intermediate data：比如，TB級數據的排序，輸出的結果是原始數據集的重新排序，結果尺寸和原始文件大小一致。Map Reduce過程會產生多大的網絡流量王權依賴于給定的Job類型。
如果你對網絡管理很感興趣，那么你將了解更多Map Reduce和你運行集群的Jobs類型，以及這些Jobs類型如何影響到網絡。如果你是一個Hadoop網絡的狂熱愛好者，那么你可能會建議寫更好的 Map Reduce jobs代碼來優化網絡性能，更快的完成Job
Hadoop通過在現有數據的基礎上提供某種商業價值，從而在你的組織內獲得成功。當人們意識到它的價值，那么你可能獲得更多的資金購買更多的機架和服務器，來擴展現有的Hadoop集群。
當增加一個裝滿服務器的新機架到Hadoop集群中時，你可能會面臨集群不平衡的局面。在上圖的例子中，Rack1和Rack2是已經存在的機器，保存著文件File.txt并且正在運行Map Reduce jogs。當我們增加兩個新的機架到集群中時，File.txt 數據并不會神奇的自動散布到新的機架中。
新的Data node服務器由于沒有數據只能空閑著，直到Client開始保存新的數據到集群中。此外當Rack1和Rack2上的服務器都滿負荷的工作，那么Job Tracker可能沒有別的選擇，只能把作用在File.txt上的Map task分配到這些沒有數據的新服務器上，新服務器需要通過網絡跨機架獲取數據。這就導致更多的網絡流量，更慢的處理速度。
為了處理這種情況，Hadoop包含一個時髦的工具叫做balancer
Balancer查看節點可用存儲的差異性，在達到特定的閥值后嘗試執行balance。有很多空閑空間的新節點將被檢測到，然后balancer 開始從空閑空間很少的Data node拷貝數據到這個新節點。Balancer通過控制臺的命令行啟動，通過控制臺取消或者關閉balancer
Balancer可用的網絡流量是非常低的，缺省設置為1MB/s。可以通過hdfs-site.xml的df.balance.bandwidthPerSec參數來修改。
Balancer是你的集群的好管家。在你增加服務器時一定會用到它，定期（每周）運行一次也是一個好主意。Balancer使用缺省帶寬可能會導致很長時間才能完成工作，比如幾天或者幾周。
本文是基于Training from Cloudera 的學習 以及對我的Hadoop實驗環境的觀測。這里討論的內容都是基于latest stable release of Cloudera's CDH3 distribution of Hadoop 。本文并沒有討論Hadoop的新技術，比如：Hadoop on Demand(HOD)和HDFS Federation ，但是這些的確值得花時間去研究。
英文原文：http://bradhedlund.com/2011/09/10/understanding-hadoop-clusters-and-the-network/
