Hadoop0.23.0 RM-NM-AM源码级分析

云计算 waitig 462℃ 百度已收录 0评论

本文装载自klose  http://blog.sina.com.cn/s/blog_4a1f59bf01011e6n.html

Hadoop0.23.0RM-NM-AM源码级分析

缩写词

NMNodeManager

RMResourceManager

AM:ApplicationManager

 

表格1 ResourceManager的服务

父类或接口

实现类

功能

Dispatcher

AsyncDispatcher

转交各种事件到不同的处理窗口,需要注册事件与处理类的绑定。(线程级)

AbstractLiveLiness

ContainerAllocationExpirer

监控Containter是否到期

AbstractLiveLiness

AMLivelinessMonitor

监控每一个App的工作状态

AbstractService/ResourceTracker

ResourceTrackerService

负责NM的注册以及与NM保持心跳

AbstractService/ClientRMProtocol

ClientRMService

连接到RM的客户端,clientRM之间的交互。

AbstractService/NodeStatusUpdater

NodeStatusUpdaterImpl

监控节点状态更新的情

AbstractLivelinessMonitor

NMLivelinessMonitor

监控每一个NM的存活状态

AbstractService/AMRMProtocol

ApplicationMasterService

AM交互的服务

/EventHandler<RMAppManagerEvent>

RMAppManager

RM统管Applist

AbstractService/   EventHandler<AMLauncherEvent>

ApplicationMasterLauncher

管理AM的启动和退出。(线程级)

AbstractService/RMAdminProtocol

AdminService

实现RM的管理的协议的服务

 

表格2 NodeManager的服务

 

父类或接口

实现类

功能

ContainerExecutor

DefaultContainerExecutor(conf可配置)

Container执行服务

AbstractService

DeletionService

删除执行容器服务

AbstractService

NodeHealthCheckerService

检查节点的健康状况,提供可访问API

AbstractService/NodeStatusUpdater

NodeStatusUpdaterImpl

提供向RM注册和心跳服务

AbstractService/   NodeResourceMonitor

NodeResourceMonitorImpl

Hadoop0.23.0没有具体的实现

CompositeService/   ServiceStateChangeListener, ContainerManager,

EventHandler<ContainerManagerEvent>

ContainerManagerImpl

提供节点内容器的管理服务

 

 

 

 

RM交互的分析:

1Client提交作业的过程。

        1Client提交作业给JobSubmitterJobSubmitter进行一系列的作业初始化工作之后,通过Java接口和继承机制,JobSubmitter中成员变量submitClient映射成YARNRunner的一个实例,YARNRunner是新版本的JobClient

        2YARNRunner执行submitJobYARNRunner中的ResourceMgrDelegate通过RPC实现ClientRMProtocol接口的客户端实例代理,实现Client端与RM进行交互的目的。

        ClientRMProtocol的接口如下表所示:

3:接口ClientRMProtocol的函数

接口

含义

getNewApplication(GetNewApplicationRequestrequest)

获取一个ApplicationID和集群当前状态的信息

submitApplication(

     SubmitApplicationRequest request

RM提交一个作业

forceKillApplication(

     KillApplicationRequest request)

RM请求停止一个作业的执行

getApplicationReport(

     GetApplicationReportRequest request)

RM获取一个作业的状态报告

getClusterMetrics(

     GetClusterMetricsRequest request)

RM获得集群的资源使用情况信息

getAllApplications(

     GetAllApplicationsRequest request)

RM获得所有作业的状态信息

getClusterNodes(

     GetClusterNodesRequest request)

RM获得所有工作节点的状态信息

getQueueInfo(

     GetQueueInfoRequest request)

RM获取作业队列的信息

getQueueUserAcls(

     GetQueueUserAclsInfoRequest request)

RM获取当前用户的ACL

 

接口使用方式:

 YarnConfiguration conf = newYarnConfiguration();

 InetSocketAddress rmAddress=

       NetUtils.createSocketAddr(this.conf.get(

           YarnConfiguration.RM_ADDRESS,

           YarnConfiguration.DEFAULT_RM_ADDRESS),

           YarnConfiguration.DEFAULT_RM_PORT,

           YarnConfiguration.RM_ADDRESS);

  ClientRMProtocol applicationsManager=

       (ClientRMProtocol) rpc.getProxy(ClientRMProtocol.class,

           rmAddress, this.conf);     

      接下来就可以使用applicationManager调用表格3中的接口,接口函数中输入和返回参数请参阅源码。

        RM中负责与Client进行交互的服务是ClientRMService

2NMRM的交互

        1NMRM注册。NM继承CompositeService,在init()中将表2中的服务添加到系统中,NM调用start(),通过super.start()会启动所有服务。NodeStatusUpdater作为NM的一种服务被启动起来,在NodeStatusUpdaterImplstart函数内,调用函数registerWithRM()实现向RM注册。

        NMRM注册服务的过程如下:

        YarnRPC rpc = YarnRPC.create(conf);

   InetSocketAddress rmAddress =NetUtils.createSocketAddr(this.rmAddress,

     YarnConfiguration.DEFAULT_RM_RESOURCE_TRACKER_PORT,

     YarnConfiguration.RM_RESOURCE_TRACKER_ADDRESS);

        ResourceTracker resourceTracker = (ResourceTracker)rpc.getProxy(ResourceTracker.class, rmAddress,
 conf);

        resourceTracker.registerNodeManager(request);//request需要根据RM的端口号等相关信息设置

        2NMRM之间心跳。在NM通过启动NodeStatusUpdater服务,借助NodeStatusUpdaterImplregisterWithRM()注册到对应的RM之后,调用startStatusUpdater(),启动一个线程,周期性地与RM之间保持心跳。

        NMRM之间的心跳链接如下:

        NodeStatus nodeStatus = getNodeStatus();

   nodeStatus.setResponseId(lastHeartBeatID);

   NodeHeartbeatRequest request = recordFactory

        .newRecordInstance(NodeHeartbeatRequest.class);

   request.setNodeStatus(nodeStatus);           

    HeartbeatResponse response =

   resourceTracker.nodeHeartbeat(request).getHeartbeatResponse();//NMRM报告状态信息,RMNM做反馈

        org.apache.hadoop.yarn.server.api.ResourceTrackerNMRM之间的注册与心跳服务接口。

        4 ResourceTracker的接口函数

接口

含义

registerNodeManager(RegisterNodeManagerRequestrequest)

NM注册到RM

nodeHeartbeat(NodeHeartbeatRequest request)

NM发送心跳到RM

        

RM中负责NM中心跳和注册功能的服务类为ResourceTrackerService

Hadoop0.23.0初探5---RM、NM与AM源码分析

1NMRM之间的交互

        3NMRM之间交互与资源分配关系

        NMRM之间进行心跳互联最为核心的工作是完成NMContainer的分配和释放。

² NM启动过程,通过ResourceTracker的协议注册到RM上,注册过程会创建一个RMNode的对象,记录NM上执行ApplicationsContainers的状况。NM注册到RM意味着,RM有更多可以调度的资源。因此,注册过程最后发送NodeAddedSchedulerEvent事件发送给RM上绑定的调度器,调度器将RMNode加入到可利用资源列表。

² NMRM之间心跳过程。NM完成注册到RM的过程之后,会启动进入到一个线程循环中。

3RMAM之间的交互

        1Job提交到RM的过程。Client将作业提交到RMRMClientRMService负责处理该请求。ClientRMService将作业封装成RMAppManagerSubmitEvent,直接调用RMAppManager的事件处理程序。

        RMAppManagerhandle函数,处理RMAppManagerSubmitEvent事件。使用RMAppImpl初始化RMApp,提交<app-idRMApp>RMContext,提交<app-id,App-ACL>ApplicationACLsManager。最后,将applicationId,
RMAppEventType.START
封装成RMAppEvent,由RM中的dispatcher分配给RMAppImplhandle函数处理。

根据RMAppImpl设置的状态机,RMAppEventType.START事件驱动RMAppState.NEW状态到RMAppState.SUBMITTED,使用RMAppAttemptImpl的实例化RMAppAttempt,将RMAppAttemptEventType.START事件传递给RMAppAttempt

        根据RMAppAttemptImpl状态机的设置,RMAppAttemptEventType.START事件驱动RMAppAttemptState.NEW转移到RMAppAttemptState.SUBMITTED,同时将AppAttempt注册到RMApplicationMasterService,并将该AppAttempt提交到ResourceSchedulerHadoop-Yarn提供的默认调度器是FifoResourceScheduler。然后,通过RMSchedulerEventDispatcher分发AppAddedSchedulerEvent事件。

        SchedulerEventDispatcher是线程模式的事件分配器,它将接收到的事件交给RM设置的ResourceScheduler来处理。FifoResourceScheduler使用handle(SchedulerEvent
event)
处理AppAddedSchedulerEvent。由AppAttemptID初始化一个SchedulerApp,并向RMAppAttemptImpl发送RMAppAttemptEventType.APP_ACCEPTED类型的事件。

        根据RMAppAttemptImpl状态机的设置,RMAppAttemptEventType.APP_ACCEPTED事件驱动RMAppAttemptState.SUBMITTED转移到RMAppAttemptState.SCHEDULED,同时将该AppAttempt被接收的事件向RMApp发送RMAppEventType.APP_ACCEPTED类型的事件,使得RMAppStateSUBMITTED转移到ACCEPTED。最后,为了启动AM,向资源调度器请求一个Container

        到此,由JobClient提交的App的过程在RM的操作全部结束,这个过程中主要是创建了RMAppAttempt实例,并且修改了RMApp的状态,系统识别到当前有一个App被添加到RM

        2NMRM的心跳,部署ApplicationMasterNMContainer

        NM周期性地向RM传送HeartbeatRMReourceTrackerService负责心跳事件,在RM端使用RMNode来标识NM节点的状态。NM心跳带来的NM节点状态的情况,会触发RMNodeStatusEvent,该类型的事件被绑定RMNodeImpl来处理。

        心跳过程产生了RMNodeEventType.STATUS_UPDATE事件,触发

RMNodeState.RUNNING->EnumSet.of(RMNodeState.RUNNING,RMNodeState.UNHEALTHY)
的状态转移,并且会调用钩子new statusUpdateWhenHealthyTransition()。在该函数内会查阅NM状态信息中Container的状态,并把Container状态的更新以及Container的增删情况同步到RMNodeImpl中的justLaunchedContainers,并且将NM中处于RUNNING状态的Container封装到newlyLaunchedContainers,将NM中处于COMPLETE状态的Container封装到completedContainers,然后封装成NodeUpdateSchedulerEvent事件交由给FifoResourceScheduler来处理。

        FifoResourceScheduler接收到NodeUpdateSchedulerEvent,执行nodeUpdate函数。nodeUpdate函数操作App到可用的Container上,或者回收已经处于FINISHED状态的Container状态资源。

        假设我们第一次提交App,跟踪AM启动流程如下:

        1)调用assignContainers(node);分配Container到心跳互联的节点。

        2assignContainers(node),遍历待提交的Application队列,计算启动AMContainer需要的资源和需求,与该node的资源和特征进行匹配。匹配成功意味着,当前节点满足Application调度的需求,进入第3)步。

        3assignContainersOnNode(node, application,priority)Container分配给app。由于App没有数据本地性的需求,因此,进入assignOffSwitchContainers(node,
application,priority)
,进入4

        4)调用assignContainer(node, application,priority,

           request.getNumContainers(), request,NodeType.OFF_SWITCH);

        创建一个新的Container,并将该Container通知app对应SchedulerApp,在SchedulerApp中创建RMContainer的实例,并添加到SchedulerAppnewlyAllocatedContainers,并触发RMContainerEventType.START事件。

        5RMContainerEventType.START事件的hook函数会触发一个RMAppAttemptContainerAllocatedEvent的事件,使得RMAppAttemptStateSCHEDULEDALLOCATED,并调用回调函数。

        6RMAppAttemptContainerAllocatedEvent的回调函数,会通过FifoResourceScheduler获得一个AM
Container
,并触发RMContainerState状态从ALLOCATEDACQUIRED,同时,将该Container设置为AppAttempt.masterContainer,最后触发AMLauncherEventType.LAUNCH事件。

        7AMLauncher调用launch(),获得NMContainerManager的代理,设置ContainerContext,然后通过代理调用NM端的ContainerManagerstartContainer,这样在该NM上就启动了一个ApplicationMaster

        8)将ApplicationMaster注册到RM中的AMLivelinessMonitor,此时,每次RMAppAttemptEventType.STATUS_UPDATE事件,会通知AMLivelinessMonitor,通过时间戳探测AM的健康状况。

ps:快写吐了,凌乱发表。是不是看到这里,也快看吐了,对不住,影响大家心情了Hadoop0.23.0初探5---RM、NM与AM源码分析建议大家还是看源码吧。为了不影响转载的心情,大家还是传递一下这个博客吧.Hadoop0.23.0初探5—RM、NM与AM源码级分析

如果愿意一起阅读Hadoop0.23的源码,并愿意与他人分享,联系方式如下。

Hadoop0.23.0初探5---RM、NM与AM源码分析


本文由【waitig】发表在等英博客
本文固定链接:Hadoop0.23.0 RM-NM-AM源码级分析
欢迎关注本站官方公众号,每日都有干货分享!
等英博客官方公众号
点赞 (0)分享 (0)