Map Reduce和YARN技術原理
學習目標
- 熟悉MapReduce和YARN是什么
- 掌握MapReduce使用的場景及其原理
- 掌握MapReduce和YARN功能與架構
- 熟悉YARN的新特性
MapReduce的概述
MapReduce基於Google發布的MapReduce論文設計開發，用於大規模數據集（大於1TB）的並行計算
具有如下特點：
- 易於編程：程序員僅需描述做什么，具體怎么做交由系統的執行框架處理。
- 良好的擴展性：可通過添加節點以擴展集群能力。
- 高容錯性：通過計算遷移或數據遷移等策略提高集群的可用性與容錯性。
當某些節點發生故障時，可以通過計算遷移或數據遷移在其他節點繼續執行任務，保證任務執行成功。
- MapReduce是一個基於集群的高性能並行計算平台（Cluster Infrastructure）。
- MapReduce是一個並行計算與運行軟件框架（Software Framework）。
- MapReduce是一個並行程序設計模型與方法（Programming Model & Methodology）。
YARN的概述
Apache Hadoop YARN (Yet Another Resource Negotiator，另一種資源協調者) 是一種新的 Hadoop 資源管理器，它是一個通用資源管理系統，可為上層應用提供統一的資源管理和調度，它的引入為集群在利用率、資源統一管理和數據共享等方面帶來了巨大好處。
MapReduce過程詳解
- 分片的必要性：MR框架將一個分片和一個Map TasK對應，即一個Map Task只負責處理一個數據分片。數據分片的數量確定了為這個Job創建Map Task的個數。
- Application Master(AM)負責一個Application生命周期內的所有工作。包括：與RM調度器協商以獲取資源；將得到的資源進一步分配給內部任務（資源的二次分配）；與NM通信以啟動/停止任務；監控所有任務運行狀態，並在任務運行失敗時重新為任務申請資源以重啟任務。
- ResourceManager(RM) 負責集群中所有資源的統一管理和分配。它接收來自各個節點（NodeManager）的資源匯報信息，並根據收集的資源按照一定的策略分配給各個應用程序。
- NodeManager（NM）是每個節點上的代理，它管理Hadoop集群中單個計算節點，包括與ResourceManger保持通信，監督Container的生命周期管理，監控每個Container的資源使用（內存、CPU等）情況，追蹤節點健康狀況，管理日志和不同應用程序用到的附屬服務（auxiliary service）。
- MR組件在FI中只有jobhistoryserver實例，它只是存儲任務的執行記錄，執行列表，沒有它，也可以運行任務。但是無法查詢任務的詳細信息。
Reduce階段的三個過程：
- Copy：Reduce Task從各個Map Task拷貝MOF文件。
- Sort：通常又叫Merge，將多個MOF文件進行合並再排序。
- Reduce：用戶自定義的Reduce邏輯。
Shuffle機制
Shuffle的定義：Map階段和Reduce階段之間傳遞中間數據的過程，包括Reduce Task從各個Map Task獲取MOF文件的過程，以及對MOF的排序與合並處理。
典型程序WorldCound舉例
假設要分析一個大文件A里每個英文單詞出現的個數，利用MapReduce框架能快速實現這一統計分析。
- 第一步：待處理的大文件A已經存放在HDFS上，大文件A被切分的數據塊A.1、A.2、A.3分別存放在Data Node #1、#2、#3上。
- 第二步：WordCount分析處理程序實現了用戶自定義的Map函數和Reduce函數。WordCount將分析應用提交給RM，RM根據請求創建對應的Job，並根據文件塊個數按文件塊分片，創建3個 MapTask 和 3個Reduce Task，這些Task運行在Container中。
- 第三步：Map Task 1、2、3的輸出是一個經分區與排序（假設沒做Combine）的MOF文件，記錄形如表所示。
- 第四步：Reduce Task從 Map Task獲取MOF文件，經過合並、排序，最后根據用戶自定義的Reduce邏輯，輸出如表所示的統計結果。
WorldCound程序功能
WorldCound的Map過程
WorldCound的Reduce過程
YARN的組件架構
在圖中有兩個客戶端向YARN提交任務，藍色表示一個任務流程，棕色表示另一個任務流程。
- 首先client提交任務，ResourceManager接收到任務，然后啟動並監控起來的第一個Container,也就是App Mstr。
- App Mstr通知nodemanager管理資源並啟動其他container。
- 任務最終是運行在Container當中。
MapReduce On YARN任務調度流程
- 步驟1：用戶向YARN 中提交應用程序， 其中包括ApplicationMaster 程序、啟動ApplicationMaster 的命令、用戶程序等。
- 步驟2：ResourceManager 為該應用程序分配第一個Container， 並與對應的NodeManager 通信，要求它在這個Container 中啟動應用程序的ApplicationMaster 。
- 步驟3：ApplicationMaster 首先向ResourceManager 注冊， 這樣用戶可以直接通過ResourceManage 查看應用程序的運行狀態，然后它將為各個任務申請資源，並監控它的運行狀態，直到運行結束，即重復步驟4~7。
- 步驟4：ApplicationMaster 采用輪詢的方式通過RPC 協議向ResourceManager 申請和領取資源。
- 步驟5：一旦ApplicationMaster 申請到資源后，便與對應的NodeManager 通信，要求它啟動任務。
- 步驟6：NodeManager 為任務設置好運行環境（包括環境變量、JAR 包、二進制程序等）后，將任務啟動命令寫到一個腳本中，並通過運行該腳本啟動任務。
- 步驟7：各個任務通過某個RPC 協議向ApplicationMaster 匯報自己的狀態和進度，以讓ApplicationMaster 隨時掌握各個任務的運行狀態，從而可以在任務失敗時重新啟動任務。在應用程序運行過程中，用戶可隨時通過RPC 向ApplicationMaster 查詢應用程序的當前運行狀態。
- 步驟8 應用程序運行完成后，ApplicationMaster 向ResourceManager 注銷並關閉自己。
YARN HA方案
YARN中的ResourceManager負責整個集群的資源管理和任務調度，YARN高可用性方案通過引入冗余的ResourceManager節點的方式，解決了ResourceManager 單點故障問題。
- ResourceManager的高可用性方案是通過設置一組Active/Standby的ResourceManager節點來實現的。與HDFS的高可用性方案類似，任何時間點上都只能有一個ResourceManager處於Active狀態。當Active狀態的ResourceManager發生故障時，可通過自動或手動的方式觸發故障轉移，進行Active/Standby狀態切換。
- 在未開啟自動故障轉移時，YARN集群啟動后，管理員需要在命令行中使用YARN rmadmin命令手動將其中一個ResourceManager切換為Active狀態。當需要執行計划性維護或故障發生時，則需要先手動將Active狀態的ResourceManager切換為Standby狀態，再將另一個ResourceManager切換為Active狀態。
- 開啟自動故障轉移后，ResourceManager會通過內置的基於ZooKeeper實現的ActiveStandbyElector來決定哪一個ResouceManager應該成為Active節點。當Active狀態的ResourceManager發生故障時，另一個ResourceManager將自動被選舉為Active狀態以接替故障節點。
- 當集群的ResourceManager以HA方式部署時，客戶端使用的“YARN-site.xml”需要配置所有ResourceManager地址。客戶端（包括ApplicationMaster和NodeManager）會以輪詢的方式尋找Active狀態的ResourceManager。如果當前Active狀態的ResourceManager無法連接，那么會繼續使用輪詢的方式找到新的ResourceManager。
YARN APPMaster容錯機制
- 在YARN中，ApplicationMaster(AM)與其他Container類似也運行在NodeManager上（忽略未管理的AM）。AM可能會由於多種原因崩潰、退出或關閉。如果AM停止運行，ResourceManager(RM)會關閉ApplicationAttempt中管理的所有Container，包括當前任務在NodeManager(NM)上正在運行的所有Container。RM會在另一計算節點上啟動新的ApplicationAttempt。
- 不同類型的應用希望以多種方式處理AM重新啟動的事件。MapReduce類應用目標是不丟失任務狀態，但也能允許一部分的狀態損失。但是對於長周期的服務而言，用戶並不希望僅僅由於AM的故障而導致整個服務停止運行。
- YARN支持在新的ApplicationAttempt啟動時，保留之前Container的狀態，因此運行中的作業可以繼續無故障的運行。
資源管理
當前YARN支持內存和CPU兩種資源類型的管理和分配。 每個NodeManager可分配的內存和CPU的數量可以通過配置選項設置（可在YARN服務配置頁面配置）。
- Yarn.nodemanager.resource.memory-mb
- Yarn.nodemanager.vmem-pmem-ratio
- Yarn.nodemanager.resource.cpu-vcore
- Yarn.nodemanager.resource.memory-mb表示用於當前NodeManager上可以分配給容器的物理內存的大小，單位：MB。必須小於NodeManager服務器上的實際內存大小。
- Yarn.nodemanager.vmem-pmem-ratio表示為容器設置內存限制時虛擬內存跟物理內存的比值。容器分配值使用物理內存表示的，虛擬內存使用率超過分配值的比例不允許大於當前這個比例。
- Yarn.nodemanager.resource.cpu-vcore表示可分配給container的CPU核數。建議配置為CPU核數的1.5-2倍。
資源分配模型
- 調度器維護一群隊列的信息。用戶可以向一個或者多個隊列提交應用。
- 每次NM心跳的時候，調度器根據一定的規則選擇一個隊列，再在隊列上選擇一個應用，嘗試在這個應用上分配資源。
- 調度器會優先匹配本地資源的申請請求，其次是同機架的，最后是任意機器的。
- 當任務提交上來，首先會聲明提交到哪個隊列上，調度器會分配隊列，如果沒有指定則任務運行在默認隊列。
- 隊列是封裝了集群資源容量的資源集合，占用集群的百分比例資源。
- 隊列分為父隊列，子隊列，任務最終是運行在子隊列上的。父隊列可以有多個子隊列。
- 調度器選擇隊列上的應用，然后根據一些算法給應用分配資源。
容量調度器的介紹
容量調度器：Capacity Scheduler 。
- 容量調度器使得Hadoop應用能夠共享的、多用戶的、操作簡便的運行在集群上，同時最大化集群的吞吐量和利用率。
- 容量調度器以隊列為單位划分資源，每個隊列都有資源使用的下限和上限。每個用戶可以設定資源使用上限。管理員可以約束單個隊列、用戶或作業的資源使用。支持作業優先級，但不支持資源搶占。
容量調度器的特點
- 容量保證：管理員可為每個隊列設置資源最低保證和資源使用上限，所有提交到該隊列的應用程序共享這些資源。
- 靈活性：如果一個隊列中的資源有剩余，可以暫時共享給那些需要資源的隊列，當該隊列有新的應用程序提交，則其他隊列釋放的資源會歸還給該隊列。
- 支持優先級：隊列支持任務優先級調度（默認是FIFO）。
- 多重租賃：支持多用戶共享集群和多應用程序同時運行。為防止單個應用程序、用戶或者隊列獨占集群資源，管理員可為之增加多重約束。
- 動態更新配置文件：管理員可根據需要動態修改配置參數，以實現在線集群管理。
容量調度器的任務選擇
調度時，首先按以下策略選擇一個合適隊列：
- 資源利用量最低的隊列優先，比如同級的兩個隊列Q1和Q2，它們的容量均為30，而Q1已使用10，Q2已使用12，則會優先將資源分配給Q1。
- 最小隊列層級優先，例如：QueueA與QueueB.childQueueB，則QueueA優先。
- 資源回收請求隊列優先。
然后按以下策略選擇該隊列中一個任務：
- 按照任務優先級和提交時間順序選擇，同時考慮用戶資源量限制和內存限制。
隊列資源限制（1）
隊列的創建是在多租戶頁面，當創建一個租戶關聯YARN服務時，會創建同名的隊列。比如先創建QueueA,QueueB兩個租戶，即對應YARN兩個隊列。
隊列資源限制（2）
隊列的資源容量（百分比），有default、QueueA、QueueB三個隊列，每個隊列都有一個[隊列名].capacity配置：
- Default隊列容量為整個集群資源的20%。
- QueueA隊列容量為整個集群資源的10%。
- QueueB隊列容量為整個集群資源的10%，后台有一個影子隊列root-default使隊列之和達到100% 。
- 在集群的Manager頁面點擊“租戶管理”》“動態資源計划”》“資源分布策略”可以看到，可配置各隊列資源容量。
- 影子隊列：是不對外呈現的一個隊列。以XX-default為名字，目的是為了使同級的隊列容量之和不夠一百時，將剩余容量值賦予此隊列（容量調度器要求同級隊列容量和要為100）。
隊列資源限制（3）
共享空閑資源
- 由於存在資源共享，因此一個隊列使用的資源可能超過其容量（例如QueueA.capacity），而最大資源使用量可通過參數限制。
- 如果某個隊列任務較少，可將剩余資源共享給其他隊列，例如QueueA的maximum-capacity配置為100，假設當前只有QueueA在運行任務，理論上QueueA可以占用整個集群100%的資源。
參數：Yarn.scheduler.capacity.root.QueueA.maximum-capacity
用戶限制和任務限制
用戶限制和任務限制的參數可通過“租戶管理”>“動態資源計划”>“隊列配置”進行配置。
用戶限制（1）
每個用戶最低資源保障（百分比）：
- 任何時刻，一個隊列中每個用戶可使用的資源量均有一定的限制，當一個隊列中同時運行多個用戶的任務時，每個用戶的可使用資源量在一個最小值與最大值之間浮動，其中，最大值取決於正在運行的任務數目，而最小值則由minimum-user-limit-percent決定。
- 例如，設置隊列A的這個值為25，即Yarn.scheduler.capacity.root.QueueA.minimum-user-limit-percent=25，那么隨着提任務的用戶增加，隊列資源的調整如下：
| 第1個用戶提交任務到QueueA |
會獲得QueueA的100%資源。 |
| 第2個用戶提交任務到QueueA |
每個用戶會最多獲得50%的資源。 |
| 第3個用戶提交任務到QueueA |
每個用戶會最多獲得33.33%的資源。 |
| 第4個用戶提交任務到QueueA | 每個用戶會最多獲得25%的資源。 |
| 第5個用戶提交任務到QueueA |
為了保障每個用戶最低能獲得25%的資源，第5個用戶將無法再獲取到QueueA的資源，必須等待資源的釋放。 |
用戶限制（2）
每個用戶最多可使用的資源量（所在隊列容量的倍數）：
- queue容量的倍數，用來設置一個user可以獲取更多的資源。 Yarn.scheduler.capacity.root.QueueD.user-limit-factor=1。默認值為1，表示一個user獲取的資源容量不能超過queue配置的capacity，無論集群有多少空閑資源，最多不超過maximum-capacity。
用戶可以使用超過capacity的資源,但不超過maximum-capacity。
任務限制
最大活躍任務數：
- 整個集群中允許的最大活躍任務數，包括運行或掛起狀態的所有任務，當提交的任務申請數據達到限制以后，新提交的任務將會被拒絕。默認值10000。
每個隊列最大任務數：
- 對於每個隊列，可以提交的最大任務數，以QueueA為例，可以在隊列配置頁面配置，默認是1000，即此隊列允許最多1000個活躍任務。
每個用戶可以提交的最大任務數：
- 這個數值依賴每個隊列最大任務數。根據上面的數據， QueueA最多可以提交1000個任務，那么對於每個用戶而言，可以向QueueA提交的最大任務數為1000* 用戶最低資源保障率（假設25%）* 用戶可使用隊列資源的倍數(假設1)。
- 用戶最低資源保障率：Yarn.scheduler.capacity.root.QueueA.minimum-user-limit-percent。
- 用戶可使用隊列資源的倍數：Yarn.scheduler.capacity.root.QueueA.user-limit-factor。
查看隊列信息
- 隊列的信息可以通過YARN webUI進行查看，進入方法是“服務管理”>“YARN”>“ResouceManager（主）”>“Scheduler”。
增強特性 - YARN動態內存管理
- 動態內存管理可用來優化NodeManager中Containers的內存利用率。任務在運行過程中可能產生多個Container。
- 當前，當單個節點上的Container超過Container運行內存大小時，即使節點總的配置內存利用還很低，NodeManager也會終止這些Containers。這樣就會經常使用戶作業失敗。
- 動態內存管理特性在當前是一個改進，只有當NodeManager中的所有Containers的總內存使用超過了已確定的閾值，NM總內存閾值的計算方法是
- Yarn.nodemanager.resource.memory-mb*1024*1024*Yarn.nodemanager.dynamic.memory.usage.threshold，單位GB，那么那些內存使用過多的Containers才會被終止。
- 舉例，假如某些Containers的物理內存利用率超過了配置的內存閾值，但所有Containers的總內存利用率並沒有超過設置的NodeManager內存閾值，那么那些內存使用過多的Containers仍可以繼續運行。
增強特性 - YARN基於標簽調度
- 在沒有標簽調度之前，任務提交到哪個節點上是無法控制的，會根據一些算法及條件，集群隨機分配到某些節點上。而標簽調度可以指定任務提交到哪些節點上。
- 比如之前需要消耗高內存的應用提交上來，由於運行在那些節點不可控，任務可能運行在普通性能的機器上。
- Label based scheduling是一種調度策略。該策略的基本思想是：用戶可以為每個nodemanager標注一個標簽，比如high-memory，high-IO等進行分類，以表明該nodemanager的特性；同時，用戶可以為調度器中每個隊列標注一個標簽，即隊列與標簽綁定，這樣，提交到某個隊列中的作業，只會使用標注有對應標簽的節點上的資源，即任務實際運行在打有對應標簽的節點上。
- 將耗內存消耗型的任務提交到綁定了high-memry的標簽的隊列上，那么任務就可以運行在高內存機器上。
思考題：
1 請簡述MapReduce的工作原理。
答案：
- 一個MapReduce作業(job)通常會把輸入的數據集切分為若干獨立的數據塊，由Map任務並行處理它們。框架會對map函數的輸出先進行排序，然后把結果輸入給Reduce任務。通常作業的輸入和輸出都會被存儲在文件系統中。整個框架負責任務的調度和和監控，以及重新執行已經失敗的任務。
2 請簡述YARN的工作原理。
答案：
- 用戶將應用程序提交到RM；RM為AM申請資源，與某個NM通信，啟動AM；AM與RM通信，為執行任務申請資源；得到資源后與NM通信，啟動相應的任務；所有任務結束后，AM向RM注銷，整個應用結束。
3 下面哪些是MapReduce的特點？（ ）
A 易於編程
B 良好的擴展性
C 實時計算
D 高容錯性
答案：ABD
4 YARN中資源抽象用什么表示？（ ）
A 內存
B CPU
C Container
D 磁盤空間
答案：C
5 下面哪個是MapReduce適合做的？（ ）
A 迭代計算
B 離線計算
C 實時交互計算
D 流式計算
答案：B
6 容量調度器有哪些特點？（ ）
A 容量保證
B 靈活性
C 多重租賃
D 動態更新配置文件
答案：ABCD