騰訊大規模Hadoop集群實踐

轉自：http://www.csdn.net/article/2014-02-19/2818473-Tencent-Hadoop

ID lishilong404740787

TDW是騰訊最大的離線數據處理平臺。本文主要從需求、挑戰、方案和未來計劃等方面，介紹了TDW在建設單個大規模集群中采取的JobTracker分散化和NameNode高可用兩個優化方案。

TDW（Tencent distributed Data Warehouse，騰訊分布式數據倉庫）基於開源軟件Hadoop和Hive進行構建，打破了傳統數據倉庫不能線性擴展、可控性差的局限，並且根據騰訊數據量大、計算復雜等特定情況進行了大量優化和改造。

TDW服務覆蓋了騰訊絕大部分業務產品，單集群規模達到4400臺，CPU總核數達到10萬左右，存儲容量達到100PB；每日作業數100多萬，每日計算量4PB，作業並發數2000左右；實際存儲數據量80PB，文件數和塊數達到6億多；存儲利用率83%左右，CPU利用率85%左右。經過四年多的持續投入和建設，TDW已經成為騰訊最大的離線數據處理平臺。

TDW的功能模塊主要包括：Hive、MapReduce、HDFS、TDBank、Lhotse等，如圖1所示。TDW Core主要包括存儲引擎HDFS、計算引擎MapReduce、查詢引擎Hive，分別提供底層的存儲、計算、查詢服務，並且根據公司業務產品的應用情況進行了很多深度訂制。TDBank負責數據采集，旨在統一數據接入入口，提供多樣的數據接入方式。Lhotse任務調度系統是整個數據倉庫的總管，提供一站式任務調度與管理。

圖1 TDW的功能模塊

建設單個大規模集群的原因

隨著業務的快速增長，TDW的節點數也在增加，對單個大規模Hadoop集群的需求也越來越強烈。TDW需要做單個大規模集群，主要是從數據共享、計算資源共享、減輕運營負擔和成本等三個方面考慮。

1. 數據共享。TDW之前在多個IDC部署數十個集群，主要是根據業務分別部署，這樣當一個業務需要其他業務的數據，或者需要公共數據時，就需要跨集群或者跨IDC訪問數據，這樣會占用IDC之間的網絡帶寬。為了減少跨IDC的數據傳輸，有時會將公共數據冗余分布到多個IDC的集群，這樣又會帶來存儲空間浪費。

2. 計算資源共享。當一個集群的計算資源由於某些原因變得緊張時，例如需要數據補錄時，這個集群的計算資源就捉襟見肘，而同時，另一個集群的計算資源可能空閑，但這兩者之間沒有做到互通有無。

3. 減輕運營負擔和成本。十幾個集群同時需要穩定運營，而且當一個集群的問題解決時，也需要解決其他集群已經出現的或者潛在的問題。一個Hadoop版本要在十幾個集群逐一變更，監控系統也要在十幾個集群上部署。這些都給運營帶來了很大負擔。此外，分散的多個小集群，資源利用率不高，機器成本較大。

建設單個大規模集群的方案及優化

面臨的挑戰

TDW從單集群400臺規模建設成單集群4000臺規模，面臨的最大挑戰是Hadoop架構的單點問題：計算引擎單點JobTracker負載重，使得調度效率低、集群擴展性不好；存儲引擎單點NameNode沒有容災，使得重啟耗時長、不支持灰度變更、具有丟失數據的風險。TDW單點瓶頸導致平臺的高可用性、高效性、高擴展性三方面都有所欠缺，將無法支撐4000臺規模。為了解決單點瓶頸，TDW主要進行了JobTracker分散化和NameNode高可用兩方面的實施。

JobTracker分散化

1.單點JobTracker的瓶頸

TDW以前的計算引擎是傳統的兩層架構，單點JobTracker負責整個集群的資源管理、任務調度和任務管理，TaskTracker負責任務執行。JobTracker的三個功能模塊耦合在一起，而且全部由一個Master節點負責執行，當集群並發任務數較少時，這種架構可以正常運行，但當集群並發任務數達到2000、節點數達到4000時，任務調度就會出現瓶頸，節點心跳處理遲緩，集群擴展也會遇到瓶頸。

2.JobTracker分散化方案

TDW借鑒YARN和Facebook版corona設計方案，進行了計算引擎的三層架構優化（如圖2所示）：將資源管理、任務調度和任務管理三個功能模塊解耦；JobTracker只負責任務管理功能，而且一個JobTracker只管理一個Job；將比較輕量的資源管理功能模塊剝離出來交給新的稱為ClusterManager的Master負責執行；任務調度也剝離出來，交給具有資源信息的ClusterManager負責執行；對性能要求較高的任務調度模塊采用更加精細的調度方式。

圖2 JobTracker分散化架構

新架構下三個角色分別是：ClusterManager負責整個集群的資源管理和任務調度，JobTracker負責單個Job的管理，TaskTracker負責任務的執行。

（1）兩路心跳。之前的架構下，TaskTracker向JobTracker上報心跳，JobTracker串行地處理這些心跳，心跳處理中進行節點管理、任務管理、任務調度等，心跳繁重，影響任務調度和集群擴展性。新架構下，心跳被拆分成兩路心跳，分別上報任務和資源信息。

JobTracker獲知任務信息通過任務上報心跳的方式。任務上報心跳是通過任務所在的TaskTracker啟動一個新的獨立線程向對應的JobTracker上報心跳這條途徑，在同一個TaskTracker上，不同Job的任務使用不同的線程向不同的JobTracker上報心跳，途徑分散，提升了心跳上報效率。

TaskTracker通過上報心跳的方式將資源信息匯報給ClusterManager。ClusterManager從TaskTracker的心跳中獲取節點的資源信息：CPU數量、內存空間大小、磁盤空間大小等的總值和剩余值，根據這些信息判斷節點是否還能執行更多的任務。同時，ClusterManager通過TaskTracker與其之間維系的心跳來管理節點的生死存亡。

以前繁重的一路心跳被拆分成了兩路輕量的心跳，心跳間隔由40s優化成1s，集群的可擴展性得到了提升。

（2）資源概念。之前架構只有slot概念，一般根據核數來設置slot數量，對內存、磁盤空間等沒有控制。新架構弱化了slot概念，加強了資源的概念。

每個資源請求包括具體的物理資源需求描述，包括內存、磁盤和CPU等。向ClusterManager進行資源申請的有三種來源類型：Map、Reduce、JobTracker，每種來源需要的具體資源量不同。在CPU資源上，調度器仍然保留slot概念，並且針對三種來源保證各自固定的資源帽。

例如，對於24核的節點，配置13個核給Map用、6個核給Reduce用、1個核給JobTracker用，則認為該節點上有1個JobTracker slot、13個Map slot、6個Reduce slot。某個Map請求的資源需要2個核，則認為需要兩個Map slot，當一個節點的Map slot用完之後，即使有剩余的CPU，也不會繼續分配Map予其執行了。內存空間、磁盤空間等資源沒有slot概念，剩余空間大小滿足需求即認為可以分配。在查找滿足資源請求的節點時，會比較節點的這些剩余資源是否滿足請求，而且還會優先選擇負載低於集群平均值的節點。

（3）獨立並發式的下推調度。之前架構下，調度器采用的是基於心跳模型的拉取調度：任務調度依賴於心跳，Map、Reduce的調度耦合在一起，而且對請求優先級采取全排序方式，時間復雜度為nlog(n)，任務調度效率低下。

新架構采用獨立並發式的下推調度。Map、Reduce、JobTracker三種資源請求使用三個線程進行獨立調度，對請求優先級采取堆排序的方式，時間復雜度為log(n)。當有資源滿足請求時，ClusterManager直接將資源下推到請求者，而不再被動地等待TaskTracker通過心跳的方式獲取分配的資源。

例如，一個Job有10個Map，每個Map需要1個核、2GB內存空間、10GB磁盤空間，如果有足夠的資源，Map調度線程查找到了滿足這10個Map的節點列表，ClusterManager會把節點列表下推到JobTracker；如果Map調度線程第一次只查找到了滿足5個Map的節點列表，ClusterManager會把這個列表下推到JobTracker，隨後Map調度線程查找到了剩下5個Map的節點列表，ClusterManager再把這個列表下推到JobTracker。

以前基於心跳模型的拉取調度被優化成獨立並發式的下推調度之後，平均調度處理時間由80ms優化至1ms，集群的調度效率得到了提升。

3. Job提交過程

新架構下，一次Job提交過程，需要Client和ClusterManager、TaskTracker均進行交互（如圖3所示）：JobClient先向ClusterManager申請啟動JobTracker所需要的資源；申請到之後，JobClient在指定的TaskTracker上啟動JobTracker進程，將Job提交給JobTracker；JobTracker再向ClusterManager申請Map和Reduce資源；申請到之後，JobTracker將任務啟動命令提交給指定的TaskTracker。

圖3 Job提交過程

4. 存在的問題及應對措施

JobTracker分散化方案給計算引擎帶來高效性和高擴展性，但沒有帶來高可用性，單一故障點的問題在此方案中仍然存在，此時的單一故障點問題有別於以前，如下所述。

（1）ClusterManager如果發生故障，不會造成Job狀態丟失而且在短時間內即可恢復。它只存儲資源情況，不存儲狀態，ClusterManager在很短的時間內可以重啟完成。重啟之後，TaskTracker重新向ClusterManager匯報資源，ClusterManager從重啟至完全獲得集群的資源情況整個階段可以在10秒內完成。

（2）JobTracker如果發生故障，只會影響單個Job，對其他Job不會造成影響。

基於以上兩點，認為新方案的單一故障點問題影響不大，而且考慮方案實施的復雜度和時效性，TDW在JobTracker分散化方案中沒有設計高可用方案，而是通過外圍系統來降低影響：監控系統保證ClusterManager故障及時發現和恢復；Lhotse調度系統從用戶任務級別保證Job重試。

NameNode高可用

1. 單點NameNode的問題

TDW以前的存儲引擎是單點NameNode，在一個業務對應一個集群的情況下，NameNode壓力較小，出故障的幾率也較小，而且NameNode單點故障帶來的影響不會波及全部業務。但當把各個小集群統一到大集群，各個業務都存儲之上時，NameNode壓力變大，出故障的幾率也變大，NameNode單點故障造成的影響將會非常嚴重。即使是計劃內變更，停止NameNode服務耗時將近2個小時，計劃內的停止服務變更也給用戶帶來了較大的影響。

2. NameNode高可用方案

TDW設計了一種一主兩熱備的NameNode高可用方案。新架構下NameNode角色有三個：一主（ActiveNameNode）兩熱備（BackupNameNode）。ActiveNameNode保存namespace和block信息，對DataNode下發命令，並且對客戶端提供服務。BackupNameNode包括standby和newbie兩種狀態：standby提供對ActiveNameNode元數據的熱備，在ActiveNameNode失效後接替其對外提供服務，newbie狀態是正處於學習階段，學習完畢之後成為standby。

（1）Replicaton協議。為了實現BackupNameNode對ActiveNameNode的元數據一致，隨時準備接管ActiveNameNode角色，元數據操作日誌需要在主備間同步。客戶端對元數據的修改不止在ActiveNameNode記錄事務日誌，事務日誌還需要從ActiveNameNode同步到BackupNameNode，客戶端的每一次寫操作，只有成功寫入ActiveNameNode以及至少一個BackupNameNode（或者ZooKeeper）時，才返回客戶端操作成功。當沒有BackupNameNode可寫入時，把事務日誌同步到ZooKeeper來保證至少有一份事務日誌備份。

客戶端寫操作記錄事務日誌遵循以下幾個原則：

①寫入ActiveNameNode，如果寫入失敗，返回操作失敗，ActiveNameNode自動退出；

②當寫入至少兩個節點（Active-NameNode和Standby/ZooKeeper/LOG_SYNC newbie）時返回操作成功，其他返回失敗；LOG_SYNC newbie表示newbie已經從ActiveNameNode獲取到全量日誌後的狀態；

③當只成功寫入ActiveNameNode，此後的Standby和ZooKeeper均寫入失敗時，返回失敗；

④當只存在ActiveNameNode時，進入只讀狀態。

（2）Learning協議。newbie學習機制確保newbie啟動後通過向ActiveNameNode學習獲取最新的元數據信息，學習到與ActiveNameNode同步時變成standby狀態。newbie從ActiveNameNode獲取最新的fsimage和edits文件列表，ActiveNameNode還會和newbie之間建立事務日誌傳輸通道，將後續操作日誌同步給newbie，newbie將這些信息載入內存，構建最新的元數據狀態。

（3）事務日誌序號。為了驗證事務日誌是否丟失或者重復，為事務日誌指定遞增連續的記錄號txid。在事務日誌文件edits中加入txid，保證txid的連續性，日誌傳輸和加載時保證txid連續遞增，保存內存中的元數據信息到fsimage文件時，將當前txid寫入fsimage頭部，載入fsimage文件到內存中時，設置元數據當前txid為fsimage頭部的txid。安全日誌序號（safe txid）保存在ZooKeeper上，ActiveNameNode周期性地將txid寫入ZooKeeper作為safe txid，在BackupNameNode轉換為ActiveNameNode時，需要檢查BackupNameNode當前的txid是否小於safe txid，若小於則禁止這次角色轉換。

（4）checkpoint協議。新架構仍然具有checkpoint功能，以減少日誌的大小，縮短重啟時構建元數據狀態的耗時。由ActiveNameNode維護一個checkpoint線程，周期性地通知所有standby做checkpoint，指定其中的一個將產生的fsimage文件上傳給ActiveNameNode。

（5）DataNode雙報。Block副本所在的節點列表是NameNode元數據信息的一部分，為了保證這部分信息在主備間一致性，DataNode采用雙報機制。DataNode對塊的改動會同時廣播到主備，對主備下發的命令，DataNode區別對待，只執行主機下發的命令而忽略掉備機下發的命令。

（6）引入ZooKeeper。主要用來做主節點選舉和記錄相關日誌：NameNode節點狀態、安全日誌序號、必要時記錄edit log。

3. 主備切換過程

當主退出時主備狀態切換的過程（如圖4所示）：當ActiveNameNode節點IP1由於某些原因退出時，兩個備節點IP2和IP3通過向ZooKeeper搶鎖競爭主節點角色；IP2搶到鎖成為ActiveNameNode，客戶端從ZooKeeper上重新獲取主節點信息，和IP2進行交互，這時即使IP1服務恢復，也是newbie狀態；事務日誌在主備間同步，newbie IP1通過向主節點IP2學習成為standby狀態。

圖4 主退出時主備狀態切換

4. 存在的問題

NameNode高可用方案給存儲引擎帶來了高可用性，但在高效性方面做出了一些犧牲，由於事務日誌需要同步，寫性能有20%左右的下降。

其他優化

TDW在實施大集群過程中，除了主要實施JobTracker分散化和NameNode高可用兩個方案，還進行了一些其他優化。

1. NameNode分散化

隨著存儲量和業務的不斷增長，一個HDFS元數據空間的訪問壓力與日俱增。通過NameNode分散化來減少一個元數據空間的訪問壓力。NameNode分散化主要對元數據信息進行分拆，對用戶透明，用戶訪問認為處於同一個存儲引擎，底層可以拆分成多個集群。TDW在Hive層增加用戶到HDFS集群的路由表，用戶表的數據將寫入對應的HDFS集群，對外透明，用戶只需使用標準的建表語句即可。TDW根據公司業務的實際應用場景，根據業務線和共享數據等把數據分散到兩個HDFS集群，有利於數據共享同時也盡量規避集群間的數據拷貝。采用簡單、改動最少的方案解決了實際的問題。

2. HDFS兼容

TDW內部有三個HDFS版本：0.20.1、CDH3u3、2.0，線上主流版本是CDH3u3，主流HDFS版本使用的RPC框架尚未優化成Thrift或者Protocol Buffers等，三個版本互不兼容，增加了互相訪問的困難。通過RPC層兼容方式實現了CDH3u3和0.20.1之間的互通，通過完全實現兩套接口方式實現了CDH3u3和2.0之間的互通。

3. 防止數據誤刪除

重要數據的誤刪除會給TDW帶來不可估量的影響，TDW為了進一步增加數據存儲可靠性，不僅開啟NameNode回收站特性，還增加兩個特性： 刪除黑白名單，刪除接口修改成重命名接口，白名單中的目錄可以被刪除，白名單中的IP可以進行刪除操作，其他則不可；DataNode回收站，塊刪除操作不會立即進行磁盤文件的刪除，而是維護在待刪除隊列裏，過期之後才進行實際的刪除操作，這樣可以保證在一定時間內如果發現重要的數據被誤刪除時可以進行數據恢復，還可以防止NameNode啟動之後元數據意外缺失而造成數據直接被刪除的風險。

結語

TDW從實際情況出發，采取了一系列的優化措施，成功實施了單個大規模集群的建設。為了滿足用戶日益增長的計算需求，TDW正在進行更大規模集群的建設，並向實時化、集約化方向發展。TDW準備引入YARN作為統一的資源管理平臺，在此基礎上構建離線計算模型和Storm、Spark、Impala等各種實時計算模型，為用戶提供更加豐富的服務。

本文出自 “李世龍” 博客，謝絕轉載！

騰訊大規模Hadoop集群實踐