8、Yarn資源調度系統架構與原理分析

@[TOC]

1、Yarn介紹

   Apache Hadoop YARN 是 apache Software Foundation Hadoop的子項目，為分離Hadoop2.0資源管理和計算組件而引入。YARN的誕生緣于存儲于HDFS的數據需要更多的交互模式，不單單是MapReduce模式。Hadoop2.0 的YARN 架構提供了更多的處理框架，比如spark框架，不再強迫使用MapReduce框架。
   從hadoop2.0 的架構圖可以看出，YARN承擔著原本由MapReduce承擔的資源管理的功能，同時將這部分的功能打包使得他們可以被新的數據處理引擎使用。這也同時簡化了MapReduce的流程，使得MapReduce專注的將數據處理做到最好。使用YARN,可以用共同的資源管理，在Hadoop上跑很多應用程序。目前，很多機構已經開發基于YARN的應用程序。

2、Yarn架構

   YARN的架構還是經典的主從（master/slave）結構，如下圖所示。大體上看，YARN服務由一個ResourceManager（RM）和多個NodeManager（NM）構成，ResourceManager為主節點（master），NodeManager為從節點（slave）

   在YARN體系結構中，全局ResourceManager作為主守護程序運行，該仲裁程序在各種競爭應用程序之間仲裁可用的群集資源。ResourceManager跟蹤群集上可用的活動節點和資源的數量，并協調用戶提交的應用程序應獲取這些資源的時間和時間。ResourceManager是具有此信息的單個進程，因此它可以以共享，安全和多租戶的方式進行分配（或者更確切地說，調度）決策（例如，根據應用程序優先級，隊列容量，ACL，數據位置等）。
   當用戶提交應用程序時，將啟動名為ApplicationMaster的輕量級進程實例，以協調應用程序中所有任務的執行。這包括監視任務，重新啟動失敗的任務，推測性地運行慢速任務以及計算應用程序計數器的總值。這些職責先前已分配給所有工作的單個JobTracker。ApplicationMaster和屬于其應用程序的任務在NodeManagers控制的資源容器中運行。
   ApplicationMaster可以在容器內運行任何類型的任務。例如，MapReduce ApplicationMaster請求容器啟動map或reduce任務，而Giraph ApplicationMaster請求容器運行Giraph任務。您還可以實現運行特定任務的自定義ApplicationMaster，并以此方式創建一個閃亮的新分布式應用程序框架，該框架可以更改大數據世界。我鼓勵您閱讀Apache Twill，它旨在簡化編寫位于YARN之上的分布式應用程序。
   一個可以運行任何分布式應用程序的集群 ResourceManager，NodeManager和容器不關心應用程序或任務的類型。所有特定于應用程序框架的代碼都被簡單地移動到其ApplicationMaster，以便YARN可以支持任何分布式框架，只要有人為它實現適當的ApplicationMaster。由于這種通用方法，運行許多不同工作負載的Hadoop YARN集群的夢想成真。想象一下：數據中心內的單個Hadoop集群可以運行MapReduce，Giraph，Storm，Spark，Tez / Impala，MPI等。

2.1 、ResourceManager

   RM是一個全局的資源管理器，集群只有一個，負責整個系統的資源管理和分配，包括處理客戶端請求、啟動/監控ApplicationMaster、監控 NodeManager、資源的分配與調度。它主要由兩個組件構成：調度器（Scheduler）和應用程序管理器（Applications Manager，ASM）。
   (1)調度器
   調度器根據容量、隊列等限制條件（如每個隊列分配一定的資源，最多執行一定數量的作業等），將系統中的資源分配給各個正在運行的應用程序。需要注意的是，該調度器是一個“純調度器”，它從事任何與具體應用程序相關的工作，比如不負責監控或者跟蹤應用的執行狀態等，也不負責重新啟動因應用執行失敗或者硬件故障而產生的失敗任務，這些均交由應用程序相關的ApplicationMaster完成。
   調度器僅根據各個應用程序的資源需求進行資源分配，而資源分配單位用一個抽象概念“資源容器”（ResourceContainer，簡稱Container）表示，Container是一個動態資源分配單位，它將內存、CPU、磁盤、網絡等資源封裝在一起，從而限定每個任務使用的資源量。
   (2)應用程序管理器
應用程序管理器主要負責管理整個系統中所有應用程序，接收job的提交請求，為應用分配第一個 Container 來運
行 ApplicationMaster，包括應用程序提交、與調度器協商資源以啟動 ApplicationMaster、監控
ApplicationMaster 運行狀態并在失敗時重新啟動它等。

2.2 、ApplicationMaster

   管理 YARN 內運行的一個應用程序的每個實例。關于 job 或應用的管理都是由 ApplicationMaster 進程負責的，Yarn 允許我們以為自己的應用開發 ApplicationMaster。
   功能：
   (1)數據切分；
   (2)為應用程序申請資源并進一步分配給內部任務（TASK）；
   (3)任務監控與容錯；
   負責協調來自ResourceManager的資源，并通過NodeManager監視容易的執行和資源使用情況。Yarn 的動態性，就是來源于多個Application 的ApplicationMaster 動態地和 ResourceManager 進行溝通，不斷地申請、釋放、再申請、再釋放資源的過程。

2.3 、NodeManager

   NodeManager 整個集群有多個，負責每個節點上的資源和使用。
   NodeManager 是一個 slave 服務：它負責接收 ResourceManager 的資源分配請求，分配具體的 Container 給應用。同時，它還負責監控并報告 Container 使用信息給 ResourceManager。通過和ResourceManager 配合，NodeManager 負責整個 Hadoop 集群中的資源分配工作。
   功能:本節點上的資源使用情況和各個 Container 的運行狀態（cpu和內存等資源）
   (1)接收及處理來自 ResourceManager 的命令請求，分配 Container 給應用的某個任務；
   (2)定時地向RM匯報以確保整個集群平穩運行，RM 通過收集每個 NodeManager 的報告信息來追蹤整個集群健康狀態的，而 NodeManager 負責監控自身的健康狀態；
   (3)處理來自 ApplicationMaster 的請求；
   (4)管理著所在節點每個 Container 的生命周期；
   (5)管理每個節點上的日志；
   (6)執行 Yarn 上面應用的一些額外的服務，比如 MapReduce 的 shuffle 過程；
   當一個節點啟動時，它會向 ResourceManager 進行注冊并告知 ResourceManager 自己有多少資源可用。在運行期，通過 NodeManager 和 ResourceManager 協同工作，這些信息會不斷被更新并保障整個集群發揮出最佳狀態。NodeManager 只負責管理自身的 Container，它并不知道運行在它上面應用的信息。負責管理應用信息的組件是ApplicationMaster

2.4 、Container

   Container 是 YARN 中的資源抽象，它封裝了某個節點上的多維度資源，如內存、CPU、磁盤、網絡等，當 AM 向RM 申請資源時，RM 為 AM 返回的資源便是用 Container 表示的。YARN 會為每個任務分配一個 Container，且任務只能使用該 Container 中描述的資源。
   Container 和集群節點的關系是：一個節點會運行多個 Container，但一個 Container 不會跨節點。任何一個 job或 application 必須運行在一個或多個 Container 中，在 Yarn 框架中，ResourceManager 只負責告訴ApplicationMaster 哪些 Containers 可以用，ApplicationMaster 還需要去找 NodeManager 請求分配具體的Container。
   需要注意的是，Container 是一個動態資源劃分單位，是根據應用程序的需求動態生成的。目前為止，YARN 僅支持 CPU 和內存兩種資源，且使用了輕量級資源隔離機制 Cgroups 進行資源隔離。

2.5 、Resource Request 及 Container

   Yarn的設計目標就是允許我們的各種應用以共享、安全、多租戶的形式使用整個集群。并且，為了保證集群資源調度和數據訪問的高效性，Yarn還必須能夠感知整個集群拓撲結構。
   為了實現這些目標，ResourceManager的調度器Scheduler為應用程序的資源請求定義了一些靈活的協議，通過它就可以對運行在集群中的各個應用做更好的調度，因此，這就誕生了Resource Request和Container。一個應用先向ApplicationMaster發送一個滿足自己需求的資源請求，然后ApplicationMaster把這個資源請求以
   resource-request的形式發送給ResourceManager的Scheduler，Scheduler再在這個原始的resource-request中返回分配到的資源描述Container。每個ResourceRequest可看做一個可序列化Java對象，包含的字段信息如下：

<!--
- resource-name：資源名稱，現階段指的是資源所在的host和rack，后期可能還會支持虛擬機或者更復雜的網絡結構
- priority：資源的優先級
- resource-requirement：資源的具體需求，現階段指內存和cpu需求的數量
- number-of-containers：滿足需求的Container的集合
-->
<resource-name, priority, resource-requirement, number-of-containers>

2.6 、JobHistoryServer

   作業歷史服務,記錄在yarn中調度的作業歷史運行情況情況 , 通過mr-jobhistory-daemon.sh start historyserver命令在集群中的數據節點機器上不需要做任何配置,單獨使用命令啟動直接啟動即可, 啟動成功后會出現JobHistoryServer進程(使用jps命令查看，下面會有介紹) , 并且可以從19888端口進行查看日志詳細信息

打開如下圖界面，在下圖中點擊History，頁面會進行一次跳轉

點擊History之后 跳轉后的頁面如下圖,是空白的，這時因為我們沒有啟動jobhistoryserver所導致的。 在三臺機器上執行mr-jobhistory-daemon.sh start historyserver命令依次啟動jobhistoryserver。在node1節點啟動

此時我們在三個節點把jobhistoryserver啟動后，在此運行wordcount程序(記得啟動前把輸出目錄刪除掉)
點擊History連接會跳轉一個贊新的頁面，在頁面下方會看到TaskType中列舉的map和reduce，Total表示此次運行的mapreduce程序執行所需要的map和reduce的任務數據.

2.7、Timeline Server

   用來寫日志服務數據 , 一般來寫與第三方結合的日志服務數據(比如spark等),從官網的介紹看，它是對jobhistoryserver功能的有效補充，jobhistoryserver只能對mapreduce類型的作業信息進行記錄，除了jobhistoryserver能夠進行對作業運行過程中信息進行記錄之外還有更細粒度的信息記錄，比如任務在哪個隊列中運行，運行任務時設置的用戶是哪個用戶。根據官網的解釋jobhistoryserver只能記錄mapreduce應用程序的記錄，timelineserver功能更強大,但不是替代jobhistory兩者是功能間的互補關系。

3、yarn應用運行原理

   YARN 是如何工作的？ YARN的基本理念是將JobTracker/TaskTracker 兩大職能分割為以下幾個實體：
1.一個全局的資源管理ResourceManager

2. 每個應用程序一個ApplicationMaster
3. 每個從節點一個NodeManager
4. 每個應用程序一個運行在NodeManager上的Container
      ResouceManager 和 NodeManager 組成了一個新的、通用的、用分布式管理應用程序的系統。ResourceManager 對系統中的應用程序資源有終極仲裁的權限。ApplicationMaster 是一個特定于框架的實體，它的責任是同ResourceManager 談判資源 ，同時為NodeManager(s)執行和監控組件任務。RessourceManager有一個調度器，根據不同的約束條件，例如隊列容量、用戶限制等，將資源進行分配給各類運行著的應用程序。調度器執行調度功能是基于應用程序的資源申請。NodeManager 負責發布應用程序容器，監控資源的使用并向ResourceManager進行匯報。每個ApplicationMaster都有職責從調度器那談判得到適當的資源容器，追蹤它們的狀態，并監控他們的進程。從系統的視圖看，ApplicationMaster 作為一個普通的容器運行著。

3.1、yarn應用提交過程

   Application在Yarn中的執行過程，整個執行過程可以總結為三步：
   (1)應用程序提交
   (2)啟動應用的ApplicationMaster實例
   (3)ApplicationMaster 實例管理應用程序的執行
具體過程：
   (1)客戶端程序向 ResourceManager 提交應用并請求一個 ApplicationMaster 實例；
(2)ResourceManager 找到一個可以運行一個 Container 的 NodeManager，并在這個 Container 中啟動ApplicationMaster 實例；
   (3)ApplicationMaster 向 ResourceManager 進行注冊，注冊之后客戶端就可以查詢 ResourceManager 獲得自己 ApplicationMaster 的詳細信息，以后就可以和自己的 ApplicationMaster 直接交互了（這個時候，客戶端主動和 ApplicationMaster 交流，應用先向 ApplicationMaster 發送一個滿足自己需求的資源請求）；
   (4)在平常的操作過程中，ApplicationMaster 根據 resource-request協議 向 ResourceManager 發送 resource-request請求；
   (5)當 Container 被成功分配后，ApplicationMaster 通過向 NodeManager 發送 container-launch-specification信息 來啟動Container，container-launch-specification信息包含了能夠讓Container 和ApplicationMaster 交流所需要的資料；
   (6)應用程序的代碼以 task 形式在啟動的 Container 中運行，并把運行的進度、狀態等信息通過 application-specific協議 發送給ApplicationMaster；
   (7)在應用程序運行期間，提交應用的客戶端主動和 ApplicationMaster 交流獲得應用的運行狀態、進度更新等信息，交流協議也是 application-specific協議；
   (8)一旦應用程序執行完成并且所有相關工作也已經完成，ApplicationMaster 向 ResourceManager 取消注冊然后關閉，用到所有的 Container 也歸還給系統。

3.2、mapreduce on yarn

   (1)客戶端提交hadoop jar
   (2)找到main()方法中的job.waitForCompletition生成job對象，運行job對象的runjob()方法，與ResourceManager通信，
返回ResourceManager分配一個ID號（applicationid），需不需要輸出，如果需要輸出，判斷輸出是否存在，如果不存在問題，在看輸入，根據hdfs得到輸入得到分片信息,根據分片信息得到map個數，將這些信息回傳給客戶端Job
   (3)把Job所需要的資源，例如Jar包，配置文件，split分片信息，上傳到hdfs中去
   (4)Job對象告知與ResourceManager提交應用
   (5)ResourceManager尋找合適的節點開啟container容器(本質上是一個JVM虛擬機)
   (6)在container中啟動一個applicationMaster,初始化Job，生成一個薄記對象(就是一個記事本)，記錄map、reduce的狀態。把job所有資源上傳到hdfs中，包括jar包，配置文件，split信息
   (7)applicationMaster向ResourceManager申請資源，返回資源信息(包括node節點地址，cpu信息，內存占比，IO信息)
   (8)applicationMaster收到信息之后，和NodeManager通信，傳遞資源信息
   (9)開啟YarnChild進程,從hdfs中獲得Job詳細信息(包括Jar包,配置文件信息)

   (一)Job初始化：1、當resourceManager收到了submitApplication()方法的調用通知后，scheduler開始分配container,隨之ResouceManager發送applicationMaster進程，告知每個nodeManager管理器。 2、由applicationMaster決定如何運行tasks,如果job數據量比較小，applicationMaster便選擇將tasks運行在一個JVM中。那么如何判別這個job是大是小呢？當一個job的mappers數量小于10個，只有一個reducer或者讀取的文件大小要小于一個HDFS block時，（可通過修改配置項mapreduce.job.ubertask.maxmaps,mapreduce.job.ubertask.maxreduces以及mapreduce.job.ubertask.maxbytes 進行調整) 3、在運行tasks之前，applicationMaster將會調用setupJob()方法，隨之創建output的輸出路徑(這就能夠解釋，不管你的mapreduce一開始是否報錯，輸出路徑都會創建)。
   (二)Task 任務分配：1、接下來applicationMaster向ResourceManager請求containers用于執行map與reduce的tasks（step 8),這里map task的優先級要高于reduce task，當所有的map tasks結束后，隨之進行sort(這里是shuffle過程后面再說）,最后進行reduce task的開始。(這里有一點，當map tasks執行了百分之5%的時候，將會請求reduce，具體下面再總結) 2、運行tasks的是需要消耗內存與CPU資源的，默認情況下，map和reduce的task資源分配為1024MB與一個核，（可修改運行的最小與最大參數配置,mapreduce.map.memory.mb,mapreduce.reduce.memory.mb,mapreduce.map.cpu.vcores,mapreduce.reduce.reduce.cpu.vcores.)
   (三)運行進度與狀態更新 1、MapReduce是一個較長運行時間的批處理過程，可以是一小時、
幾小時甚至幾天，那么Job的運行狀態監控就非常重要。每個job以及每個task都有一個包含
job（running,successfully completed,failed）的狀態，以及value的計數器，狀態信息及描述信息（描述信息一般都是在代碼中加的打印信息），那么，這些信息是如何與客戶端進行通信的呢？ 2、當一個task開始執行，它將會保持運行記錄，記錄task完成的比例，對于map的任務，將會記錄其運行的百分比，對于reduce來說可能復雜點，但系統依舊會估計reduce的完成比例。當一個map或reduce任務執行時，子進程會持續每三秒鐘與applicationMaster進行交互。

4、 yarn使用

4.1 、配置文件

<!-- $HADOOP_HOME/etc/hadoop/mapred-site.xml -->
<configuration>
<property>
<name>mapreduce.framework.name</name>
<value>yarn</value>
</property>
</configuration>
<!-- $HADOOP_HOME/etc/hadoop/yarn-site.xml -->
<configuration>
<property>
<name>yarn.nodemanager.aux-services</name>
<value>mapreduce_shuffle</value>
</property>
</configuration>

4.2、 yarn啟動停止

啟動 ResourceManager 和 NodeManager （以下分別簡稱RM、NM）

#主節點運行命令
$HADOOP_HOME/sbin/start-yarn.sh
#主節點運行命令
$HADOOP_HOME/sbin/stop-yarn.sh
若RM沒有啟動起來，可以單獨啟動

若RM沒有啟動起來，可以單獨啟動

#若RM沒有啟動，在主節點運行命令
$HADOOP_HOME/sbin/yarn-daemon.sh start resouremanager
#相反，可單獨關閉
$HADOOP_HOME/sbin/yarn-daemon.sh stop resouremanager

若NM沒有啟動起來，可以單獨啟動

#若NM沒有啟動，在相應節點運行命令
$HADOOP_HOME/sbin/yarn-daemon.sh start nodemanager
#相反，可單獨關閉
$HADOOP_HOME/sbin/yarn-daemon.sh stop nodemanager

4.3、 yarn常用命令

#1.查看正在運行的任務
yarn application -list
#2.殺掉正在運行任務
yarn application -kill 任務id
#3.查看節點列表
yarn node -list
#4.查看節點狀況 TODO
yarn node -status node3:40652
#5.查看yarn依賴jar的環境變量
yarn classpath

5、Yarn調度器

在Yarn中有三種調度器可以選擇：FIFO Scheduler ，Capacity Scheduler，FairS cheduler

5.1、 FIFO Scheduler

   FIFO Scheduler把應用按提交的順序排成一個隊列，這是一個先進先出隊列，在進行資源分配的時候，先給隊列中最頭上的應用進行分配資源，待最頭上的應用需求滿足后再給下一個分配，以此類推。FIFO Scheduler是最簡單也是最容易理解的調度器，也不需要任何配置，但它并不適用于共享集群。大的應用可能會占用所有集群資源，這就導致其它應用被阻塞。在共享集群中，更適合采用Capacity Scheduler或FairScheduler，這兩個調度器都允許大任務和小任務在提交的同時獲得一定的系統資源。下面“Yarn調度器對比圖”展示了這幾個調度器的區別，從圖中可以看出，在FIFO 調度器中

5.2、 Capacity Scheduler

而對于Capacity調度器，有一個專門的隊列用來運行小任務，但是為小任務專門設置一個隊列會預先占用一定的集
群資源，這就導致大任務的執行時間會落后于使用FIFO調度器時的時間。
如何配置容量調度器
隊列層級結構如下：
root
├── prod
└── dev
     ├── spark
     └── hdp
HADOOP_HOME/etc/hadoop/中建立一個新的capacity-scheduler.xml；內容如下：

<?xml version="1.0" encoding="utf-8"?>
<configuration>
   <property>
      <name>yarn.scheduler.capacity.root.queues</name>
      <value>prod,dev</value>
   </property>
   <property>
       <name>yarn.scheduler.capacity.root.dev.queues</name>
       <value>hdp,spark</value>
   </property>
   <property>
       <name>yarn.scheduler.capacity.root.prod.capacity</name>
       <value>40</value>
   </property>
   <property>
      <name>yarn.scheduler.capacity.root.dev.capacity</name>
      <value>60</value>
  </property>
  <property>
      <name>yarn.scheduler.capacity.root.dev.maximum-capacity</name>
      <value>75</value>
  </property>
  <property>
   <name>yarn.scheduler.capacity.root.dev.hdp.capacity</name>
     <value>50</value>
     </property>
   <property>
      <name>yarn.scheduler.capacity.root.dev.spark.capacity</name>
      <value>50</value>
    </property>
  </configuration>

將應用放置在哪個隊列中，取決于應用本身。
例如MR，可以通過設置屬性mapreduce.job.queuename指定相應隊列。以WordCount為例，如下，如果指定的隊列不存在，則發生錯誤。如果不指定，默認使用"default"隊列，如下圖

5.3、 Fair Scheduler

   在Fair調度器中，我們不需要預先占用一定的系統資源，Fair調度器會為所有運行的job動態的調整系統資源。當第一個大job提交時，只有這一個job在運行，此時它獲得了所有集群資源；當第二個小任務提交后，Fair調度器會分配一半資源給這個小任務，讓這兩個任務公平的共享集群資源。需要注意的是，在下圖Fair調度器中，從第二個任務提交到獲得資源會有一定的延遲，因為它需要等待第一個任務釋放占用的Container。小任務執行完成之后也會釋放自己占用的資源，大任務又獲得了全部的系統資源。最終的效果就是Fair調度器即得到了高的資源利用率又能保證小任務及時完成.