前言YARN是一个资源管理系统负责集群资源的管理和分配。yarn就好比hadoop集群的操作系统当用户向YARN中提交一个应用程序后需要同操作系统做一些交互这样才能运行在yarn上。yarn分两个阶段运行应用程序 第一个阶段是启动ApplicationMaster 第二个阶段是由ApplicationMaster创建应用程序为它申请资源并监控它的整个运行过程直到运行完成。如果想要将一个新的应用程序跑在yarn上通常需要编写这两个组件客户端和ApplicationMaster。这两个组件的编写非常复杂尤其是ApplicationMaster需要考虑rpc调用容错等细节往往由专业的开发人员编写这两个组件然后提供给上层的应用程序用户使用。如果大量 应用程序可以抽象为一种通用的框架 那么只需要实现一个客户端和一个ApplicationMaster这样所有的应用程序直接用这两个组件即可比如MapReduce是一种通用的计算框架YARN已经内嵌了一个可以直接使用的客户端MRClientService和ApplicationMaster—MRAppMaster。对于无法用MapReduce计算框架抽象出来的应用程序要想跑在yarn上就必须用户自己编写这两个组件客户端和ApplicationMaster其中客户端负责向ResourceManager提交ApplicationMaster并查询应用程序运行状态ApplicationMaster负责向ResourceManager申请资源以Container形式表示并与NodeManager通信以启动各个Container此外ApplicationMaster还负责监控各个任务运行状态并在失败是为其重新申请资源。开发Client启动AMClient部分是用于将应用提交到YARN, 从而可以启动application master.客户端通常只需与ResourceManager交互期间涉及到多个数据结构和一个RPC协议具体如下image.png客户端通过rpc协议ApplicationClientProtocal向ResourceManager具体是ApplicationsManager、ASM发送应用程序提交请求GetNewApplicationRequestResourceManager为其返回应答GetNewApplicationResponse该数据结构中包含多种信息包括ApplicationId、可资源使用上限和下限等。初始化并启动一个yarnClient:YarnClient yarnClientYarnClient.createYarnClient();yarnClient.init(conf);yarnClient.start();YarnClientApplication appyarnClient.createApplication();GetNewApplicationResponse appResponseapp.getNewApplicationResponse();Client部分最关键的是构建一个ApplicationSubmissionContext。启动ApplicationMaster所需的所有信息打包到数据结构ApplicationSubmissionContext中主要包括以下几种信息(1)applicationid(2)application 名称(3)application优先级(4)application 所属队列(5)application 启动用户名(6)ApplicationMaster对应的Container信息ContainerLaunchContext包括启动ApplicationMaster所需各种文件资源、jar包、环境变量、启动命令、运行ApplicationMaster所需的资源主要指内存等。// set the application nameApplicationSubmissionContextappContextapp.getApplicationSubmissionContext();ApplicationIdappIdappContext.getApplicationId();appContext.setKeepContainersAcrossApplicationAttempts(keepContainers);appContext.setApplicationName(appName);// Set up the container launch context for the application masterContainerLaunchContextamContainerContainerLaunchContext.newInstance(localResources,env,commands,null,null,null);// Set up resource type requirements// For now, both memory and vcores are supported, so we set memory and// vcores requirementsResourcecapabilityResource.newInstance(amMemory,amVCores);appContext.setResource(capability);客户端调用ClientRMProtocol#submitApplication(ApplicationSubmissionContext)将ApplicationMaster提交到ResourceManager上。ResourceManager收到请求后会为ApplicationMaster寻找合适的节点并在该节点上启动它。LOG.info(Submitting application to ASM);yarnClient.submitApplication(appContext);客户端可通过多种方式查询应用程序的运行状态其中一种是调用RPC函数ClientRMProtocol#getApplicationReport获取一个应用程序当前运行状况报告该报告内容包括应用程序名称、所属用户、所在队列、ApplicationMaster所在节点、一些诊断信息、启动时间等。// Get application report for the appId we are interested inApplicationReportreportyarnClient.getApplicationReport(appId);LOG.info(Got application report from ASM for, appIdappId.getId(), clientToAMTokenreport.getClientToAMToken(), appDiagnosticsreport.getDiagnostics(), appMasterHostreport.getHost(), appQueuereport.getQueue(), appMasterRpcPortreport.getRpcPort(), appStartTimereport.getStartTime(), yarnAppStatereport.getYarnApplicationState().toString(), distributedFinalStatereport.getFinalApplicationStatus().toString(), appTrackingUrlreport.getTrackingUrl(), appUserreport.getUser());YarnApplicationStatestatereport.getYarnApplicationState();FinalApplicationStatusdsStatusreport.getFinalApplicationStatus();如果有异常或者其他情况可以通过yarnClient.killApplication(appId);来kill掉应用开发ApplicationMasterApplicationMaster需要与ResoureManager和NodeManager交互以申请资源和启动Container期间涉及到多个数据结构和两个RPC协议。具体步骤如下ApplicationMaster首先需通过RPC协议AMRMProtocol向ResourceManager发送注册请求RegisterApplicationMasterRequest该数据结构中包含ApplicationMaster所在节点的host、RPC port和TrackingUrl等信息而ResourceManager将返回RegisterApplicationMasterResponse该数据结构中包含多种信息包括该应用程序的ACL列表、可资源使用上限和下限等。image.pngApplicationMaster与RM之间的心跳整个运行过程中ApplicationMaster需通过心跳与ResourceManager保持联系这是因为如果一段时间内默认是10minResourceManager未收到ApplicationMaster信息则认为它死掉了会重新调度或者让其失败。通常而言ApplicationMaster周期性调用RPC函数ApplicationMasterProtocol.allocate向其发送空的AllocateRequest请求即可。构造Container根据每个任务的资源需求ApplicationMaster可向ResourceManager申请一系列用于运行任务的ContainerApplicationMaster使用ResourceRequest类描述每个Container一个container只能运行一个任务1Hostname期望Container所在的节点如果是*表示可以为任意节点。 2Resource capability运行该任务所需的资源量如(memory/disk/cpu)。 3Priority任务优先级。一个应用程序中的任务可能有多种优先级ResourceManager会优先为高优先级的任务分配资源。 4numContainers符合以上条件的container数目。申请资源分配Container一旦为任务构造了Container后ApplicationMaster会使用RPC函数AMRMProtocol#allocate向ResourceManager发送一个AllocateRequest对象以请求分配这些ContainerAllocateRequest中包含以下信息1Requested containers所需的Container列表 2Released containers有些情况下比如有些任务在某些节点上失败过则ApplicationMaster不想再在这些节点上运行任务此时可要求释放这些节点上的Container。 3Progress update information应用程序执行进度 4ResponseIdRPC响应ID每次调用RPC该值会加1。ResourceManager会为ApplicationMaster返回一个AllocateResponse对象该对象中主要信息包含在AMResponse中1rebootApplicationMaster是否需要重新初始化.当ResourceManager端出现不一致状态时会要求对应的ApplicationMaster重新初始化。 2Allocated Containers新分配的container列表。 3Completed Containers已运行完成的container列表该列表中包含运行成功和未成功的ContainerApplicationMaster可能需要重新运行那些未运行成功的Container。ApplicationMaster会不断追踪已经获取的container且只有当需求发生变化时才允许重新为Container申请资源。image.png启动Container, 当ApplicationMaster从ResourceManager端收到新分配的Container列表后会使用ContainerManagementProtocol#startContainer向对应的NodeManager发送ContainerLaunchContext以启动ContainerContainerLaunchContext包含以下内容1ContainerIdContainerid2Resource该Container可使用的资源量当前仅支持内存3UserContainer所属用户4Security tokens安全令牌只有持有该令牌才可启动container5LocalResource运行Container所需的本地资源比如jar包、二进制文件、其他外部文件等。6ServiceData应用程序可能使用其他外部服务这些服务相关的数据通过该参数指定。7Environment启动container所需的环境变量8command启动container的命令监控ContainerApplicationMaster可以通过2种途径监控启动的Container使用ApplicationMasterProtocol.allocate向ResourceManager发送查询请求 使用ContainerManagementProtocol查询指定的ContainerId对应的Container的状态ApplicationMaster会不断重复前面的步骤直到所有任务运行成功此时它会发送FinishApplicationMasterRequest以告诉ResourceManage自己运行结束。distributedshell实例分析Distributedshell Client的入口main函数如下:publicstaticvoidmain(String[]args){...Client clientnewClient();boolean doRunclient.init(args);if(!doRun){System.exit(0);}resultclient.run();...}DistributedShell Client中最重要的是函数为run()该函数实现过程如下利用Hadoop RPC接口创建一个可以直接与ResourceManager交互的RPC client句柄applicationsManagerprivatevoidconnectToASM()throwsIOException{YarnConfigurationyarnConfnewYarnConfiguration(conf);InetSocketAddressrmAddressyarnConf.getSocketAddr(YarnConfiguration.RM_ADDRESS,YarnConfiguration.DEFAULT_RM_ADDRESS,YarnConfiguration.DEFAULT_RM_PORT);LOG.info(Connecting to ResourceManager at rmAddress);applicationsManager((ClientRMProtocol)rpc.getProxy(ClientRMProtocol.class,rmAddress,conf));}获取application id。与ResourceManager通信请求application idGetNewApplicationRequestrequestRecords.newRecord(GetNewApplicationRequest.class);GetNewApplicationResponseresponseapplicationsManager.getNewApplication(request);构造ContainerLaunchContext。构造一个用于运行ApplicationMaster的containercontainer相关信息被封装到ContainerLaunchContext对象中ContainerLaunchContext amContainer Records.newRecord(ContainerLaunchContext.class); //添加本地资源 //填充localResources amContainer.setLocalResources(localResources); //添加运行ApplicationMaster所需的环境变量 MapString, String env new HashMapString, String(); //填充env amContainer.setEnvironment(env); //添加启动ApplicationMaster的命令 //填充commands; amContainer.setCommands(commands); //设置ApplicationMaster所需的资源 amContainer.setResource(capability);构造ApplicationSubmissionContext。构造一个用于提交ApplicationMaster的ApplicationSubmissionContextApplicationSubmissionContextappContextRecords.newRecord(ApplicationSubmissionContext.class);//设置application id调用GetNewApplicationResponse#getApplicationId()appContext.setApplicationId(appId);//设置Application名称“DistributedShell”appContext.setApplicationName(appName);//设置前面创建的containerappContext.setAMContainerSpec(amContainer);//设置application的优先级默认是0pri.setPriority(amPriority);//设置application的所在队列默认是appContext.setQueue(amQueue);//设置application的所属用户默认是appContext.setUser(amUser);提交ApplicationMaster。将ApplicationMaster提交到ResourceManager上从而完成作业提交功能applicationsManager.submitApplication(appRequest);显示应用程序运行状态。为了让用户知道应用程序进度Client会每隔几秒在shell终端上打印一次应用程序运行状态while(true){Thread.sleep(1000);GetApplicationReportRequest reportRequestRecords.newRecord(GetApplicationReportRequest.class);reportRequest.setApplicationId(appId);GetApplicationReportResponse reportResponseapplicationsManager.getApplicationReport(reportRequest);ApplicationReport reportreportResponse.getApplicationReport();//打印report内容...YarnApplicationState statereport.getYarnApplicationState();FinalApplicationStatus dsStatusreport.getFinalApplicationStatus();if(YarnApplicationState.FINISHEDstate){if(FinalApplicationStatus.SUCCEEDEDdsStatus){returntrue;}else{returnfalse;}}elseif(YarnApplicationState.KILLEDstate||YarnApplicationState.FAILEDstate){returnfalse;}}Distributedshell ApplicationMaster源码分析ApplicationMaster由ResourceManager分配的一个container启用之后它与ResourceManager通信注册自己以告知自己所在的节点hostporttrackingurl客户端可通过该url直接查询AM运行状态等。RegisterApplicationMasterRequest appMasterRequestRecords.newRecord(RegisterApplicationMasterRequest.class);appMasterRequest.setApplicationAttemptId(appAttemptID);appMasterRequest.setHost(appMasterHostname);appMasterRequest.setRpcPort(appMasterRpcPort);appMasterRequest.setTrackingUrl(appMasterTrackingUrl);returnresourceManager.registerApplicationMaster(appMasterRequest);ApplicationMaster周期性向ResourceManager发送心跳信息以告知ResourceManager自己仍然活着这是通过周期性调用AMRMProtocol#allocate实现的。为了完成计算任务ApplicationMaster需向ResourceManage发送一个ResourceRequest描述对资源的需求包括container个数、期望资源所在的节点、需要的CPU和内存等而ResourceManager则为ApplicationMaster返回一个AllocateResponse结构以告知新分配到的container列表、运行完成的container列表和当前可用的资源量等信息。while(numCompletedContainers.get()numTotalContainers!appDone){Thread.sleep(1000);ListResourceRequestresourceReqnewArrayListResourceRequest();if(askCount0){ResourceRequestcontainerAsksetupContainerAskForRM(askCount);resourceReq.add(containerAsk);}//如果resourceReq为null则可看做心跳信息否则就是申请资源AMResponseamRespsendContainerAskToRM(resourceReq);}对于每个新分配到的containerApplicationMaster将创建一个ContainerLaunchContext对象该对象包含container id启动container所需环境、启动container命令然后与对应的节点通信以启动container。LaunchContainerRunnablerunnableLaunchContainernewLaunchContainerRunnable(allocatedContainer);//每个container由一个线程启动ThreadlaunchThreadnewThread(runnableLaunchContainer);launchThreads.add(launchThread);launchThread.start();ApplicationMaster通过AMRMProtocol#allocate获取各个container的运行状况一旦发现某个container失败了则会重新向ResourceManager发送资源请求以重新运行失败的container。作业运行失败后ApplicationMaster向ResourceManager发送FinishApplicationMasterRequest请求以告知自己运行结束。FinishApplicationMasterRequestfinishReqRecords.newRecord(FinishApplicationMasterRequest.class);finishReq.setAppAttemptId(appAttemptID);booleanisSuccesstrue;if(numFailedContainers.get()0){finishReq.setFinishApplicationStatus(FinalApplicationStatus.SUCCEEDED);}