Map Reduce & YARN

A、簡(jiǎn)介

Apache Hadoop 是一個(gè)開源軟件框架，可安裝在一個(gè)商用機(jī)器集群中，使機(jī)器可彼此通信并協(xié)同工作，以高度分布式的方式共同存儲(chǔ)和處理大量數(shù)據(jù)。最初，Hadoop 包含以下兩個(gè)主要組件：Hadoop Distributed File System (HDFS) 和一個(gè)分布式計(jì)算引擎，該引擎支持以 MapReduce 作業(yè)的形式實(shí)現(xiàn)和運(yùn)行程序。

MapReduce 是 Google 推廣的一個(gè)簡(jiǎn)單的編程模型，它對(duì)以高度并行和可擴(kuò)展的方式處理大數(shù)據(jù)集很有用。MapReduce 的靈感來源于函數(shù)式編程，用戶可將他們的計(jì)算表達(dá)為 map 和 reduce 函數(shù)，將數(shù)據(jù)作為鍵值對(duì)來處理。Hadoop 提供了一個(gè)高級(jí) API 來在各種語言中實(shí)現(xiàn)自定義的 map 和 reduce 函數(shù)。

Hadoop 還提供了軟件基礎(chǔ)架構(gòu)，以一系列 map 和 reduce 任務(wù)的形式運(yùn)行 MapReduce 作業(yè)。Map 任務(wù) 在輸入數(shù)據(jù)的子集上調(diào)用 map 函數(shù)。在完成這些調(diào)用后，reduce 任務(wù) 開始在 map 函數(shù)所生成的中間數(shù)據(jù)上調(diào)用 reduce 任務(wù)，生成最終的輸出。 map 和 reduce 任務(wù)彼此單獨(dú)運(yùn)行，這支持并行和容錯(cuò)的計(jì)算。

最重要的是，Hadoop 基礎(chǔ)架構(gòu)負(fù)責(zé)處理分布式處理的所有復(fù)雜方面：并行化、調(diào)度、資源管理、機(jī)器間通信、軟件和硬件故障處理，等等。得益于這種干凈的抽象，實(shí)現(xiàn)處理數(shù)百(或者甚至數(shù)千)個(gè)機(jī)器上的數(shù) TB 數(shù)據(jù)的分布式應(yīng)用程序從未像現(xiàn)在這么容易過，甚至對(duì)于之前沒有使用分布式系統(tǒng)的經(jīng)驗(yàn)的開發(fā)人員也是如此。

B、MR架構(gòu)

將任務(wù)分割為 Map 端和 reduce 端。

一、JobClient JobTracker TaskTracker

• JobClient 向 JobTracker 請(qǐng)求一個(gè)新的 jobID
• 檢查作業(yè)輸出說明
• 計(jì)算作業(yè)輸出劃分split
• 將運(yùn)行作業(yè)所需要的資源(作業(yè)的jar文件、配置文件、計(jì)算所得的輸入劃分)復(fù)制到一個(gè)以作業(yè)ID命名的目錄中JobTracker的文件系統(tǒng)。

通過調(diào)用JobTracker的submitJob()方法，告訴JobTracker作業(yè)準(zhǔn)備執(zhí)行

JobTracker接收到submitJob()方法調(diào)用后，把此調(diào)用放到一個(gè)內(nèi)部隊(duì)列中，交由作業(yè)調(diào)度器進(jìn)行調(diào)度，并對(duì)其進(jìn)行初始化

創(chuàng)建運(yùn)行任務(wù)列表，作業(yè)調(diào)度去首先從共享文件系統(tǒng)中獲取JobClient已經(jīng)計(jì)算好的輸入劃分信息(圖中step6)，然后為每個(gè)劃分創(chuàng)建一個(gè)Map任務(wù)(一個(gè)split對(duì)應(yīng)一個(gè)map，有多少split就有多少map)。

TaskTracker執(zhí)行一個(gè)簡(jiǎn)單的循環(huán)，定期發(fā)送心跳(heartbeat)調(diào)用JobTracker

二、shuffle combine

整體的Shuffle過程包含以下幾個(gè)部分：Map端Shuffle、Sort階段、Reduce端Shuffle。即是說：Shuffle 過程橫跨 map 和 reduce 兩端，中間包含 sort 階段，就是數(shù)據(jù)從 map task 輸出到reduce task輸入的這段過程。

sort、combine 是在 map 端的，combine 是提前的 reduce ，需要自己設(shè)置。

Hadoop 集群中，大部分 map task 與 reduce task 的執(zhí)行是在不同的節(jié)點(diǎn)上。當(dāng)然很多情況下 Reduce 執(zhí)行時(shí)需要跨節(jié)點(diǎn)去拉取其它節(jié)點(diǎn)上的map task結(jié)果。如果集群正在運(yùn)行的 job 有很多，那么 task 的正常執(zhí)行對(duì)集群內(nèi)部的網(wǎng)絡(luò)資源消耗會(huì)很嚴(yán)重。而對(duì)于必要的網(wǎng)絡(luò)資源消耗，最終的目的就是***化地減少不必要的消耗。還有在節(jié)點(diǎn)內(nèi)，相比于內(nèi)存，磁盤 IO 對(duì) job 完成時(shí)間的影響也是可觀的。從最基本的要求來說，對(duì)于 MapReduce 的 job 性能調(diào)優(yōu)的 Shuffle 過程，目標(biāo)期望可以有：

• 完整地從map task端拉取數(shù)據(jù)到reduce 端。
• 在跨節(jié)點(diǎn)拉取數(shù)據(jù)時(shí)，盡可能地減少對(duì)帶寬的不必要消耗。
• 減少磁盤IO對(duì)task執(zhí)行的影響。
• 總體來講這段Shuffle過程，能優(yōu)化的地方主要在于減少拉取數(shù)據(jù)的量及盡量使用內(nèi)存而不是磁盤。

三、Map Shuffle

1、輸入

在map task 執(zhí)行時(shí)，其輸入來源 HDFS的 block ，map task 只讀取split 。Split 與 block 的對(duì)應(yīng)關(guān)系可能是多對(duì)一，默認(rèn)為一對(duì)一。

2、切分

決定于當(dāng)前的 mapper的 part交給哪個(gè) reduce的方法是：mapreduce 提供的Partitioner接口，對(duì)key 進(jìn)行 hash 后，再以 reducetask 數(shù)量取模，然后到指定的 job 上。

然后將數(shù)據(jù)寫入內(nèi)存緩沖區(qū)中，緩沖區(qū)的作用是批量收集map結(jié)果，減少磁盤IO的影響。key/value對(duì)以及 Partition 的結(jié)果都會(huì)被寫入緩沖區(qū)。寫入之前，key 與value 值都會(huì)被序列化成字節(jié)數(shù)組。

3、溢寫

由于內(nèi)存緩沖區(qū)的大小限制(默認(rèn)100MB)，當(dāng)map task輸出結(jié)果很多時(shí)就可能發(fā)生內(nèi)存溢出，所以需要在一定條件下將緩沖區(qū)的數(shù)據(jù)臨時(shí)寫入磁盤，然后重新利用這塊緩沖區(qū)。這個(gè)從內(nèi)存往磁盤寫數(shù)據(jù)的過程被稱為Spill，中文可譯為溢寫。

這個(gè)溢寫是由另外單獨(dú)線程來完成，不影響往緩沖區(qū)寫map結(jié)果的線程。

整個(gè)緩沖區(qū)有個(gè)溢寫的比例spill.percent。這個(gè)比例默認(rèn)是0.8，

Combiner 將有相同key的 key/value 對(duì)加起來，減少溢寫spill到磁盤的數(shù)據(jù)量。Combiner的適用場(chǎng)景：由于Combiner的輸出是Reducer的輸入，Combiner絕不能改變最終的計(jì)算結(jié)果。故大多數(shù)情況下，combiner適用于輸入輸出的key/value類型完全一致，且不影響最終結(jié)果的場(chǎng)景(比如累加、***值等……)。

4、Merge

map 很大時(shí)，每次溢寫會(huì)產(chǎn)生一個(gè) spill_file，這樣會(huì)有多個(gè) spill_file，而最終的輸出只有一個(gè)文件，在最終輸出之前會(huì)對(duì)多個(gè)中間過程多次產(chǎn)生的溢寫文件 spill_file 進(jìn)行合并，此過程就是 merge。

merge 就是把相同 key 的結(jié)果加起來。(當(dāng)然，如果設(shè)置過combiner，也會(huì)使用combiner來合并相同的key)

四、Reduce Shuffle

1、reduce shuffle

在 reduce task 之前，不斷拉取當(dāng)前 job 里每個(gè) maptask 的最終結(jié)果，然后對(duì)從不同地方拉取過來的數(shù)據(jù)不斷地做 merge ，也最終形成一個(gè)文件作為 reduce task 的輸入文件。

2、copy

Reduce進(jìn)程啟動(dòng)一些數(shù)據(jù)copy線程(Fetcher)，通過HTTP方式請(qǐng)求map task所在的TaskTracker獲取map task的輸出文件。因?yàn)閙aptask早已結(jié)束，這些文件就歸TaskTracker管理在本地磁盤中。

3、merge

Copy 過來的數(shù)據(jù)會(huì)先放入內(nèi)存緩沖區(qū)中，這里的緩沖區(qū)大小要比 map 端的更為靈活，它基于 JVM 的 heap size 設(shè)置，因?yàn)?Shuffle 階段 Reducer 不運(yùn)行，所以應(yīng)該把絕大部分的內(nèi)存都給 Shuffle 用。這里需要強(qiáng)調(diào)的是，merge 有三種形式：1)內(nèi)存到內(nèi)存 2)內(nèi)存到磁盤 3)磁盤到磁盤。默認(rèn)情況下***種形式不啟用，讓人比較困惑，是吧。當(dāng)內(nèi)存中的數(shù)據(jù)量到達(dá)一定閾值，就啟動(dòng)內(nèi)存到磁盤的 merge 。與 map 端類似，這也是溢寫的過程，這個(gè)過程中如果你設(shè)置有Combiner，也是會(huì)啟用的，然后在磁盤中生成了眾多的溢寫文件。第二種merge方式一直在運(yùn)行，直到?jīng)]有 map 端的數(shù)據(jù)時(shí)才結(jié)束，然后啟動(dòng)第三種磁盤到磁盤的 merge 方式生成最終的那個(gè)文件。

4、reducer的輸入

merge 的***會(huì)生成一個(gè)文件，大多數(shù)情況下存在于磁盤中，但是需要將其放入內(nèi)存中。當(dāng)reducer 輸入文件已定，整個(gè) Shuffle 階段才算結(jié)束。然后就是 Reducer 執(zhí)行，把結(jié)果放到 HDFS 上。

C、YARN

YARN(Yet Another Resource Negotiator),下一代MapReduce框架的名稱，為了容易記憶，一般稱為MRv2(MapReduce version 2)。該框架已經(jīng)不再是一個(gè)傳統(tǒng)的MapReduce框架，甚至與MapReduce無關(guān)，她是一個(gè)通用的運(yùn)行時(shí)框架，用戶可以編寫自己的計(jì)算框架，在該運(yùn)行環(huán)境中運(yùn)行。用于自己編寫的框架作為客戶端的一個(gè)lib，在運(yùn)用提交作業(yè)時(shí)打包即可。

五、why YARN instead of MR

MR 的缺點(diǎn)

經(jīng)典 MapReduce 的最嚴(yán)重的限制主要關(guān)系到可伸縮性、資源利用和對(duì)與 MapReduce 不同的工作負(fù)載的支持。在 MapReduce 框架中，作業(yè)執(zhí)行受兩種類型的進(jìn)程控制：

• 一個(gè)稱為 JobTracker 的主要進(jìn)程，它協(xié)調(diào)在集群上運(yùn)行的所有作業(yè)，分配要在 TaskTracker 上運(yùn)行的 map 和 reduce 任務(wù)。
• 許多稱為 TaskTracker 的下級(jí)進(jìn)程，它們運(yùn)行分配的任務(wù)并定期向 JobTracker 報(bào)告進(jìn)度。
• 大型的 Hadoop 集群顯現(xiàn)出了由單個(gè) JobTracker 導(dǎo)致的可伸縮性瓶頸。

此外，較小和較大的 Hadoop 集群都從未***效地使用他們的計(jì)算資源。在 Hadoop MapReduce 中，每個(gè)從屬節(jié)點(diǎn)上的計(jì)算資源由集群管理員分解為固定數(shù)量的 map 和 reduce slot，這些 slot 不可替代。設(shè)定 map slot 和 reduce slot 的數(shù)量后，節(jié)點(diǎn)在任何時(shí)刻都不能運(yùn)行比 map slot 更多的 map 任務(wù)，即使沒有 reduce 任務(wù)在運(yùn)行。這影響了集群的利用率，因?yàn)樵谒? map slot 都被使用(而且我們還需要更多)時(shí)，我們無法使用任何 reduce slot，即使它們可用，反之亦然。

Hadoop 設(shè)計(jì)為僅運(yùn)行 MapReduce 作業(yè)。隨著替代性的編程模型(比如 Apache Giraph 所提供的圖形處理)的到來，除 MapReduce 外，越來越需要為可通過高效的、公平的方式在同一個(gè)集群上運(yùn)行并共享資源的其他編程模型提供支持。

• 原MapReduce框架的不足
• JobTracker是集群事務(wù)的集中處理點(diǎn)，存在單點(diǎn)故障
• JobTracker需要完成的任務(wù)太多，既要維護(hù)job的狀態(tài)又要維護(hù)job的task的狀態(tài)，造成過多的資源消耗
• 在taskTracker端，用map/reduce task作為資源的表示過于簡(jiǎn)單，沒有考慮到CPU、內(nèi)存等資源情況，當(dāng)把兩個(gè)需要消耗大內(nèi)存的task調(diào)度到一起，很容易出現(xiàn)OOM
• 把資源強(qiáng)制劃分為map/reduce slot,當(dāng)只有map task時(shí)，reduce slot不能用;當(dāng)只有reduce task時(shí)，map slot不能用，容易造成資源利用不足。
• 解決可伸縮性問題

在 Hadoop MapReduce 中，JobTracker 具有兩種不同的職責(zé)：

管理集群中的計(jì)算資源，這涉及到維護(hù)活動(dòng)節(jié)點(diǎn)列表、可用和占用的 map 和 reduce slots 列表，以及依據(jù)所選的調(diào)度策略將可用 slots 分配給合適的作業(yè)和任務(wù)

協(xié)調(diào)在集群上運(yùn)行的所有任務(wù)，這涉及到指導(dǎo) TaskTracker 啟動(dòng) map 和 reduce 任務(wù)，監(jiān)視任務(wù)的執(zhí)行，重新啟動(dòng)失敗的任務(wù)，推測(cè)性地運(yùn)行緩慢的任務(wù)，計(jì)算作業(yè)計(jì)數(shù)器值的總和，等等

為單個(gè)進(jìn)程安排大量職責(zé)會(huì)導(dǎo)致重大的可伸縮性問題，尤其是在較大的集群上，JobTracker 必須不斷跟蹤數(shù)千個(gè) TaskTracker、數(shù)百個(gè)作業(yè)，以及數(shù)萬個(gè) map 和 reduce 任務(wù)。相反，TaskTracker 通常近運(yùn)行十來個(gè)任務(wù)，這些任務(wù)由勤勉的 JobTracker 分配給它們。

為了解決可伸縮性問題，一個(gè)簡(jiǎn)單而又絕妙的想法應(yīng)運(yùn)而生：我們減少了單個(gè) JobTracker 的職責(zé)，將部分職責(zé)委派給 TaskTracker，因?yàn)榧褐杏性S多 TaskTracker。在新設(shè)計(jì)中，這個(gè)概念通過將 JobTracker 的雙重職責(zé)(集群資源管理和任務(wù)協(xié)調(diào))分開為兩種不同類型的進(jìn)程來反映。

六、YARN 的優(yōu)點(diǎn)

• 更快地MapReduce計(jì)算
• 對(duì)多框架支持
• 框架升級(jí)更容易

YARN

• ResourceManager 代替集群管理器
• ApplicationMaster 代替一個(gè)專用且短暫的 JobTracker
• NodeManager 代替 TaskTracker
• 一個(gè)分布式應(yīng)用程序代替一個(gè) MapReduce 作業(yè)

一個(gè)全局 ResourceManager 以主要后臺(tái)進(jìn)程的形式運(yùn)行，它通常在專用機(jī)器上運(yùn)行，在各種競(jìng)爭(zhēng)的應(yīng)用程序之間仲裁可用的集群資源。

在用戶提交一個(gè)應(yīng)用程序時(shí)，一個(gè)稱為 ApplicationMaster 的輕量型進(jìn)程實(shí)例會(huì)啟動(dòng)來協(xié)調(diào)應(yīng)用程序內(nèi)的所有任務(wù)的執(zhí)行。這包括監(jiān)視任務(wù)，重新啟動(dòng)失敗的任務(wù)，推測(cè)性地運(yùn)行緩慢的任務(wù)，以及計(jì)算應(yīng)用程序計(jì)數(shù)器值的總和。有趣的是，ApplicationMaster 可在容器內(nèi)運(yùn)行任何類型的任務(wù)。

NodeManager 是 TaskTracker 的一種更加普通和高效的版本。沒有固定數(shù)量的 map 和 reduce slots，NodeManager 擁有許多動(dòng)態(tài)創(chuàng)建的資源容器。