[ 深入雲計算 ] Hadoop 的體系結構

Preface: 
Hadoop 是 Apache 下的一個項目, 它並不是一個用於儲存的分布式文件系統而已, 而是設計用來在通用計算設備組成的大型叢集上執行分布式應用的框架. 由 HDFS, MapReduce, HBase, Hive 和 ZooKeeper 等成員的組成, 其中 HDFS 和 MapReduce 是 Hadoop 中兩個最基礎, 最重要的成員, 它們提供了互補性服務並在核心層上提供更高層的服務. Hadoop 項目結構如下: 
 

- Core/Common 

從 Hadoop 0.2 版開始, Hadoop Core 項目更名為 Common. 它是 Hadoop 其他子項目提供支持的常用工具, 主要包括 FileSystem, RPC 和串行操作 library. 它們為在廉價的硬件上搭建計算環境並提供基本的服務, 並且為運行在該平台上的軟件開發提供了所需要的 API.

- Avro 

Avro 是用於數據序列化的系統. 它提供了豐富的數據結構類型, 快速可壓縮的二進制數據格式, 存儲持久性數據的文件集, 遠程調用 RPC 的功能和簡單的動態語言集成功能. 其中代碼生成器既不需要讀/寫文件數據, 也不需要使用或實現 RPC 協議, 它只是一個可選的對靜態語言的實現.

HDFS 的體系結構: 
Hadoop 分布式文件系統 HDFS 可以部署在廉價硬體之上, 能夠高容錯, 可靠性的提供存儲海量數據 (TB 甚至是 PB 級). 它可以和 MapReduce 編成模型很好的結合, 以為能夠為應用程序提供高吞吐量的數據訪問, 並適用於大數據應用程序. 

HDFS 的設計目標 
1. 檢測及快速恢復硬體故障: 整個 HDFS 系統由成千上百存儲文件的服務器組成, 如此多的服務器意味故障會常發生, 因此故障的檢測與快速恢復是 HDFS 的第一個核心目標 
2. Stream 的數據訪問: HDFS 使應用程序能 Stream 的訪問它們的數據集. HDFS 被設計成適合進行批量處理而不是用戶交互式的處理, 重視數據吞吐量而不是數據訪問的反應數度. 
3. 簡化一致性模型: 大部分的 HDFS 程序操作文件時需要一次性寫入, 多次讀取. 一個文件一旦經過創建, 寫入與關閉後就不太需要修改, 從而簡化一致性問題並提高數據吞吐量. 
4. 移動計算的代價比移動數據的代價低: 一個應用請求的計算, 離它操作的數據越近效能就越好. 這在數據達到海量級別時更是如此. 將計算移動到數據比數據移動到計算處更高效. 
5. 超大規模數據集: HDFS 的一般企業文件大小可能在 TB 或是 PB 級. 
6. 異構軟硬體平台間的可移植姓: 這個特性便於 HDFS 作為大規模數據應用平台的推廣. 

HDFS 結構模型 
HDFS 是一個主從 (Master/Slave) 結構模型. 從最終用戶的角度來看, 它就像是傳統的文件系統一樣, 可以透過目錄路徑對文件執行 CRUD (Create/Read/Update/Delete) 操作. 一個 HDFS 叢集是由一個 NameNode 和若干的 DataNode 組成. NameNode 主節點是主服務器, 管理文件系統的命名空間和客戶端對文件的方問操作; DataNode 是叢集中的一般節點, 負責節點數據的存儲. 客戶端通過 NameNode 向 DataNode 節點交互訪問文件系統, 聯系 NameNode 獲取文件, 而文件的 I/O 操作則是直接和 DataNode 進行交互作用. HDFS 允許用戶以文件形式存儲數據, 架構如下: 
 

假設客戶端要訪問一個文件, 首先會從 NameNode 中獲取組成該文件數據塊所在位置列表, 即知道數據塊存儲在那些 DataNode 上; 然後客戶端直接從 DataNode 上讀取文件數據. 在這一過程中 NameNode 不參與文件的傳輸. NameNode 與 DataNode 都可以設計成在廉價的 Linux 主機上面運行, HDFS 採用 Java 語言開發, 因此可以佈署在大範圍的機器上. 一個典型的案例是一台機器跑一個單獨的 NameNode 節點, 叢集中的其他機器則各跑一個 DataNode 實例. 

NameNode 是 HDFS 的守護程序, 它主要負責記錄大數據文件如何被分割成數據塊, 被分割後的數據塊分別是存儲到那些 DataNode 數據節點上. NameNode 的主要功能是對內存 I/O 進行集中管理. NameNode 節點是單一的, 這樣就可以大大簡化系統的結構. NameNode 也負責管理與保管所有 HDFS 元數據, 因而用戶文件數據的 讀/寫 就可以直接在 DataNode 上而不需要透過 NameNode. 在一般狀況下, 如果是 DataNode 所在的服務器出現故障, Hadoop 叢集依舊可以正常運轉, 或者快速重啟. 但是如果是 NameNode 出現故障, 整個 Hadoop 的服務將因此而中止! 

文件系統的命名空間 NameSpace 
HDFS 支持傳統的 階層架構文件組織, 用戶可以創建目錄, 並在目錄中創建, 刪除, 移動和重新命名文件. 但是 HDFS 不支持用戶磁碟配額與訪問權限的控制, 也不支援軟/硬鏈接. NameNode 負責維護文件系統的名稱空間, 任何對文件系統名稱空間或屬性的修改都將被 NameNode 記錄下來, 文件副本的數目稱為文件副本係數, 這個設定也是記錄在 NameNode. 

數據複製與存放 
HDFS 是在一個大叢集中跨機器可靠的存儲超大文件. 它將每個文件存儲成一系列的數據塊, 數據按 64 MB 分成大小等同的一個個數據塊, 除了最後一個, 所有的數據塊大小都是相同的. 
1. 數據的複製 

文件的所有數據塊都會有副本, 這樣可以提高數據的容錯性, 應用程序可以指定某個文件的副本數目. 數據塊的副本係數可以在文件創建時候指定也可以在日後進行修改. HDFS 中的文件都是一次性寫入的, 並在任何時候只能有一個寫入者.

NameNode 管理者數據的複製, 採用週期性地從叢集中的每個 DataNode 接收 Heart bean 訊號和塊狀報告 (Blockreport). 如果收到的心跳報告就表明該 DataNode 節點是正常運作; 如果沒有收到 Heartbeat 訊號, 則說明該 DataNode 節點發生異常. 數據塊狀態報告中包含了一個 DataNode 節點上所有數據塊的列表訊息:
 

2. 副本的存放是 HDFS 可靠性和性能的關鍵 

優化的副本存放策略是 HDFS 區分其他大部分分布式文件系統的重要特性. HDFS 採用一種 Rack-aware 的策略來改進 HDFS 數據的可靠性, 可用性與網絡頻寬的利用率. 一個 Rack-aware 的過程就是 NameNode 可以確定每個 DataNode 所屬的 Rack-ID. 一個簡單但沒有優化的策略就是將副本存放在不同的 Rack 上, 這樣可以有效防止整個 Rack 失效時數據的丟失, 並且允許讀數據的時候充分利用多個 Rack 的頻寬. 這種策略就是將副本均勻的分佈在叢集中, 有利於負載的平衡. 但是這個策略也有缺點, 即一個寫操作需要把數據傳送到多個 Rack 而增加了寫的代價.

在大多數情況下, 數據塊的副本係數是 3, HDFS 的存放策略是將一個副本存放在本機節點上, 一個副本放在同一個 Rack-ID 的另一個節點上, 最後一個副本放在不同 Rack-ID 的節點上. 這種策略減少了 Rack 間的數據傳輸, 提高了寫操作的效率. Rack 的錯誤遠遠比節點的錯誤少, 所以這個策略不會影響到數據的可靠性和可用性. 與此同時, 因為數據只放在兩個 (少於 3 個) 不同的 Rack 上, 所以此策略減少了讀取數據時所需要的網絡傳輸量.

3. 副本選擇 

為了降低整體的頻寬消耗與讀取延時, HDFS 會盡量讓讀取程序讀取離它最近的副本. 如果在讀取程序的同一個 Rack 上有一個副本, 那就讀取該副本. 如果一個 HDFS 叢集跨越多個數據中心, 那麼客戶端也將首先讀取本地數據中心的副本.

4. 安全模式 

NameNode 啟動後會進入安全模式的特殊狀態, 在處於安全模式狀態中的 NameNode 是不會複製數據塊. NameNode 接收所有的 DataNode 的 Heartbean 訊號和狀態報告, 塊的狀態報告中包含該 DataNode 中所有的數據塊列表訊息, 每個數據塊都有一個最小的副本數. 當 NameNode 檢測了某個數據塊的副本並達到這個數據塊最少副本數值, 就說明該數據塊是安全的; 在某百分比 (百分比透過參數設定配置) 的數據塊通過 NameNode 檢測確定是安全後, 這時候會加一個額外的 30 秒等待時間, NameNode 就會退出安全模式. NameNode 接下來確定那些數據塊的副本數沒有達到指定的數目, 並把這些數據塊的副本複製到其他的 DataNode 節點上.

文件系統元數據的持久化 
NameNode 節點上保存著 HDFS 的命名空間, 在對任何文件系統的元數據進行創建, 修改, 刪除等操作時, NameNode 會採用 Editlog 的事務日誌文件記錄下來. 整個 HDFS 的文件系統的命名空間, 包含數據塊的映射, 文件屬性, 數據塊的副本訊息都會存儲在這個 FsImage 文件中. 而透過這個 Editlog, 系統在重開機後就能回復到關機前的狀態. 

HDFS 的通信協議 
HDFS 通訊協議是建立在 TCP/IP 網絡協議之上. 客戶端透過 TCP 端口連接到 NameNode, 然後使用 ClientProtocol 協議與 NameNode 節點進行資訊交換. 在 DataNode 上採用 DataNodeProtocol 協議來與 NameNode 主節點進行資訊交換. 遠程調用 RPC 模型被抽象出來並封裝成 ClientProtocol 和 DataProtocol 協議. 一般情況下 NameNode 節點不會主動發起 RPC 請求 而是 接收來自客戶端或是 DataNode 的 RPC 請求. 

HDFS 的 Robust 
HDFS 的可靠性主要表現在即使出現錯誤情況下, 也要保證能夠進行數據存儲. 常見的 3 種錯誤狀況是 NameNode 故障, DataNode 故障 與 網路割裂 (Network Partitions). 
1. 磁碟數據錯誤 

每個 DataNode 節點都會週期性的向 NameNode 發送 Heartbeat. 網路故障可能導致一部分的 DataNode 和 NameNode 失去聯繫. NameNode 透過接收 Heartbean 來檢測這樣的情況, 把近期那些不在發送 Heartbeat 的 DataNode 標記起來, 並不在給它們分配新的 I/O 請求. 存儲在那些 DataNode 節點上的所有數據將不在有效. DataNode 出現這種狀況會導致數據塊副本系數低於設定值, 這時 NameNode 不斷的檢測這些需要複製的數據塊並啟動增加副本. 除了網絡問題造成 DataNode 失效, 如 副本遭到破壞, DataNode 上的硬碟錯誤, 或者文件的副本係數增大都會造成 NameNode 啟動複製數據塊副本的動作!

2. 叢集均衡 

HDFS 分布式文件系統支持數據均衡策略. 如果某個 DataNode 節點上的空閒空間低於某個特定臨界點, 系統就會自動地將數據從這個 DataNode 移動到其他較空閒的 DataNode, 這就是 HDFS 的叢集均衡策略. 當某個文件的請求突然增加, 這時就需要啟動一個計劃創建該文件新的副本, 並且同時重新對叢集進行負載平衡. 但是目前這個均衡策略還沒有實作, 等待日後版本支援.

3. 數據完整性 

如果從某個 DataNode 獲取的數據塊是毀損的, 原因可能是由 DataNode 自身的存儲設備錯誤或是網絡數據傳輸錯誤造成. 針對這種狀況, HDFS 客戶端可以對 HDFS 文件內容進行 Checksum 檢查. 在客戶端創建一個新的 HDFS 文件時, 計算這個文件每個數據塊的校驗結果作為一個獨立隱藏文件並與原始文件保存在同一個 HDFS 名字空間下. 一旦客戶端獲取文件內容, 它就開始檢驗從 DataNode 獲取的數據塊與當初建立的校驗結果是否匹配. 如果不匹配, 客戶端可以選擇從其他 DataNode 獲取該數據塊的副本.

4. 元數據磁碟錯誤 

FsImage 和 Ediglog 是 HDFS 的核心數據結構. FsImge 保存數據塊的文件映射, 文件屬性, 數據塊的副本等訊息. Ediglog 則是事務日誌文件, 保存的是元數據進行創建, 修改, 刪除等操作. 如果這麼重要的文件損壞, 就會導致 HDFS 實例都將失效. 解決這個問題的方法是: NameNode 配置成支持多個 FsImage 和 Editlog 的副本, 對 FsImage 或 Editlog 的修改都會同步到它們的副本上. 這麼多的副本的同步將會降低 NameNode 處理命名空間事務的效率, 這個代價相對於損壞了 FsImage 和 Editlog 帶來的破壞性影響是可以接受的. HDFS 的應用是非元數據密集的, 當 NameNode 重新啟動時, 它會選取最近的, 完整的 FsImage 和 Editlog 使用.

NameNode 是 HDFS 叢集中的 Single Point of Failure 所在. 如果 NameNode 機器故障是需要手工恢復的. 目前自動重啟或在另一台機器上做 NameNode 故障轉移的功能還沒實現.

5. 快照 

快照支持某一特定時刻數據的複製備份. 利用快照, 可以讓 HDFS 在數據損壞時恢復到一個已知正確的時間點. HDFS 目前還不支持快照功能, 但計畫在未來版本支援.

HDFS 的數據組織 
1. 數據塊 

HDFS 被設計成支持大容量的文件, 處理大規模的數據運用. 這些應用都是只寫入一次, 讀取多次, 並且讀取速度應能滿足 Streaming 讀取的要求. 一個典型的數據塊大小是 64MB. 因而 HDFS 中的文件是按照 64MB 被劃分成不同的數據塊, 每個塊盡可能的儲存在不同的 DataNode.

2. 數據塊的存放 

客戶端創建文件的請求會先將文件數據緩存到本地的一個臨時文件中, 並沒有立即發送給 NameNode. 應用程序的寫操作被透明地重新定位到這個臨時文件中, 當臨時文件累計超過一個數據塊的大小時, 客戶端才會聯繫 NameNode, NameNode 將文件名插入到 HDFS 文件層次結構中, 並且分配一個數據塊給它, NameNode 返回 DataNode 的標示符和目標數據塊給用戶端, 接著客戶端將這塊數據塊從本地臨時文件上傳到指定的 DataNode 上. 如果在文件關閉時臨時文件還有剩餘, 則沒有上傳的的數據塊也會傳輸到指定的 DataNode 上, 然後客戶端會告訴 NameNode 該文件已經關閉, 這時 NameNode 才把文件創建的操作提交到日誌 Editlog 文件中存儲. 如果該文件在文件關閉前, NameNode 出現故障, 那該文件就會丟失.

在上述方法中, HDFS 在進行文件傳輸時充分考慮了應用程序中需要進行文件的 Stream 寫入, 採用客戶端的緩存, 這樣可以避免由於網路速度的堵塞造成不好的寫入效率

3. 數據塊流水線複製 

客戶端向 HDFS 文件系統寫入數據的過程為:
1. 開始時寫到本地的臨時文件中, 也就是客戶端得緩存中, 當本地臨時文件大小累積到一個數據塊 64MB 大小時, 客戶端就會從 NameNode 節點中獲取一個 DataNode 列表訊息用來存儲數據塊.
2. 接著客戶端向 DataNode 節點傳輸數據, 假設數據塊的副本為 3, 第一個 DataNode 會一小部分一小部分 (大概 4KB) 地接收數據. 而接收到的每一部分數據都會寫入到本地倉儲.
3. 在同時數據也依據 Data Node 列表向第二個 DataNode 節點傳輸副本.
4. 同時在第二個 DataNode 收到數據後, 也同時向第三個 DataNode 節點傳輸副本.

HDFS 的可訪問性 
應用程序可以通過多種方式來訪問 HDFS 

方式一: HDFS 提供 JavaAPI 接口供用戶訪問
方式二: HDFS 提供 C 語言封裝的 API 給用戶提供訪問
方式三: HDFS 還提供了瀏覽器方式提供用戶訪問.
方式四: HDFS 正在開發 WebDAV 協議供用戶訪問.

HDFS 的存儲空間回收 
1. 文件的刪除和恢復 

當用戶或應用程序要刪除某個文件, 這個文件並沒有立刻從 HDFS 文件系統刪除. 實際情況如下: HDFS 首先將這個要刪除的文件進行重命名並轉移到 /trash 目錄. 只要該文件還在 /trash 目錄中, 就表明該文件還沒有徹底刪除, 該文件還是可以恢復的. 該文件在 /trash 目錄的保存時間可以透過配置設定, 當這個時間超過, NameNode 就會從名字空間中刪除該文件, 文件刪除後與該文件相關的數據塊將被釋放. 默認策略是刪除文件在 /trash 目錄中的保存時間是 6 hrs, 超過 6 hrs 將自動刪除文件.

2. 減少副本數 

重新設置一個文件的副本係數, 副本係數減少, 這時 NameNode 會選擇多餘的副本進行刪除. 下次 Heartbeat 檢測將訊息除遞給 DataNode 節點, DataNode 根據訊息內容刪除相應的數據塊, 釋放空間使得叢集中的可用空間加大.

MapReduce 的體系結構: 
MapReduce 是 Hadoop 的主要核心組件之一. Hadoop Map/Reduce 是一個使用簡易的軟體框架, 基於它寫出來的應用程序能夠運行在由上千個節點組成的大型叢集上, 並以一種可靠容錯的方式並行處理 TB 級別的數據集. 採用 MapReduce 架構實現的程序能夠在由大量的普通配置的計算機構成的叢集中實現並行化操作. MapReduce 系統在運行過程中只關心數據如何分割, 如何調度, 以及集群中計算機如何對錯誤進行處理, 管理著計算機之間的通信. 採用 MapReduce 架構可以使那些沒有進行過併行計算和分布式計算的開發人員能充分利用分布式系統的豐富資源進行併行式, 分布式的開發. 

MapReduce 框架由一個單獨的 master JobTracker 和叢集節點上的 slave TaskTracker 共同組成. master 負責調度一個作業中的所有任務, 把這些任務分布在不同的 slave 上. master 監控 slave 節點上這些任務的執行情況, 並重新執行失敗的任務, 而 slave 僅負責執行由 master 指派的任務. 

MapReduce 是一種編成模式 
MapReduce 是一種編成模式, 一種雲計算的核心計算方式並採用分布式運算計算模式, 也是簡化的分布式編成模式. 
1 MapReduce 主要解決問題 

MapReduce 致力於解決大規模數據處理的問題. 因此 MapReduce 在設計之初就考慮了數據的局部性原理, 利用局部性原理將整個問題分而治之 (有點像 MergeSort). 數據在處理之前已經分布到各個節點上, 處理時每個節點先就近讀取本地存儲的數據來進行 Map 處理, 將 Map 處理後的數據再進行合併 (combine), 排序 (shuffle and sort) 然後再分發到 Reduce 節點. 在數據傳輸過程中, 為了避免大量的數據傳輸, 提高數據傳輸效率, 採用無共享式架構的好處就是配合複製 (replication) 策略, 為叢集帶來良好的容錯能力, 當一部分的節點出現故障對叢集的正常工作不會造成太大的影響.

2. MapReduce 編成模式的核心思想 

MapReduce 編成模式的主要思想是自動分割要執行的問題 (程序) 並拆解成 Map (映射) 和 Reduce (簡化) 的方式. 在數據被分割後通過 Map 函數的程序將數據映射成不同的區塊, 分配給計算機群處理, 達到分布式運算的效果. 在通過 Reduce 函數的程序將結果彙整從而輸出開發者的結果.

簡單來說 Map 函數是把一組數據一對一的映射到另一組數據中. 映射的規則是由一個函數來指定, 如一組數據 [1,2,3,4] 乘與 3 的映射就變成 [3,6,9,12]. Reduce 函數的作用就是對這一組數據進行 "歸約". 歸約的規則也是由一個函數指定, 如規則為對 [3,6,9,12] 進行求和得到的結果為 30. 總的來說, Map 函數是要把任務分解成多個小任務, Reduce 函數負責把分解後的各個任務處理的結果進行匯總. 對於其他複雜的問題, 如工作調度, 分布是存儲, 容錯處理, 負載均衡, 網絡通訊, 則由 MapReduce 框架來負責處理.

3. MapReduce 與分布式文件系統 HDFS 的關係 

通常 Map/Reduce 框架和分布式文件系統 HDFS 是運行在同一組相同的節點上. 換句話說計算節點與存儲節點通常再一起. 採用這種配置的優勢是在框架中的那些數據可以被高效的調度運用而使得整個叢集網絡得到高效的運用.

一個 Map/Reduce 作業 (job) 會把輸入的數據劃分成多個獨立的數據塊, 這個工作由 Map 任務 (task) 採用並行的方式處理. 框架會對 Map 的輸出結果進行排序, 通過 Map 函數處理後, 把處理的結果輸入給 Reduce 任務. 通常作業的輸入與輸出都會被存儲在文件系統中. 框架負責任務的調度與監控, 以及重新執行失敗的任務.

Map/Reduce 
MapReduce 處理大規模數據其核心就是 Map/Reduce 函數. 這兩個函數的具體功能由用戶根據自己的需求設計實現, 只要能夠按照用戶自定義的規則, 將輸入的 轉換成另一個或一批 的輸出. 在 Map 階段, MapReduce 框架將任務的輸入數據分割成固定大小的數據片段 (splits), 隨後將每個 split 進一步分解成一批鍵值對 . Hadoop 為每一個 split 創建一個 Map 任務用於執行用戶自定義的 Map 函數, 並將對應的數據塊 split 中的 作為輸出, 得到了計算的中間結果 . 接著將中間結果按 k2 進行排序, 並將 key 值相同的 value 放在一起形成一個新列表 元組. 最後根據 key 值範圍將這些元組進行分組以對應不同的 Reduce 任務. 

在 Reduce 階段, Reduce 任務從不同的 Map 接收來的數據整合在一起並進行排序, 然後調用用戶自定義的 Reduce 函數, 對輸入的 進行相應的處理, 得到鍵值對 並輸出到 HDFS 上. 既然 MapReduce 框架為每個 split 創建一個 Map, 那麼誰來確認 Reduce 任務的數目? 答案是用戶, 用戶必須確認 Reduce 數量. Mapred-site.xml 配置文件中有一個表示 Reduce 任務數目的屬性 Mapred.Reduce.tasks, 該屬性默認值為 1, 開發人員可以通過 job.setNumReduceTasks() 方法重新設置該值. 

MapReduce 將處理大數據的過程拆解成 Map 與 Reduce 的過程如下圖: 
 

Map/Reduce 範例: 
接著我們要了解 Map/Reduce 最好的方法就是來看一個簡單的範例. 考慮你有一堆 Words, 你希望知道每個 Word 出現的次數, 如果放在 Map/Reduce 的流程中, 看起來會像下面的圖: 
 

Supplement: 
* [ Python 文章收集 ] python內置函數 map/reduce/filter