Yarn NodeManager解析篇

NodeManager簡(jiǎn)介

NodeManager是Yarn中單節(jié)點(diǎn)的代理摆霉，它管理Hadoop集群中單個(gè)計(jì)算節(jié)點(diǎn),他需要與應(yīng)用程序的ApplicationMaster和集群管理器RM交互漓概，從ApplicationMaster上接收到相關(guān)Container的命令(啟動(dòng)柴淘，停止Container)并執(zhí)行,向RM匯報(bào)各個(gè)Container的運(yùn)行狀態(tài)和節(jié)點(diǎn)健康狀態(tài)栋烤，并領(lǐng)取相關(guān)的Container的命令執(zhí)行;
功能包括與RM保持通信推掸，管理Container的生命周期滔岳，監(jiān)控每個(gè)Container的資源使用情況，追蹤節(jié)點(diǎn)健康狀況编丘，管理日志以及不同應(yīng)用程序用到的附屬服務(wù);

NodeManager基本職能

NodeManager通過(guò)兩個(gè)RPC協(xié)議與RM和各個(gè)ApplicationMaster進(jìn)行通信:
1.ResourceTrackerProtocol協(xié)議
NodeManager通過(guò)該RPC協(xié)議向RM注冊(cè)与学，匯報(bào)節(jié)點(diǎn)的健康狀態(tài)以及Container的運(yùn)行狀態(tài)，并領(lǐng)取RM下發(fā)的命令例如重新初始化Container嘉抓，清理Container等索守；在這個(gè)協(xié)議總NodeManager主動(dòng)向RM發(fā)請(qǐng)求，RM相應(yīng)NodeManager的請(qǐng)求抑片；該RPC協(xié)議的代碼如下:

option java_package = "org.apache.hadoop.yarn.proto";
option java_outer_classname = "ResourceTracker";
option java_generic_services = true;
option java_generate_equals_and_hash = true;
package hadoop.yarn;

import "yarn_server_common_service_protos.proto";

service ResourceTrackerService {
  //NodeManager啟動(dòng)時(shí)通過(guò)該RPC函數(shù)向RM注冊(cè),注冊(cè)信息包括:NM的http端口,NM所在host的RPC端口,該NM可分配的總資源
  rpc registerNodeManager(RegisterNodeManagerRequestProto) returns (RegisterNodeManagerResponseProto);
  //此RPC函數(shù)比較重要蕾盯，NM啟動(dòng)后定期的通過(guò)此函數(shù)向RM匯報(bào)Container的運(yùn)行信息和節(jié)點(diǎn)健康狀況，并領(lǐng)取RM的命令蓝丙，例如：kill掉一些Container
  rpc nodeHeartbeat(NodeHeartbeatRequestProto) returns (NodeHeartbeatResponseProto);
}

2.ContainerManagementProtocol協(xié)議
應(yīng)用程序的ApplicationMaster通過(guò)該協(xié)議與NodeManager通信發(fā)起針對(duì)Container的命令操作级遭，例如：?jiǎn)?dòng)，殺死Container渺尘，獲取Container的運(yùn)行狀態(tài)等;在該協(xié)議中ApplicationMaster主動(dòng)向NodeManager發(fā)送請(qǐng)求挫鸽，NodeManager接收到請(qǐng)求做出相應(yīng),該RPC協(xié)議的代碼如下:

option java_package = "org.apache.hadoop.yarn.proto";
option java_outer_classname = "ContainerManagementProtocol";
option java_generic_services = true;
option java_generate_equals_and_hash = true;
package hadoop.yarn;

import "yarn_service_protos.proto";

service ContainerManagementProtocolService {
  //AM通過(guò)該函數(shù)要求NM啟動(dòng)一個(gè)Container
  rpc startContainers(StartContainersRequestProto) returns (StartContainersResponseProto);
  //AM通過(guò)該函數(shù)要求NM殺死一個(gè)Container
  rpc stopContainers(StopContainersRequestProto) returns (StopContainersResponseProto);
  //AM通過(guò)該函數(shù)獲取Container的運(yùn)行狀態(tài)
  rpc getContainerStatuses(GetContainerStatusesRequestProto) returns (GetContainerStatusesResponseProto);
}

NodeManager內(nèi)部架構(gòu)

NodeManager的內(nèi)部架構(gòu)如下圖所示:

image

1. NodeStatusUpdater: 這個(gè)組件是NodeManager與RM通信的唯一通道，包括NM注冊(cè)之后向RM的注冊(cè)匯報(bào)資源鸥跟，以及周期性的匯報(bào)節(jié)點(diǎn)信息和Container的運(yùn)行狀態(tài)丢郊，同時(shí)RM會(huì)返回給待清理的Container列表，待清理的應(yīng)用程序医咨，診斷信息等;
2. ContainerManager: 該組件是NodeManager最核心的組件之一枫匾，它內(nèi)部有很多子組件組成，如上圖所示拟淮；
   -RPC Server: 該協(xié)議實(shí)現(xiàn)了ContainerManagementProtocol協(xié)議干茉，實(shí)現(xiàn)AM與NM之前的通信通道，通過(guò)該協(xié)議接受來(lái)自各個(gè)AM的啟動(dòng)或者殺死Container的命令;
   -ResourceLocalizationService: 負(fù)責(zé)Container所需資源的本地化很泊，按照資源描述將資源從HDFS上下載所需資源角虫，并將這些資源均攤給各個(gè)磁盤(pán)以防數(shù)據(jù)熱點(diǎn);
   -ContainerLauncher: 維護(hù)一個(gè)線(xiàn)程池以并行的方式完成Container的操作，比如來(lái)自AM的啟動(dòng)Container委造，來(lái)自AM或者RM的kill Container戳鹅；
   -AuxServices: NM允許用戶(hù)添加附屬服務(wù);
   -ContainersMonitor: 監(jiān)控Container資源的使用狀況，周期性的檢查Container資源使用情況昏兆，一旦有Container超過(guò)它允許使用資源的上限則kill掉枫虏；
   -Loghandler: 可插拔的組件，控制Container日志保存的方式爬虱；
   -ContainerEventDispatcher: Container的事件調(diào)度器隶债，負(fù)責(zé)將ContainerEvent類(lèi)型的事件調(diào)度給對(duì)應(yīng)Container的狀態(tài)機(jī)ContainerImpl;
   -ApplicationEventDispatcher: Application的事件調(diào)度器,負(fù)責(zé)將ApplicationEvent類(lèi)型的事件調(diào)度給Application的狀態(tài)機(jī)ApplicationImpl;
3. ContainerExecutor: 它可與底層的操作系統(tǒng)交互，安全存放Container的文件和目錄饮潦，進(jìn)而安全的啟動(dòng)和清理Container燃异；
4. NodeHealthCheckerService: 周期性的運(yùn)行一個(gè)向磁盤(pán)寫(xiě)文件的腳本,檢測(cè)節(jié)點(diǎn)的健康狀態(tài)，并匯報(bào)給RM继蜡；
5. DeletionService: NodeManager刪除文件組件回俐；
6. Security: 安全模塊;
7. WebServer： web頁(yè)面展示該節(jié)點(diǎn)上應(yīng)用程序的狀態(tài);

NodeManager的事件與事件處理器

NodeManager主要組件也是通過(guò)事件進(jìn)行交互的，這使得組件能夠異步并發(fā)完成各種功能稀并；如下圖所示:

image

節(jié)點(diǎn)健康狀態(tài)監(jiān)測(cè)

節(jié)點(diǎn)健康狀態(tài)監(jiān)測(cè)是NodeManager自帶的健康狀態(tài)診斷機(jī)制仅颇，通過(guò)該機(jī)制，NodeManager可以時(shí)刻的掌握資深的健康狀況碘举，并及時(shí)的匯報(bào)給RM忘瓦，RM根據(jù)節(jié)點(diǎn)的健康狀況調(diào)整分配的任務(wù)數(shù)目;如果NodeManager監(jiān)測(cè)到自身的狀態(tài)為“unhealthy”,則通知RM不再為該節(jié)點(diǎn)分配任務(wù)，直到狀態(tài)轉(zhuǎn)為"healthy"再向該節(jié)點(diǎn)分配任務(wù);

自定義監(jiān)測(cè)腳本
NodeManager有兩種判斷節(jié)點(diǎn)健康狀態(tài)的方法：
第一種: 管理員自定義shell腳本引颈，NodeManager上有一個(gè)周期性的執(zhí)行腳本耕皮，如果腳本監(jiān)測(cè)到節(jié)點(diǎn)為不健康狀態(tài)境蜕，則腳本只要標(biāo)準(zhǔn)輸出“ERROR”開(kāi)頭的字符串即可，NodeHealthScriptRunner周期性的監(jiān)測(cè)發(fā)現(xiàn)腳本輸出“ERROR”開(kāi)頭的字符串則認(rèn)為該節(jié)點(diǎn)為不健康的節(jié)點(diǎn)凌停，并通過(guò)心跳通知RM粱年，RM將會(huì)把該節(jié)點(diǎn)加入黑名單，知道該節(jié)點(diǎn)的狀態(tài)轉(zhuǎn)為健康狀態(tài)罚拟，RM再將其移除黑名單台诗，并為該節(jié)點(diǎn)分配任務(wù)；
NodeHealthScriptRunner服務(wù)主要工作是周期性的執(zhí)行節(jié)點(diǎn)健康狀況的監(jiān)測(cè)腳本赐俗，即上面所說(shuō)的自定義腳本,但需要在yarn-site.xml中配置,如下:

image

image

第二種: 判斷磁盤(pán)的好壞拉队，NodeManager上有一個(gè)周期性的執(zhí)行腳本監(jiān)測(cè)磁盤(pán)的好壞，默認(rèn)的情況下該功能是開(kāi)啟的阻逮，該功能由LocalDirsHandlerService服務(wù)實(shí)現(xiàn)的粱快，它周期性的任務(wù)檢測(cè)NodeManager本地磁盤(pán)的好壞，并通過(guò)心跳將健康狀態(tài)匯報(bào)給RM夺鲜；NodeManager有兩個(gè)目錄一個(gè)是本地可用的目錄列表用于存放程序或者任務(wù)的中間結(jié)果皆尔，另一個(gè)是日志目錄，而LocalDirsHandlerService服務(wù)就是檢測(cè)這些目錄的好壞币励，一旦發(fā)現(xiàn)正常磁盤(pán)的比例低于設(shè)定的值(yarn.nodemanager.disk-health-checker.min-healthy-disks默認(rèn)0.25)則將節(jié)點(diǎn)置為不健康狀態(tài);

分布式緩存

分布式緩存介紹
在Yarn中慷蠕，分布式緩存是一中分布式文件分發(fā)和緩存機(jī)制，主要作用是將用戶(hù)應(yīng)用程序需要的外部文件資源自動(dòng)透明的下載并緩存到各個(gè)節(jié)點(diǎn)上食呻，從而省去了用戶(hù)手動(dòng)部署的麻煩流炕，注意，這里的分布式緩存并不是將文件緩存到集群各個(gè)節(jié)點(diǎn)的內(nèi)存中,而是將文件緩存到各個(gè)節(jié)點(diǎn)的本地磁盤(pán)上;
Yarn的分布式緩存工作流程如下:

image

1. 客戶(hù)端將應(yīng)用程序所需的文件資源提交到HDFS上;
2. 客戶(hù)端將應(yīng)用程序提交給RM仅胞；
3. RM與某個(gè)NM通信每辟，啟動(dòng)應(yīng)用程序的ApplicationMaster，NodeManager收到命令后干旧，先從HDFS上下載文件資源渠欺，然后啟動(dòng)ApplicationMaster；
4. ApplicationMaster與RM通信請(qǐng)求獲取計(jì)算資源椎眯；
5. ApplicationMaster收到RM分配的資源后與NM通信啟動(dòng)任務(wù)挠将；
6. 如果該應(yīng)用程序第一次再該節(jié)點(diǎn)上啟動(dòng)任務(wù)，則NN先從HDFS上下載文件資源緩存在本地编整，然后再啟動(dòng)任務(wù)舔稀；
7. NM后續(xù)在收到啟動(dòng)任務(wù)的請(qǐng)求時(shí)，先判斷所需的文件資源是否已經(jīng)下載到本地掌测，如果沒(méi)有下載到本地則等待文件緩存完畢后再啟動(dòng)任務(wù);

資源可見(jiàn)性和分類(lèi)
分布式緩存機(jī)制是由各個(gè)NM實(shí)現(xiàn)的内贮，主要功能是將應(yīng)用程序所需的文件資源緩存到本地，以便后續(xù)任務(wù)的使用，資源緩存是用時(shí)觸發(fā)的夜郁，也就是第一個(gè)用到該資源的任務(wù)觸發(fā)什燕，后續(xù)任務(wù)無(wú)需再進(jìn)行緩存，直接使用即可拂酣；
根據(jù)可見(jiàn)性,NM將資源分為三類(lèi)：

1. Public:節(jié)點(diǎn)上所有的用戶(hù)都可以共享該資源秋冰，只要有一個(gè)用戶(hù)的應(yīng)用程序?qū)⒅@些資源緩存到本地，其他所有用戶(hù)的所有應(yīng)用程序都可以使用婶熬；
2. Private: 節(jié)點(diǎn)上同一用戶(hù)的所有應(yīng)用程序共享該資源,只要該用戶(hù)其中一個(gè)應(yīng)用程序?qū)①Y源緩存到本地，該用戶(hù)的所有應(yīng)用程序都可以使用埃撵；

3. Application: 節(jié)點(diǎn)上同一應(yīng)用程序的所有Container共享該資源;

   
   
   image

根據(jù)資源類(lèi)型赵颅，NM向資源分為三類(lèi)：

1. archive: 歸檔文件,支持.jar,.zip,.tar.gz,.tgz,.tar5中歸檔文件；
2. file : 普通文件暂刘，NM只是將這類(lèi)文件下載到本地目錄饺谬，不做任何處理；
3. pattern: 以上兩種文件的混合體谣拣；

注意點(diǎn):

1. YARN是通過(guò)比較resource募寨、type、timestamp和pattern四個(gè)字段是否相同來(lái)判斷兩個(gè)資源請(qǐng)求是否相同的森缠。如果一個(gè)已經(jīng)被緩存到各個(gè)節(jié)點(diǎn)上的文件被用戶(hù)修改了拔鹰，則下次使用時(shí)會(huì)自動(dòng)觸發(fā)一次緩存更新，以重新從HDFS上下載文件贵涵。
2. 分布式緩存完成的主要功能是文件下載列肢，涉及大量的磁盤(pán)讀寫(xiě)，因此整個(gè)過(guò)程采用了異步并發(fā)模型加快文件下載速度宾茂，以避免同步模型帶來(lái)的性能開(kāi)銷(xiāo)瓷马。為了防止緩存過(guò)多磁盤(pán)被撐爆，NM會(huì)采用LRU算法定期清理這些文件;

目錄結(jié)構(gòu)

NodeManager上的目錄可分為兩種:

1. 數(shù)據(jù)目錄: 存放執(zhí)行Container所需的數(shù)據(jù)(比如可執(zhí)行程序或JAR包跨晴、配置文件等)和運(yùn)行過(guò)程中產(chǎn)生的臨時(shí)數(shù)據(jù),由yarn.nodemanager.local-dirs指定;
2. 日志目錄: 存放Container運(yùn)行輸出日志欧聘，由yarn.nodemanager.log-dirs指定;

NM在每個(gè)磁盤(pán)上為該作業(yè)創(chuàng)建相同的目錄結(jié)構(gòu)，且采用輪詢(xún)的調(diào)度方式將目錄（磁盤(pán)）分配給不同的Container的不同模塊以避免干擾端盆。
考慮到一個(gè)應(yīng)用程序的不同Container之間存在依賴(lài)怀骤，為了避免提前清除已經(jīng)完成的Container輸出的中間數(shù)據(jù)破壞應(yīng)用程序的設(shè)計(jì)邏輯，YARN統(tǒng)一規(guī)定爱谁，只有當(dāng)應(yīng)用程序運(yùn)行結(jié)束后晒喷，才統(tǒng)一清楚Container產(chǎn)生的中間數(shù)據(jù)。

狀態(tài)機(jī)管理

NM維護(hù)了三類(lèi)狀態(tài)機(jī)访敌，分別是:Application凉敲、Container和LocalizedResource,它們均直接或者間接參與維護(hù)一個(gè)應(yīng)用程序的生命周期;
當(dāng)NodeManager收到來(lái)自某個(gè)應(yīng)用程序第一次Container啟動(dòng)命令時(shí)，會(huì)創(chuàng)建一個(gè)Application狀態(tài)機(jī)跟蹤該應(yīng)用程序在該結(jié)點(diǎn)上的生命周期，而每個(gè)Container的運(yùn)行過(guò)程同樣由一個(gè)狀態(tài)機(jī)維護(hù)爷抓。此外Application所需的資源(比如文本文件势决、JAR包、歸檔文件等）需要從HDFS上下載蓝撇，每個(gè)資源的下載過(guò)程均由一個(gè)狀態(tài)機(jī)LocalizedResouce維護(hù)和跟蹤果复。

狀態(tài)機(jī)如下：

Application狀態(tài)機(jī)
NM上Application維護(hù)的信息是RM端Application信息的子集，這有利于對(duì)一個(gè)節(jié)點(diǎn)上的同一個(gè)Application的所有Container進(jìn)行統(tǒng)一管理（比如記錄每一個(gè)Application運(yùn)行在該節(jié)點(diǎn)上的Container列表渤昌，殺死一個(gè)Application的所有Container等）虽抄。它實(shí)際的實(shí)現(xiàn)類(lèi)是ApplicationImpl，該類(lèi)維護(hù)了一個(gè)Application狀態(tài)機(jī)独柑，記錄了Application可能存在的各個(gè)狀態(tài)以及導(dǎo)致?tīng)顟B(tài)間轉(zhuǎn)換的事件迈窟。需要注意的是NM上的Application生命周期與ResourceManager上Application的生命周期是一致的。

Container狀態(tài)機(jī)
Container是NM中用于維護(hù)Container生命周期的數(shù)據(jù)結(jié)構(gòu)忌栅，它的是現(xiàn)實(shí)ContainerImpl车酣，該類(lèi)維護(hù)了一個(gè)Container的狀態(tài)機(jī)，記錄了Container可能存在的各種狀態(tài)以及導(dǎo)致?tīng)顟B(tài)轉(zhuǎn)換的事件;

LocalizedResource狀態(tài)機(jī)
LocalizedResource是NodeManager中維護(hù)一種“資源”(資源文件索绪、JAR包湖员、歸檔文件等外部文件資源)生命周期的數(shù)據(jù)結(jié)構(gòu)，它維護(hù)了一個(gè)狀態(tài)瑞驱，記錄了"資源"可能存在的各種狀態(tài)以及導(dǎo)致?tīng)顟B(tài)間轉(zhuǎn)換的事件娘摔。

Container生命周期

Container啟動(dòng)命令是由各個(gè)ApplicationMaster通過(guò)RPC函數(shù)ContainerManager.startContainer()向NodeManager發(fā)起的；
Container啟動(dòng)過(guò)程主要經(jīng)歷三個(gè)階段： 資源本地化钱烟、啟動(dòng)container,運(yùn)行container晰筛、資源回收;

資源本地化
資源本地化是指準(zhǔn)備container運(yùn)行所需的環(huán)境，包括創(chuàng)建container工作目錄拴袭，從HDFS下載運(yùn)行container所需的各種資源（jar包读第、可執(zhí)行文件等）等。

YARN將資源分為兩類(lèi)拥刻，一類(lèi)是public級(jí)別的資源怜瞒，這類(lèi)資源被放到一個(gè)公共目錄下，由所有用戶(hù)共享般哼，另一類(lèi)是private級(jí)別的資源吴汪，這類(lèi)資源是用戶(hù)私有的，只能在所屬用戶(hù)的各個(gè)作業(yè)間共享蒸眠。資源本地化過(guò)程實(shí)際上就是準(zhǔn)備public和private資源的過(guò)程漾橙，它由ResourceLocalizationService服務(wù)完成，其中楞卡，所有application的public資源由專(zhuān)門(mén)的線(xiàn)程PublicLocalizer下載完成霜运，該線(xiàn)程內(nèi)部維護(hù)了一個(gè)線(xiàn)程池以加快資源下載速度脾歇，每個(gè)application的private資源由一個(gè)專(zhuān)門(mén)的線(xiàn)程LocalizerRunner下載完成。

啟動(dòng)container:
啟動(dòng)Container是由ContainersLauncher完成的淘捡，該過(guò)程主要工作是將運(yùn)行container對(duì)應(yīng)的完整shell命令寫(xiě)到私有目錄下的launch_container.sh中藕各，并將token文件寫(xiě)到container_tokens中。之所以要將container運(yùn)行命令寫(xiě)到launch_container.sh中焦除，然后通過(guò)運(yùn)行shell腳本的形式運(yùn)行container激况，主要是因?yàn)橹苯訄?zhí)行命令可能會(huì)有些特殊符號(hào)不識(shí)別。

運(yùn)行container
而運(yùn)行container是由插拔式組件ContainerExecutor完成的膘魄，YARN提供了兩種ContainerExecutor實(shí)現(xiàn)乌逐，一種是DefaultContainerExecutor，另一種是LinuxContainerExecutor瓣距。DefaultContainerExecutor只是簡(jiǎn)單的以管理員身份運(yùn)行l(wèi)aunch_container.sh腳本黔帕，而LinuxContainerExecutor則是以container所屬用戶(hù)身份運(yùn)行該腳本，它是Hadoop引入安全機(jī)制后加入的蹈丸，此外，在不久的將來(lái)呐芥，container會(huì)引入cgroups隔離cpu資源逻杖，相關(guān)的代碼也會(huì)存放在LinuxContainerExecutor中。

資源回收:
Container運(yùn)行完成后思瘟，NM需要回收它占用的資源荸百，這些資源主要是Container運(yùn)行時(shí)使用的臨時(shí)文件；
資源回收由ResourceLocalizationService服務(wù)和ContainerExecutor組件完成的滨攻，該過(guò)程與資源本地化正好相反够话，它負(fù)責(zé)撤銷(xiāo)container運(yùn)行過(guò)程中使用的各種資源。

資源隔離

YARN對(duì)內(nèi)存資源和CPU資源采用了不同的資源隔離方案光绕。
對(duì)于內(nèi)存資源女嘲，為了能夠更靈活的控制內(nèi)存使用量，YARN采用了進(jìn)程監(jiān)控的方案控制內(nèi)存使用诞帐，即每個(gè)NodeManager會(huì)啟動(dòng)一個(gè)額外監(jiān)控線(xiàn)程監(jiān)控每個(gè)container內(nèi)存資源使用量欣尼，一旦發(fā)現(xiàn)它超過(guò)約定的資源量，則會(huì)將其殺死停蕉。采用這種機(jī)制的另一個(gè)原因是Java中創(chuàng)建子進(jìn)程采用了fork()+exec()的方案愕鼓，子進(jìn)程啟動(dòng)瞬間，它使用的內(nèi)存量與父進(jìn)程一致慧起，從外面看來(lái)菇晃，一個(gè)進(jìn)程使用內(nèi)存量可能瞬間翻倍，然后又降下來(lái)蚓挤，采用線(xiàn)程監(jiān)控的方法可防止這種情況下導(dǎo)致swap操作磺送。
對(duì)于CPU資源驻子，則采用了Cgroups進(jìn)行資源隔離；

參考:
《Hadoop技術(shù)內(nèi)幕 : 深入解析YARN架構(gòu)設(shè)計(jì)與實(shí)現(xiàn)原理》