柚子快報邀請碼778899分享：大數(shù)據(jù) hdfs Hadoop

http://yzkb.51969.com/

1 Hadoop常用端口號

hadoop2.xHadoop3.x訪問HDFS端口500709870訪問MR執(zhí)行情況端口80888088歷史服務(wù)器1988819888客戶端訪問集群端口90008020

2 Hadoop配置文件

??hadoop2.x core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml slaves

??hadoop3.x core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml workers

3 HDFS讀流程和寫流程

3.1 讀

??1）客戶端向namenode請求下載文件，namenode通過查詢元數(shù)據(jù)，找到文件塊所在的datanode地址。

??2）挑選一臺datanode（就近原則，然后隨機）服務(wù)器，請求讀取數(shù)據(jù)。

??3）datanode開始傳輸數(shù)據(jù)給客戶端（從磁盤里面讀取數(shù)據(jù)放入流，以packet為單位來做校驗）。

??4）客戶端以packet為單位接收，先在本地緩存，然后寫入目標文件。（后面的block塊Append到前面的block塊合成最終文件）

3.2 寫

??1）客戶端向namenode請求上傳文件，namenode檢查目標文件是否已存在，父目錄是否存在。

??2）namenode返回是否可以上傳。

??3）客戶端請求第一個 block上傳到哪幾個datanode服務(wù)器上。（文件先經(jīng)過切分處理）

??4）namenode返回3個datanode節(jié)點，分別為dn1、dn2、dn3。(遵循機架感知原則把副本分別放在不同機架，甚至不同數(shù)據(jù)中心)

??5）客戶端請求dn1上傳數(shù)據(jù)，dn1收到請求會繼續(xù)調(diào)用dn2，然后dn2調(diào)用dn3，將這個通信管道建立完成

??6）dn1、dn2、dn3逐級應(yīng)答客戶端

??7）客戶端開始往dn1上傳第一個block（先從磁盤讀取數(shù)據(jù)放到一個本地內(nèi)存緩存），以packet為單位，dn1收到一個packet就會傳給dn2，dn2傳給dn3；dn1每傳一個packet會放入一個應(yīng)答隊列等待應(yīng)答,當(dāng)一個block傳輸完成之后，客戶端再次請求namenode上傳第二個block的服務(wù)器。（重復(fù)執(zhí)行3-7步）。（數(shù)據(jù)傳輸完成后Datanode會向Client通信，同時向Namenode報告存儲完成）

4 secondary namenode工作機制

4.1 secondary namenode工作機制

1）第一階段：namenode啟動

??（1）第一次啟動namenode格式化后，創(chuàng)建fsimage和edits文件。如果不是第一次啟動，直接加載編輯日志和鏡像文件到內(nèi)存。

??（2）客戶端對元數(shù)據(jù)進行增刪改的請求

??（3）namenode記錄操作日志，更新滾動日志

??（4）namenode在內(nèi)存中對數(shù)據(jù)進行增刪改查

2）第二階段：Secondary NameNode工作

??（1）Secondary NameNode詢問namenode是否需要checkpoint。直接帶回namenode是否檢查結(jié)果。

??（2）Secondary NameNode請求執(zhí)行checkpoint。

?? （3）namenode滾動正在寫的edits日志

?? （4）將滾動前的編輯日志和鏡像文件拷貝到Secondary NameNode

?? （5）Secondary NameNode加載編輯日志和鏡像文件到內(nèi)存，并合并。

?? （6）生成新的鏡像文件fsimage.chkpoint

?? （7）拷貝fsimage.chkpoint到namenode

?? （8）namenode將fsimage.chkpoint重新命名成fsimage

4.2 NameNode 在啟動的時候會做哪些操作

??NameNode 數(shù)據(jù)存儲在內(nèi)存和本地磁盤，本地磁盤數(shù)據(jù)存儲在 fsimage 鏡像文件和edits 編輯日志文件。

4.2.1 首次啟動 NameNode：

??1. 格式化文件系統(tǒng),為了生成 fsimage 鏡像文件；

??2. 啟動 NameNode：

????? 讀取 fsimage 文件，將文件內(nèi)容加載進內(nèi)存

????? 等待 DataNade 注冊與發(fā)送 block report

??3. 啟動 DataNode：

????? 向 NameNode 注冊

????? 發(fā)送 block report

????? 檢查 fsimage 中記錄的塊的數(shù)量和 block report 中的塊的總數(shù)是否相同

??4. 對文件系統(tǒng)進行操作（創(chuàng)建目錄，上傳文件，刪除文件等）：

??此時內(nèi)存中已經(jīng)有文件系統(tǒng)改變的信息，但是磁盤中沒有文件系統(tǒng)改變的信息，此時會將這些改變信息寫入 edits 文件中，edits 文件中存儲的是文件系統(tǒng)元數(shù)據(jù)改變的信息。

4.2.2 第二次啟動 NameNode：

??1. 讀取 fsimage 和 edits 文件；

??2. 將 fsimage 和 edits 文件合并成新的 fsimage 文件；

??3. 創(chuàng)建新的 edits 文件，內(nèi)容開始為空；

??4. 啟動 DataNode。

5 NameNode與SecondaryNameNode 的區(qū)別與聯(lián)系？

5.1 區(qū)別

??（1）NameNode負責(zé)管理整個文件系統(tǒng)的元數(shù)據(jù)，以及每一個路徑（文件）所對應(yīng)的數(shù)據(jù)塊信息。

??（2）SecondaryNameNode主要用于定期合并命名空間鏡像和命名空間鏡像的編輯日志。

5.2 聯(lián)系：

?? （1）SecondaryNameNode中保存了一份和namenode一致的鏡像文件（fsimage）和編輯日志（edits）。

??（2）在主namenode發(fā)生故障時（假設(shè)沒有及時備份數(shù)據(jù)），可以從SecondaryNameNode恢復(fù)數(shù)據(jù)。

6 hadoop節(jié)點動態(tài)上線下線怎么操作?

6.1 節(jié)點上線操作

??當(dāng)要新上線數(shù)據(jù)節(jié)點的時候，需要把數(shù)據(jù)節(jié)點的名字追加在 dfs.hosts 文件中

??（1）關(guān)閉新增節(jié)點的防火墻 ??（2）在 NameNode 節(jié)點的 hosts 文件中加入新增數(shù)據(jù)節(jié)點的 hostname ??（3）在每個新增數(shù)據(jù)節(jié)點的 hosts 文件中加入 NameNode 的 hostname ??（4）在 NameNode 節(jié)點上增加新增節(jié)點的 SSH 免密碼登錄的操作 ??（5）在 NameNode 節(jié)點上的 dfs.hosts 中追加上新增節(jié)點的 hostname, ??（6）在其他節(jié)點上執(zhí)行刷新操作：hdfs dfsadmin -refreshNodes ??（7）在 NameNode 節(jié)點上，更改 slaves 文件，將要上線的數(shù)據(jù)節(jié)點 hostname 追加到 slaves 文件中 ??（8）啟動 DataNode 節(jié)點 ??（9）查看 NameNode 的監(jiān)控頁面看是否有新增加的節(jié)點

6.2 節(jié)點下線操作

??（1）修改/conf/hdfs-site.xml 文件 ??（2）確定需要下線的機器，dfs.osts.exclude 文件中配置好需要下架的機器，這個是阻 止下架的機器去連接 NameNode。 ??（3）配置完成之后進行配置的刷新操作./bin/hadoop dfsadmin -refreshNodes,這個操作的作用是在后臺進行 block 塊的移動。 ??（4）當(dāng)執(zhí)行三的命令完成之后，需要下架的機器就可以關(guān)閉了，可以查看現(xiàn)在集群上連接的節(jié)點，正在執(zhí)行 Decommission，會顯示：Decommission Status : Decommission in progress 執(zhí)行完畢后，會顯示：Decommission Status : Decommissioned ??（5）機器下線完畢，將他們從excludes 文件中移除。

7 HAnamenode 是如何工作的?

7.1 ZKFailoverController主要職責(zé)

??1）健康監(jiān)測：周期性的向它監(jiān)控的NN發(fā)送健康探測命令，從而來確定某個NameNode是否處于健康狀態(tài)，如果機器宕機，心跳失敗，那么zkfc就會標記它處于一個不健康的狀態(tài)。

??2）會話管理：如果NN是健康的，zkfc就會在zookeeper中保持一個打開的會話，如果NameNode同時還是Active狀態(tài)的，那么zkfc還會在Zookeeper中占有一個類型為短暫類型的znode，當(dāng)這個NN掛掉時，這個znode將會被刪除，然后備用的NN，將會得到這把鎖，升級為主NN，同時標記狀態(tài)為Active。

??3）當(dāng)宕機的NN新啟動時，它會再次注冊zookeper，發(fā)現(xiàn)已經(jīng)有znode鎖了，便會自動變?yōu)镾tandby狀態(tài)，如此往復(fù)循環(huán)，保證高可靠，需要注意，目前僅僅支持最多配置2個NN。

??4）master選舉：如上所述，通過在zookeeper中維持一個短暫類型的znode，來實現(xiàn)搶占式的鎖機制，從而判斷那個NameNode為Active狀態(tài)

7.2 NameNode HA

??1. 元數(shù)據(jù)信息同步在 HA 方案中采用的是“共享存儲”。每次寫文件時，需要將日志同步寫入共享存儲，這個步驟成功才能認定寫文件成功。然后備份節(jié)點定期從共享存儲同步日志，以便進行主備切換。

??NameNode 共享存儲方案有很多，比如 Linux HA, VMware FT, QJM等，目前社區(qū)已經(jīng)把由 Clouderea 公司實現(xiàn)的基于 QJM（Quorum Journal Manager）的方案合并到 HDFS 的trunk 之中并且作為默認的共享存儲實現(xiàn)。

??基于QJM 的共享存儲系統(tǒng)主要用于保存 EditLog，并不保存 FSImage 文件。FSImage文件還是在 NameNode 的本地磁盤上。

??QJM 共享存儲的基本思想來自于 Paxos 算法，采用多個稱為 JournalNode 的節(jié)點組成的JournalNode 集群來存儲 EditLog。每個 JournalNode 保存同樣的 EditLog 副本。每次 NameNode 寫EditLog 的時候，除了向本地磁盤寫入 EditLog 之外，也會并行地向 JournalNode 集群之中的每一個 JournalNode 發(fā)送寫請求，只要大多數(shù)的JournalNode 節(jié)點返回成功就認為向 JournalNode 集群寫入 EditLog 成功。如果有2N+1 臺 JournalNode，那么根據(jù)大多數(shù)的原則，最多可以容忍有 N 臺

??2. 監(jiān)控 NameNode 狀態(tài)采用 zookeeper，兩個 NameNode 節(jié)點的狀態(tài)存放在zookeeper 中，另外兩個 NameNode 節(jié)點分別有一個進程監(jiān)控程序，實施讀取 zookeeper 中有 NameNode 的狀態(tài)，來判斷當(dāng)前的 NameNode 是不是已經(jīng) down 機。如果 Standby 的 NameNode 節(jié)點的 ZKFC 發(fā)現(xiàn)主節(jié)點已經(jīng)掛掉，那么就會強制給原本的 Active NameNode 節(jié)點發(fā)送強制關(guān)閉請求，之后將備用的 NameNode 設(shè)置為 Active。

7.3 7. 在 NameNode HA 中，會出現(xiàn)腦裂問題嗎？怎么解決腦裂

??腦裂對于 NameNode 這類對數(shù)據(jù)一致性要求非常高的系統(tǒng)來說是災(zāi)難性的，數(shù)據(jù)會發(fā)生錯亂且無法恢復(fù)。zookeeper 社區(qū)對這種問題的解決方法叫做 fencing，中文翻譯為隔離，也就是想辦法把舊的 Active NameNode 隔離起來，使它不能正常對外提供服務(wù)。

??在進行 fencing 的時候，會執(zhí)行以下的操作：

??(1) 首先嘗試調(diào)用這個舊 Active NameNode 的 HAServiceProtocol RPC 接口的TransitionToStandby 方法，看能不能把它轉(zhuǎn)換為 Standby 狀態(tài)。

??(2) 如果 transitionToStandby 方法調(diào)用失敗，那么就執(zhí)行 Hadoop 配置文件之中預(yù)定義的隔離措施，Hadoop 目前主要提供兩種隔離措施，通常會選擇 sshfence：

????① sshfence：通過 SSH 登錄到目標機器上，執(zhí)行命令 fuser 將對應(yīng)的進程殺死；

????② shellfence：執(zhí)行一個用戶自定義的 shell 腳本來將對應(yīng)的進程隔離。

8 hadoop2.x Federation

??多namespace的方式可以直接減輕單一NameNode的壓力。

??NameNode共享集群中所有的DataNode，它們還是在同一個集群內(nèi).

??HDFS Federation方案的優(yōu)勢:

??第一點,命名空間的擴展。因為隨著集群使用時間的加長，HDFS上存放的數(shù)據(jù)也將會越來越多。這個時候如果還是將所有的數(shù)據(jù)都往一個NameNode上存放,這個文件系統(tǒng)會顯得非常的龐大。這時候我們可以進行橫向擴展，把一些大的目錄分離出去.使得每個NameNode下的數(shù)據(jù)看起來更加的精簡。

??第二點,性能的提升.這個也很好理解。當(dāng)NameNode所持有的數(shù)據(jù)量達到了一個非常大規(guī)模的量級的時候(比如超過1億個文件)，這個時候NameNode的處理效率可能就會有影響，它可能比較容易的會陷入一個繁忙的狀態(tài)。而整個集群將會受限于一個單點NameNode的處理效率,從而影響集群整體的吞吐量。這個時候多NameNode機制顯然可以減輕很多這部分的壓力。

??第三點,資源的隔離。這一點考慮的就比較深了。通過多個命名空間，我們可以將關(guān)鍵數(shù)據(jù)文件目錄移到不同的NameNode上，以此不讓這些關(guān)鍵數(shù)據(jù)的讀寫操作受到其他普通文件讀寫操作的影響。也就是說這些NameNode將會只處理特定的關(guān)鍵的任務(wù)所發(fā)來的請求，而屏蔽了其他普通任務(wù)的文件讀寫請求，以此做到了資源的隔離。千萬不要小看這一點，當(dāng)你發(fā)現(xiàn)NameNode正在處理某個不良任務(wù)的大規(guī)模的請求操作導(dǎo)致響應(yīng)速度極慢時，你一定會非常的懊惱。

9 TextInputFormat和KeyValueInputFormat的區(qū)別是什么？

??1）相同點：

??TextInputformat和KeyValueTextInputFormat都繼承了FileInputFormat類，都是每一行作為一個記錄；

??2）區(qū)別：

??TextInputformat將每一行在文件中的起始偏移量作為 key，每一行的內(nèi)容作為value。默認以\n或回車鍵作為一行記錄。

??KeyValueTextInputFormat 適合處理輸入數(shù)據(jù)的每一行是兩列，并用 tab 分離的形式。

10 FileInputFormat源碼解析(input.getSplits(job))

??（1）找到你數(shù)據(jù)存儲的目錄。

??（2）開始遍歷處理（規(guī)劃切片）目錄下的每一個文件

??（3）遍歷第一個文件ss.txt

?? ??a）獲取文件大小fs.sizeOf(ss.txt);

???? b）計算切片大小 ??????computeSliteSize(Math.max(minSize,Math.min(maxSize,blocksize)))=blocksize=128M

????c）默認情況下，切片大小=blocksize

???? d）開始切，形成第1個切片：ss.txt—0:128M 第2個切片ss.txt—128:256M 第3個切片ss.txt—256M:300M（每次切片時，都要判斷切完剩下的部分是否大于塊的1.1倍，不大于1.1倍就劃分一塊切片）

???? e）將切片信息寫到一個切片規(guī)劃文件中

???? f）整個切片的核心過程在getSplit()方法中完成。

????g）數(shù)據(jù)切片只是在邏輯上對輸入數(shù)據(jù)進行分片，并不會再磁盤上將其切分成分片進行存儲。InputSplit只記錄了分片的元數(shù)據(jù)信息，比如起始位置、長度以及所在的節(jié)點列表等。

????h）注意：block是HDFS上物理上存儲的存儲的數(shù)據(jù)，切片是對數(shù)據(jù)邏輯上的劃分。

?? （4）提交切片規(guī)劃文件到y(tǒng)arn上，yarn上的MrAppMaster就可以根據(jù)切片規(guī)劃文件計算開啟maptask個數(shù)。

12 job的map和reduce的數(shù)量?

12.1 map數(shù)量

?? splitSize=max{minSize,min{maxSize,blockSize}}

?? map數(shù)量由處理的數(shù)據(jù)分成的block數(shù)量決定default_num = total_size / split_size;

12.2 reduce數(shù)量

?? reduce的數(shù)量job.setNumReduceTasks(x);x 為reduce的數(shù)量,不設(shè)置的話默認為 1。

13 MapTask工作機制

??（1）Read階段：Map Task通過用戶編寫的RecordReader，從輸入InputSplit中解析出一個個key/value。

??（2）Map階段：該節(jié)點主要是將解析出的key/value交給用戶編寫map()函數(shù)處理，并產(chǎn)生一系列新的key/value。

??（3）Collect收集階段：在用戶編寫map()函數(shù)中，當(dāng)數(shù)據(jù)處理完成后，一般會調(diào)用OutputCollector.collect()輸出結(jié)果。在該函數(shù)內(nèi)部，它會將生成的key/value分區(qū)（調(diào)用Partitioner），并寫入一個環(huán)形內(nèi)存緩沖區(qū)中。

??（4）Spill階段：即“溢寫”，當(dāng)環(huán)形緩沖區(qū)滿后，MapReduce會將數(shù)據(jù)寫到本地磁盤上，生成一個臨時文件。需要注意的是，將數(shù)據(jù)寫入本地磁盤之前，先要對數(shù)據(jù)進行一次本地排序，并在必要時對數(shù)據(jù)進行合并、壓縮等操作。

??溢寫階段詳情：

??步驟1：利用快速排序算法對緩存區(qū)內(nèi)的數(shù)據(jù)進行排序，排序方式是，先按照分區(qū)編號partition進行排序，然后按照key進行排序。這樣，經(jīng)過排序后，數(shù)據(jù)以分區(qū)為單位聚集在一起，且同一分區(qū)內(nèi)所有數(shù)據(jù)按照key有序。

??步驟2：按照分區(qū)編號由小到大依次將每個分區(qū)中的數(shù)據(jù)寫入任務(wù)工作目錄下的臨時文件output/spillN.out（N表示當(dāng)前溢寫次數(shù)）中。如果用戶設(shè)置了Combiner，則寫入文件之前，對每個分區(qū)中的數(shù)據(jù)進行一次聚集操作。

??步驟3：將分區(qū)數(shù)據(jù)的元信息寫到內(nèi)存索引數(shù)據(jù)結(jié)構(gòu)SpillRecord中，其中每個分區(qū)的元信息包括在臨時文件中的偏移量、壓縮前數(shù)據(jù)大小和壓縮后數(shù)據(jù)大小。如果當(dāng)前內(nèi)存索引大小超過1MB，則將內(nèi)存索引寫到文件output/spillN.out.index中。

??（5）Combine階段：當(dāng)所有數(shù)據(jù)處理完成后，MapTask對所有臨時文件進行一次合并(多路歸并算法)，以確保最終只會生成一個數(shù)據(jù)文件。

??當(dāng)所有數(shù)據(jù)處理完后，MapTask會將所有臨時文件合并成一個大文件，并保存到文件output/file.out中，同時生成相應(yīng)的索引文件output/file.out.index。

??在進行文件合并過程中，MapTask以分區(qū)為單位進行合并。對于某個分區(qū)，它將采用多輪遞歸合并的方式。每輪合并io.sort.factor（默認100）個文件，并將產(chǎn)生的文件重新加入待合并列表中，對文件排序后，重復(fù)以上過程，直到最終得到一個大文件。

??讓每個MapTask最終只生成一個數(shù)據(jù)文件，可避免同時打開大量文件和同時讀取大量小文件產(chǎn)生的隨機讀取帶來的開銷。

14 ReduceTask工作機制

??（1）Copy階段：ReduceTask從各個MapTask上遠程拷貝一片數(shù)據(jù)，并針對某一片數(shù)據(jù)，如果其大小超過一定閾值，則寫到磁盤上，否則直接放到內(nèi)存中。

?? （2）Merge階段：在遠程拷貝數(shù)據(jù)的同時，ReduceTask啟動了兩個后臺線程對內(nèi)存和磁盤上的文件進行合并（合并同一個分區(qū)號的數(shù)據(jù)），以防止內(nèi)存使用過多或磁盤上文件過多。

?? （3）Sort階段：按照MapReduce語義，用戶編寫reduce()函數(shù)輸入數(shù)據(jù)是按key進行聚集的一組數(shù)據(jù)。為了將key相同的數(shù)據(jù)聚在一起，Hadoop采用了基于排序的策略。由于各個MapTask已經(jīng)實現(xiàn)對自己的處理結(jié)果進行了局部排序，因此，ReduceTask只需對所有數(shù)據(jù)進行一次歸并排序即可。

?? （4）Reduce階段：reduce()函數(shù)將計算結(jié)果寫到HDFS上。

15 mapReduce有幾種排序及排序發(fā)生的階段

15.1 排序的分類：

?? （1）部分排序：

????MapReduce根據(jù)輸入記錄的鍵對數(shù)據(jù)集排序。保證輸出的每個文件內(nèi)部排序。

?? （2）全排序：

????如何用Hadoop產(chǎn)生一個全局排序的文件？最簡單的方法是使用一個分區(qū)。但該方法在處理大型文件時效率極低，因為一臺機器必須處理所有輸出文件，從而完全喪失了MapReduce所提供的并行架構(gòu)。

????替代方案：首先創(chuàng)建一系列排好序的文件；其次，串聯(lián)這些文件；最后，生成一個全局排序的文件。主要思路是使用一個分區(qū)來描述輸出的全局排序。例如：可以為待分析文件創(chuàng)建3個分區(qū)，在第一分區(qū)中，記錄的單詞首字母a-g，第二分區(qū)記錄單詞首字母h-n, 第三分區(qū)記錄單詞首字母o-z。

??（3）輔助排序：（GroupingComparator分組）

??Mapreduce框架在記錄到達reducer之前按鍵對記錄排序，但鍵所對應(yīng)的值并沒有被排序。甚至在不同的執(zhí)行輪次中，這些值的排序也不固定，因為它們來自不同的map任務(wù)且這些map任務(wù)在不同輪次中完成時間各不相同。一般來說，大多數(shù)MapReduce程序會避免讓reduce函數(shù)依賴于值的排序。但是，有時也需要通過特定的方法對鍵進行排序和分組等以實現(xiàn)對值的排序。 ??

（4）二次排序：

??在自定義排序過程中，如果compareTo中的判斷條件為兩個即為二次排序。

15.2 自定義排序WritableComparable

??bean對象實現(xiàn)WritableComparable接口重寫compareTo方法，就可以實現(xiàn)排序

@Override

public int compareTo(FlowBean o) {

// 倒序排列，從大到小

return this.sumFlow > o.getSumFlow() ? -1 : 1;

}

15.3 排序發(fā)生的階段：

（1）一個是在map side發(fā)生在spill后partition前。

（2）一個是在reduce side發(fā)生在copy后 reduce前。

16 mapReduce中combiner的作用是什么，一般使用情景，哪些情況不需要，及和reduce的區(qū)別？

??1）Combiner的意義就是對每一個maptask的輸出進行局部匯總，以減小網(wǎng)絡(luò)傳輸量。

??2）Combiner能夠應(yīng)用的前提不能夠影響任務(wù)的運行結(jié)果的局部匯總，適用于求和類，不適用于求平均值，而且Combiner的輸出kv應(yīng)該跟reducer的輸入kv類型要對應(yīng)起來。

??3）Combiner和reducer的區(qū)別在于運行的位置:

????Combiner是在每一個maptask所在的節(jié)點運行。

????Reducer是接收全局所有Mapper的輸出結(jié)果。

17 Yarn提交任務(wù)流程

18 HDFS的數(shù)據(jù)壓縮算法

??Hadoop中常用的壓縮算法有bzip2、gzip、lzo、snappy，其中l(wèi)zo、snappy需要操作系統(tǒng)安裝native庫才可以支持。企業(yè)開發(fā)用的比較多的是snappy

19 Hadoop的調(diào)度器總結(jié)

??Apache默認的資源調(diào)度器是容量調(diào)度器；CDH默認的資源調(diào)度器是公平調(diào)度器。

（1）FIFO

??先按照作業(yè)的優(yōu)先級高低，再按照到達時間的先后選擇被執(zhí)行的作業(yè)。

（2）計算能力調(diào)度器Capacity Scheduler

??支持多個隊列，每個隊列可配置一定的資源量，每個隊列采用FIFO調(diào)度策略，為了防止同一個用戶的作業(yè)獨占隊列中的資源，該調(diào)度器會對同一用戶提交的作業(yè)所占資源量進行限定。調(diào)度時，首先按以下策略選擇一個合適隊列：計算每個隊列中正在運行的任務(wù)數(shù)與其應(yīng)該分得的計算資源之間的比值，選擇一個該比值最小的隊列；然后按以下策略選擇該隊列中一個作業(yè)：按照作業(yè)優(yōu)先級和提交時間順序選擇，同時考慮用戶資源量限制和內(nèi)存限制。

（3）公平調(diào)度器Fair Scheduler

??同計算能力調(diào)度器類似，支持多隊列多用戶，每個隊列中的資源量可以配置，同一隊列中的作業(yè)公平共享隊列中所有資源，在 Fair 調(diào)度器中，我們不需要預(yù)先占用一定的系統(tǒng)資源，F(xiàn)air 調(diào)度器會為所有運行的 job 動態(tài)的調(diào)整系統(tǒng)資源。

??在 Fair 調(diào)度器中，從第二個任務(wù)提交到獲得資源會有一定的延遲，因為它需要等待第一個任務(wù)釋放占用的 Container。小任務(wù)執(zhí)行完成之后也會釋放自己占用的資源，大任務(wù)又獲得了全部的系統(tǒng)資源。最終的效果就是 Fair調(diào)度器即得到了高的資源利用率又能保證小任務(wù)及時完成。

??調(diào)度時計算每個任務(wù)應(yīng)該分得的計算資源與實際獲得的資源差值，差值越大作業(yè)優(yōu)先級越高。

??實際上，Hadoop的調(diào)度器遠不止以上三種，最近，出現(xiàn)了很多針對新型應(yīng)用的Hadoop調(diào)度器。

（4）適用于異構(gòu)集群的調(diào)度器LATE

??現(xiàn)有的Hadoop調(diào)度器都是建立在同構(gòu)集群的假設(shè)前提下，具體假設(shè)如下：

????1）集群中各個節(jié)點的性能完全一樣 ????2）對于reduce task，它的三個階段：copy、sort和reduce，用時各占1/3 ????3）同一job的同類型的task是一批一批完成的，他們用時基本一樣。

??現(xiàn)有的Hadoop調(diào)度器存在較大缺陷，主要體現(xiàn)在探測落后任務(wù)的算法上：如果一個task的進度落后于同類型task進度的20%，則把該task當(dāng)做落后任務(wù)(這種任務(wù)決定了job的完成時間，需盡量縮短它的執(zhí)行時間)，從而為它啟動一個備份任務(wù)（speculative task）。如果集群異構(gòu)的，對于同一個task，即使是在相同節(jié)點上的執(zhí)行時間也會有較大差別，因而在異構(gòu)集群中很容易產(chǎn)生大量的備份任務(wù)。

??LATE（Longest Approximate Time to End，參考資料[4]）調(diào)度器從某種程度上解決了現(xiàn)有調(diào)度器的問題，它定義三個閾值：SpeculativeCap，系統(tǒng)中最大同時執(zhí)行的speculative task數(shù)目（作者推薦值為總slot數(shù)的10%）；

??SlowNodeThreshold（作者推薦值為25%）：得分（分數(shù)計算方法見論文）低于該閾值的node（快節(jié)點）上不會啟動speculative task；SlowTaskThreshold（作者推薦值為25%）：當(dāng)task進度低于同批同類task的平均進度的SlowTaskThreshold時，會為該task啟動speculative task。它的調(diào)度策略是：當(dāng)一個節(jié)點出現(xiàn)空閑資源且系統(tǒng)中總的備份任務(wù)數(shù)小于SpeculativeCap時，（1）如果該節(jié)點是慢節(jié)點（節(jié)點得分高于SlowNodeThreshold），則忽略這個請求。 （2）對當(dāng)前正在運行的task按估算的剩余完成時間排序 （3）選擇剩余完成時間最大且進度低于SlowTaskThreshold的task，為該task啟動備份任務(wù)。

（5）適用于實時作業(yè)的調(diào)度器Deadline Scheduler和Constraint-based Scheduler

??這種調(diào)度器主要用于有時間限制的作業(yè)（Deadline Job），即給作業(yè)一個deadline時間，讓它在該時間內(nèi)完成。實際上，這類調(diào)度器分為兩種，軟實時（允許作業(yè)有一定的超時）作業(yè)調(diào)度器和硬實時（作業(yè)必須嚴格按時完成）作業(yè)調(diào)度器。

??Deadline Scheduler（參考資料[5]）主要針對的是軟實時作業(yè)，該調(diào)度器根據(jù)作業(yè)的運行進度和剩余時間動態(tài)調(diào)整作業(yè)獲得的資源量，以便作業(yè)盡可能的在deadline時間內(nèi)完成。

??Constraint-based Scheduler（參考資料[6]）主要針對的是硬實時作業(yè)，該調(diào)度器根據(jù)作業(yè)的deadline和當(dāng)前系統(tǒng)中的實時作業(yè)運行情況，預(yù)測新提交的實時作業(yè)能不能在deadline時間內(nèi)完成，如果不能，則將作業(yè)反饋給用戶，讓他重調(diào)整作業(yè)的deadline。

20 mapreduce 優(yōu)化方法

20.1 數(shù)據(jù)輸入

（1）合并小文件：在執(zhí)行mr任務(wù)前將小文件進行合并，大量的小文件會產(chǎn)生大量的map任務(wù)，增大map任務(wù)裝載次數(shù)，而任務(wù)的裝載比較耗時，從而導(dǎo)致 mr 運行較慢。

（2）采用ConbinFileInputFormat來作為輸入，解決輸入端大量小文件場景。

20.2 map階段

（1）減少spill次數(shù)：通過調(diào)整io.sort.mb(默認100MB)及sort.spill.percent(默認80%)參數(shù)值，增大觸發(fā)spill的內(nèi)存上限，減少spill次數(shù)，從而減少磁盤 IO。

（2）減少merge次數(shù)：通過調(diào)整io.sort.factor(默認10)參數(shù)，增大merge的文件數(shù)目，減少merge的次數(shù)，從而縮短mr處理時間。

（3）不影響業(yè)務(wù)前提下提前在 map 進行Combiner處理，減少 I/O。

20.3 reduce階段

（1）合理設(shè)置map和reduce數(shù)：兩個都不能設(shè)置太少，也不能設(shè)置太多。太少，會導(dǎo)致task等待，延長處理時間；太多，會導(dǎo)致 map、reduce任務(wù)間競爭資源，造成處理超時等錯誤。

（2）設(shè)置map、reduce共存：調(diào)整slowstart.completedmaps參數(shù)，使map運行到一定程度后，reduce也開始運行，減少reduce的等待時間。

（3）規(guī)避使用reduce，因為Reduce在用于連接數(shù)據(jù)集的時候?qū)a(chǎn)生大量的網(wǎng)絡(luò)消耗。

（4）合理設(shè)置reduc端的buffer，默認情況下，數(shù)據(jù)達到一個閾值的時候，buffer中的數(shù)據(jù)就會寫入磁盤，然后reduce會從磁盤中獲得所有的數(shù)據(jù)。也就是說，buffer和reduce是沒有直接關(guān)聯(lián)的，中間多個一個寫磁盤->讀磁盤的過程，既然有這個弊端，那么就可以通過參數(shù)來配置，使得buffer中的一部分數(shù)據(jù)可以直接輸送到reduce，從而減少IO開銷：mapred.job.reduce.input.buffer.percent，默認為0.0。當(dāng)值大于0的時候，會保留指定比例的內(nèi)存讀buffer中的數(shù)據(jù)直接拿給reduce使用。這樣一來，設(shè)置buffer需要內(nèi)存，讀取數(shù)據(jù)需要內(nèi)存，reduce計算也要內(nèi)存，所以要根據(jù)作業(yè)的運行情況進行調(diào)整。

20.4 IO傳輸

（1）采用數(shù)據(jù)壓縮的方式，減少網(wǎng)絡(luò)IO的的時間。安裝Snappy和LZOP壓縮編碼器。

（2）使用SequenceFile二進制文件

20.5 數(shù)據(jù)傾斜問題

（1）數(shù)據(jù)傾斜現(xiàn)象

??數(shù)據(jù)頻率傾斜——某一個區(qū)域的數(shù)據(jù)量要遠遠大于其他區(qū)域。

??數(shù)據(jù)大小傾斜——部分記錄的大小遠遠大于平均值。

（2）如何收集傾斜數(shù)據(jù)

??在reduce方法中加入記錄map輸出鍵的詳細情況的功能。

public static final String MAX_VALUES = "skew.maxvalues";

private int maxValueThreshold;

@Override

public void configure(JobConf job) {

maxValueThreshold = job.getInt(MAX_VALUES, 100);

}

@Override

public void reduce(Text key, Iterator values,

OutputCollector output,

Reporter reporter) throws IOException {

int i = 0;

while (values.hasNext()) {

values.next();

i++;

}

if (++i > maxValueThreshold) {

log.info("Received " + i + " values for key " + key);

}

}

（3）減少數(shù)據(jù)傾斜的方法

方法1：抽樣和范圍分區(qū)

??可以通過對原始數(shù)據(jù)進行抽樣得到的結(jié)果集來預(yù)設(shè)分區(qū)邊界值。

方法2：自定義分區(qū)

??根據(jù)數(shù)據(jù)分布情況，自定義散列函數(shù)，將key均勻分配到不同Reducer。例如，如果map輸出鍵的單詞來源于一本書。其中大部分必然是省略詞（stopword）。那么就可以將自定義分區(qū)將這部分省略詞發(fā)送給固定的一部分reduce實例。而將其他的都發(fā)送給剩余的reduce實例。

方法3：Combine

??使用Combine可以大量地減小數(shù)據(jù)頻率傾斜和數(shù)據(jù)大小傾斜。在可能的情況下，combine的目的就是聚合并精簡數(shù)據(jù)。

方法4：局部聚合加全局聚合

??第一次在map階段對那些導(dǎo)致了數(shù)據(jù)傾斜的key 加上1到n的隨機前綴，這樣本來相同的key 也會被分到多個Reducer中進行局部聚合，數(shù)量就會大大降低。

??第二次mapreduce，去掉key的隨機前綴，進行全局聚合。

??思想：二次mr，第一次將key隨機散列到不同reducer進行處理達到負載均衡目的。第二次再根據(jù)去掉key的隨機前綴，按原key進行reduce處理。 這個方法進行兩次mapreduce，性能稍差。

方法5：增加Reducer，提升并行度

??JobConf.setNumReduceTasks(int)

20.6 常用的調(diào)優(yōu)參數(shù)

（1）資源相關(guān)參數(shù)

（a）以下參數(shù)是在用戶自己的mr應(yīng)用程序中配置就可以生效（mapred-default.xml）

配置參數(shù)參數(shù)說明mapreduce.map.memory.mb一個Map Task可使用的資源上限（單位:MB），默認為1024。如果Map Task實際使用的資源量超過該值，則會被強制殺死。mapreduce.reduce.memory.mb一個Reduce Task可使用的資源上限（單位:MB），默認為1024。如果Reduce Task實際使用的資源量超過該值，則會被強制殺死。mapreduce.map.cpu.vcores每個Map task可使用的最多cpu core數(shù)目，默認值: 1mapreduce.reduce.cpu.vcores每個Reduce task可使用的最多cpu core數(shù)目，默認值: 1mapreduce.reduce.shuffle.parallelcopies每個reduce去map中拿數(shù)據(jù)的并行數(shù)。默認值是5mapreduce.reduce.shuffle.merge.percentbuffer中的數(shù)據(jù)達到多少比例開始寫入磁盤。默認值0.66mapreduce.reduce.shuffle.input.buffer.percentbuffer大小占reduce可用內(nèi)存的比例。默認值0.7mapreduce.reduce.input.buffer.percent指定多少比例的內(nèi)存用來存放buffer中的數(shù)據(jù)，默認值是0.0

（b）應(yīng)該在yarn啟動之前就配置在服務(wù)器的配置文件中才能生效（yarn-default.xml）

配置參數(shù)參數(shù)說明yarn.scheduler.minimum-allocation-mb 1024給應(yīng)用程序container分配的最小內(nèi)存yarn.scheduler.maximum-allocation-mb 8192給應(yīng)用程序container分配的最大內(nèi)存yarn.scheduler.minimum-allocation-vcores 1每個container申請的最小CPU核數(shù)yarn.scheduler.maximum-allocation-vcores 32每個container申請的最大CPU核數(shù)yarn.nodemanager.resource.memory-mb 8192給containers分配的最大物理內(nèi)存

（c）shuffle性能優(yōu)化的關(guān)鍵參數(shù)，應(yīng)在yarn啟動之前就配置好（mapred-default.xml）

配置參數(shù)參數(shù)說明mapreduce.task.io.sort.mb 100shuffle的環(huán)形緩沖區(qū)大小，默認100mmapreduce.map.sort.spill.percent 0.8環(huán)形緩沖區(qū)溢出的閾值，默認80%

（2）容錯相關(guān)參數(shù)(mapreduce性能優(yōu)化)

配置參數(shù)參數(shù)說明mapreduce.map.maxattempts每個Map Task最大重試次數(shù)，一旦重試參數(shù)超過該值，則認為Map Task運行失敗，默認值：4。mapreduce.reduce.maxattempts每個Reduce Task最大重試次數(shù)，一旦重試參數(shù)超過該值，則認為Map Task運行失敗，默認值：4。mapreduce.task.timeoutTask超時時間，經(jīng)常需要設(shè)置的一個參數(shù)，該參數(shù)表達的意思為：如果一個task在一定時間內(nèi)沒有任何進入，即不會讀取新的數(shù)據(jù)，也沒有輸出數(shù)據(jù)，則認為該task處于block狀態(tài)，可能是卡住了，也許永遠會卡主，為了防止因為用戶程序永遠block住不退出，則強制設(shè)置了一個該超時時間（單位毫秒），默認是600000。如果你的程序?qū)γ織l輸入數(shù)據(jù)的處理時間過長（比如會訪問數(shù)據(jù)庫，通過網(wǎng)絡(luò)拉取數(shù)據(jù)等），建議將該參數(shù)調(diào)大，該參數(shù)過小常出現(xiàn)的錯誤提示是“AttemptID:attempt_14267829456721_123456_m_000224_0 Timed out after 300 secsContainer killed by the ApplicationMaster.”。

21 HDFS小文件優(yōu)化方法

1）HDFS小文件弊端：

??HDFS上每個文件都要在namenode上建立一個索引，這個索引的大小約為150byte，這樣當(dāng)小文件比較多的時候，就會產(chǎn)生很多的索引文件，一方面會大量占用namenode的內(nèi)存空間，另一方面就是索引文件過大導(dǎo)致索引速度變慢。計算方面，每個小文件都會起到一個MapTask，1個MapTask默認內(nèi)存1G。浪費資源。

2）解決的方式：

（1）Hadoop Archive:

??是一個高效地將小文件放入HDFS塊中的文件存檔工具，它能夠?qū)⒍鄠€小文件打包成一個HAR文件，以減少namenode的內(nèi)存使用。

（2）CombineFileInputFormat：

??CombineFileInputFormat是一種新的inputformat，用于將多個文件合并成一個單獨的split，另外，它會考慮數(shù)據(jù)的存儲位置。

（3）有小文件場景開啟JVM重用；如果沒有小文件，不要開啟JVM重用，因為會一直占用使用到的task卡槽，直到任務(wù)完成才釋放。

??JVM重用可以使得JVM實例在同一個job中重新使用N次，N的值可以在Hadoop的mapred-site.xml文件中進行配置。通常在10-20之間

mapreduce.job.jvm.numtasks

10

How many tasks to run per jvm,if set to -1 ,there is no limit

22 HDFS的NameNode內(nèi)存

??1）Hadoop2.x系列，配置NameNode默認2000m

??2）Hadoop3.x系列，配置NameNode內(nèi)存是動態(tài)分配的

??NameNode內(nèi)存最小值1G，每增加100萬個block，增加1G內(nèi)存。

23 NameNode心跳并發(fā)配置

??企業(yè)經(jīng)驗：dfs.namenode.handler.count=20×〖log〗_e^(Cluster Size)，比如集群規(guī)模（DataNode臺數(shù)）為3臺時，此參數(shù)設(shè)置為21。

24 糾刪碼原理

??CPU資源換取存儲空間

25 異構(gòu)存儲（冷熱數(shù)據(jù)分離）

??期望經(jīng)常使用的數(shù)據(jù)存儲在固態(tài)硬盤或者內(nèi)存鏡像硬盤；不經(jīng)常使用的歷史數(shù)據(jù)存儲在老舊的破舊硬盤。

26 Hadoop宕機

??1）如果MR造成系統(tǒng)宕機。此時要控制Yarn同時運行的任務(wù)數(shù)，和每個任務(wù)申請的最大內(nèi)存。調(diào)整參數(shù)：yarn.scheduler.maximum-allocation-mb（單個任務(wù)可申請的最多物理內(nèi)存量，默認是8192MB）

??2）如果寫入文件過快造成NameNode宕機。那么調(diào)高Kafka的存儲大小，控制從Kafka到HDFS的寫入速度。例如，可以調(diào)整Flume每批次拉取數(shù)據(jù)量的大小參數(shù)batchsize。

27 HDFS 在讀取文件的時候， 如果其中一個塊突然損壞了怎么辦?

??客戶端讀取完 DataNode 上的塊之后會進行 checksum 驗證， 也就是把客戶端讀取到本地的塊與 HDFS 上的原始塊進行校驗， 如果發(fā)現(xiàn)校驗結(jié)果不一致， 客戶端會通知 NameNode， 然后再從下一個擁有該 block 副本的 DataNode 繼續(xù)讀。

28 HDFS 在上傳文件的時候，如果其中一個 DataNode突然掛掉了怎么辦?

??客戶端上傳文件時與 DataNode 建立 pipeline 管道，管道的正方向是客戶端向DataNode 發(fā)送的數(shù)據(jù)包，管道反向是 DataNode 向客戶端發(fā)送 ack 確認，也就是正確接收到數(shù)據(jù)包之后發(fā)送一個已確認接收到的應(yīng)答。

??當(dāng) DataNode 突然掛掉了，客戶端接收不到這個 DataNode 發(fā)送的 ack 確認，客戶端會通知NameNode，NameNode 檢查該塊的副本與規(guī)定的不符，NameNode會通知DataNode去復(fù)制副本，并將掛掉的 DataNode 作下線處理，不再讓它參與文件上傳與下載。

柚子快報邀請碼778899分享：大數(shù)據(jù) hdfs Hadoop

http://yzkb.51969.com/

推薦文章