Hadoop重點(diǎn)難點(diǎn)：可靠性/Failover/Shuffle

HDFS – 可靠性

HDFS 的可靠性主要有一下幾點(diǎn):

• 冗余副本策略
• 機(jī)架策略
• 心跳機(jī)制
• 安全模式
• 效驗(yàn)和
• 回收站
• 元數(shù)據(jù)保護(hù)
• 快照機(jī)制

1.冗余副本策略

• 可以在 hdfs-site.xml 中設(shè)置復(fù)制因子指定副本數(shù)量
• 所有數(shù)據(jù)塊都可副本
• DataNode 啟動(dòng)時(shí),遍歷本地文件系統(tǒng),產(chǎn)生一份 HDFS 數(shù)據(jù)塊和本地文件的對(duì)應(yīng)關(guān)系列表 (blockreport) 匯報(bào)給 Namenode

2.機(jī)架策略

• HDFS 的”機(jī)架感知”,通過節(jié)點(diǎn)之間發(fā)送一個(gè)數(shù)據(jù)包,來感應(yīng)它們是否在同一個(gè)機(jī)架
• 一般在本機(jī)架放一個(gè)副本,在其他機(jī)架再存放一個(gè)副本,這樣可以防止機(jī)架失效時(shí)丟失數(shù)據(jù),也可以提高帶寬利用率

3.心跳機(jī)制

• NameNode 周期性從 DataNode 接受心跳信息和塊報(bào)告
• NameNode 根據(jù)快報(bào)告驗(yàn)證元數(shù)據(jù)
• 沒有按時(shí)發(fā)送心跳的 DataNode 會(huì)被標(biāo)記為宕機(jī),不會(huì)再給他任何 I/O 請(qǐng)求
• 如果 DataNode 失效造成副本數(shù)量下降,并且低于預(yù)先設(shè)定的值,NameNode 會(huì)檢測(cè)出這些數(shù)據(jù)庫(kù),并在合適的時(shí)機(jī)從新復(fù)制
• 引發(fā)重新復(fù)制的原因還包括數(shù)據(jù)副本本身?yè)p壞,磁盤錯(cuò)誤,復(fù)制因子被增大等

4.安全模式

• NameNode 啟動(dòng)時(shí)會(huì)先經(jīng)過一個(gè) “安全模式” 階段
• 安全模式階段不會(huì)產(chǎn)生數(shù)據(jù)寫
• 在此階段 NameNode 收集各個(gè) DataNode 的報(bào)告, 當(dāng)數(shù)據(jù)塊達(dá)到最小副本數(shù)以上時(shí),會(huì)被認(rèn)為是”安全”的
• 在一定比例(可設(shè)置) 的數(shù)據(jù)塊被確定為”安全” 后 ,在過若干時(shí)間,安全模式結(jié)束
• 當(dāng)檢測(cè)到副本數(shù)不足的數(shù)據(jù)塊是,該塊會(huì)被復(fù)制,直到達(dá)到最小副本數(shù)

5.checksum

• 在文件創(chuàng)立時(shí),每個(gè)數(shù)據(jù)塊都產(chǎn)生checksum
• checksum會(huì)作為單獨(dú)一個(gè)隱藏文件保存在命名空間下
• 客戶端獲取數(shù)據(jù)時(shí)可以檢查checksum是否相同,從而發(fā)現(xiàn)數(shù)據(jù)塊是否損壞
• 如果正在讀取的數(shù)據(jù)塊損壞,則可以繼續(xù)讀取其他副本

6.回收站

• 刪除文件時(shí),其實(shí)是放入回收站 /trash
• 回收站里的文件是可以快速恢復(fù)的
• 可以設(shè)置一個(gè)時(shí)間值,當(dāng)回收站里文件的存放時(shí)間超過了這個(gè)值,就被徹底刪除,并且釋放占用的數(shù)據(jù)塊

7.元數(shù)據(jù)保護(hù)

• 映像文件和事物日志是 NameNode 的核心數(shù)據(jù).可以配置為擁有多個(gè)副本
• 副本會(huì)降低 NameNode 的處理速度,但增加安全性
• NameNode 依然是單點(diǎn),如果發(fā)生故障要手工切換

YARN – Failover

失敗類型

• 程序問題
• 進(jìn)程崩潰
• 硬件問題

失敗處理

• 任務(wù)失敗

1. 運(yùn)行時(shí)異?；蛘逬VM退出都會(huì)報(bào)告給ApplicationMaster
2. 通過心跳來檢查掛住的任務(wù)(timeout)，會(huì)檢查多次（可配置）才判斷該任務(wù)是否失效
3. 一個(gè)作業(yè)的任務(wù)失敗率超過配置，則認(rèn)為該作業(yè)失敗
4. 失敗的任務(wù)或作業(yè)都會(huì)有ApplicationMaster重新運(yùn)行

• ApplicationMaster失敗

1. ApplicationMaster定時(shí)發(fā)送心跳信號(hào)到ResourceManager，通常一旦ApplicationMaster失敗，則認(rèn)為失敗，但也可以通過配置多次后才失敗
2. 一旦ApplicationMaster失敗，ResourceManager會(huì)啟動(dòng)一個(gè)新的ApplicationMaster
3. 新的ApplicationMaster負(fù)責(zé)恢復(fù)之前錯(cuò)誤的ApplicationMaster的狀態(tài)(yarn.app.mapreduce.am.job.recovery.enable=true)，這一步是通過將應(yīng)用運(yùn)行狀態(tài)保存到共享的存儲(chǔ)上來實(shí)現(xiàn)的，ResourceManager不會(huì)負(fù)責(zé)任務(wù)狀態(tài)的保存和恢復(fù)
4. Client也會(huì)定時(shí)向ApplicationMaster查詢進(jìn)度和狀態(tài)，一旦發(fā)現(xiàn)其失敗，則向ResouceManager詢問新的ApplicationMaster

• NodeManager失敗

1. NodeManager定時(shí)發(fā)送心跳到ResourceManager，如果超過一段時(shí)間沒有收到心跳消息，ResourceManager就會(huì)將其移除
2. 任何運(yùn)行在該NodeManager上的任務(wù)和ApplicationMaster都會(huì)在其他NodeManager上進(jìn)行恢復(fù)
3. 如果某個(gè)NodeManager失敗的次數(shù)太多，ApplicationMaster會(huì)將其加入黑名單（ResourceManager沒有），任務(wù)調(diào)度時(shí)不在其上運(yùn)行任務(wù)

• ResourceManager失敗

1. 通過checkpoint機(jī)制，定時(shí)將其狀態(tài)保存到磁盤，然后失敗的時(shí)候，重新運(yùn)行
2. 通過zookeeper同步狀態(tài)和實(shí)現(xiàn)透明的HA
3. 可以看出，一般的錯(cuò)誤處理都是由當(dāng)前模塊的父模塊進(jìn)行監(jiān)控（心跳）和恢復(fù)。而最頂端的模塊則通過定時(shí)保存、同步狀態(tài)和zookeeper來?實(shí)現(xiàn)HA

Hadoop Shuffle

MapReduce – Shuffle

對(duì)Map的結(jié)果進(jìn)行排序并傳輸?shù)絉educe進(jìn)行處理 Map的結(jié)果并不是直接存放到硬盤,而是利用緩存做一些預(yù)排序處理 Map會(huì)調(diào)用Combiner，壓縮，按key進(jìn)行分區(qū)、排序等，盡量減少結(jié)果的大小 每個(gè)Map完成后都會(huì)通知Task，然后Reduce就可以進(jìn)行處理。

Map端

• 當(dāng)Map程序開始產(chǎn)生結(jié)果的時(shí)候，并不是直接寫到文件的，而是利用緩存做一些排序方面的預(yù)處理操作
• 每個(gè)Map任務(wù)都有一個(gè)循環(huán)內(nèi)存緩沖區(qū)（默認(rèn)100MB），當(dāng)緩存的內(nèi)容達(dá)到80%時(shí)，后臺(tái)線程開始將內(nèi)容寫到文件，此時(shí)Map任務(wù)可以繼續(xù)輸出結(jié)果，但如果緩沖區(qū)滿了，Map任務(wù)則需要等待
• 寫文件使用round-robin方式。在寫入文件之前，先將數(shù)據(jù)按照Reduce進(jìn)行分區(qū)。對(duì)于每一個(gè)分區(qū)，都會(huì)在內(nèi)存中根據(jù)key進(jìn)行排序，如果配置了Combiner，則排序后執(zhí)行Combiner（Combine之后可以減少寫入文件和傳輸?shù)臄?shù)據(jù)）
• 每次結(jié)果達(dá)到緩沖區(qū)的閥值時(shí)，都會(huì)創(chuàng)建一個(gè)文件，在Map結(jié)束時(shí)，可能會(huì)產(chǎn)生大量的文件。在Map完成前，會(huì)將這些文件進(jìn)行合并和排序。如果文件的數(shù)量超過3個(gè)，則合并后會(huì)再次運(yùn)行Combiner（1、2個(gè)文件就沒有必要了）
• 如果配置了壓縮，則最終寫入的文件會(huì)先進(jìn)行壓縮，這樣可以減少寫入和傳輸?shù)臄?shù)據(jù)
• 一旦Map完成，則通知任務(wù)管理器，此時(shí)Reduce就可以開始復(fù)制結(jié)果數(shù)據(jù)

Reduce端

• Map的結(jié)果文件都存放到運(yùn)行Map任務(wù)的機(jī)器的本地硬盤中
• 如果Map的結(jié)果很少，則直接放到內(nèi)存，否則寫入文件中
• 同時(shí)后臺(tái)線程將這些文件進(jìn)行合并和排序到一個(gè)更大的文件中（如果文件是壓縮的，則需要先解壓）
• 當(dāng)所有的Map結(jié)果都被復(fù)制和合并后，就會(huì)調(diào)用Reduce方法
• Reduce結(jié)果會(huì)寫入到HDFS中

調(diào)優(yōu)

• 一般的原則是給shuffle分配盡可能多的內(nèi)存，但前提是要保證Map、Reduce任務(wù)有足夠的內(nèi)存
• 對(duì)于Map，主要就是避免把文件寫入磁盤，例如使用Combiner，增大io.sort.mb的值
• 對(duì)于Reduce，主要是把Map的結(jié)果盡可能地保存到內(nèi)存中，同樣也是要避免把中間結(jié)果寫入磁盤。默認(rèn)情況下，所有的內(nèi)存都是分配給Reduce方法的，如果Reduce方法不怎么消耗內(nèi)存，可以mapred.inmem.merge.threshold設(shè)成0，mapred.job.reduce.input.buffer.percent設(shè)成1.0
• 在任務(wù)監(jiān)控中可通過Spilled records counter來監(jiān)控寫入磁盤的數(shù)，但這個(gè)值是包括map和reduce的
• 對(duì)于IO方面，可以Map的結(jié)果可以使用壓縮，同時(shí)增大buffer size（io.file.buffer.size，默認(rèn)4kb）