《董的博客YARN编程实例》由会员分享，可在线阅读，更多相关《董的博客YARN编程实例（9页珍藏版）》请在金锄头文库上搜索。
1、董的博客 YARN 编程实例1. 概述本文介绍 YARN 自带的一个非常简单的应用程序编程实例 distributedshell ，他可以看做 YARN 编程中的“ hello world 它的主要功能是并行执行用户提供的 shell 命令或者 shell 脚 本。本文主要介绍 distributedshell 的实现方法。 Distributedshell 的源代码在文件夹 srchadoop-yarn-projecthadoop-yarnhadoop-yarn-applica tionshadoop-yarn-applications-distributedshell 下。 Distrib
2、utedshell 的实现完全与文章“如何编写 YARN 应用程 序”所描述的一般 YARN 应用程序的编写方法完全一致。2. Distributedshell 客户端源码分析Distributedshell Client 的入口 main 函数如下： public static void main(String args) Client client = new Client();boolean doRun = client.init(args);if (!doRun) System.exit(0);result = client.run();DistributedShell Client
3、中最重要的是函数为 run() ，该函数实 现过程如下： (1 )构造 RPC 句柄。利用 Hadoop RPC 接口创建一个可以直接与ResourceManager 交互的 RPC client 句柄 applicationsManager ：private void connectToASM() throws IOException YarnConfiguration yarnConf = new YarnConfiguration(conf);InetSocketAddress rmAddress = yarnConf.getSocketAddr( YarnConfiguration.R
4、M_ADDRESS,YarnConfiguration.DEFAULT_RM_ADDRESS, YarnConfiguration.DEFAULT_RM_PORT);LOG.info( “ Connecting to ResourceManager at”rmAddress);applicationsManager = (ClientRMProtocol) rpc.getProxy( ClientRMProtocol.class, rmAddress, conf);(2 )获取 application id 。与 ResourceManager 通信，请求 application id ：Ge
5、tNewApplicationRequest request = Records.newRecord(GetNewApplicationRequest.class); GetNewApplicationResponse response = applicationsManager.getNewApplication(request);对象(3 )构造 ContainerLaunchContext 。 构造一个用于运行 ApplicationMaster 的 container ， container 相关信息被封装到 ContainerLaunchContext 中：ContainerLaun
6、chContext amContainer = Records.newRecord(ContainerLaunchContext.class); /添加本地资源/填充 localResources amContainer.setLocalResources(localResources); /添加运行 ApplicationMaster 所需的环境变量 Map<String, String> env = new HashMap<String, String>();/填充 envamContainer.setEnvironment(env);/添加启动 Applicati
7、onMaster 的命令/填充 commands; amContainer.setCommands(commands);/设置 ApplicationMaster 所需的资源amContainer.setResource(capability);(4 )构造 ApplicationSubmissionContext 构造一个用于提交 ApplicationMaster 的 ApplicationSubmissionContext ： ApplicationSubmissionContext appContext = Records.newRecord(ApplicationSubmission
8、Context.class) /设置 application id ，调用 GetNewApplicationResponse#getApplicationId() appContext.setApplicationId(appId);/设置 Application 名称：“ DistributedShell ” appContext.setApplicationName(appName);/设置前面创建的 container appContext.setAMContainerSpec(amContainer);/设置 application 的优先级，默认是 0 pri.setPriorit
9、y(amPriority);/设置 application 的所在队列，默认是” appContext.setQueue(amQueue);/设置 application 的所属用户，默认是” appContext.setUser(amUser);(5 )提交 ApplicationMaster 。将 ApplicationMaster 提交到 ResourceManager 上，从而完 成作业提交功能： applicationsManager.submitApplication(appRequest);(6 ) 显示应用程序运行状态。 为了让用户知道应用程序进度， Client 会每隔几秒在
10、 shell 终端上打印一次应用程序运行状态：while (true) Thread.sleep(1000);GetApplicationReportRequest reportRequest =Records.newRecord(GetApplicationReportRequest.class); reportRequest.setApplicationId(appId);GetApplicationReportResponse reportResponse = applicationsManager.getApplicationReport(reportRequest)Applicati
11、onReport report = reportResponse.getApplicationReport();/打印 report 内容YarnApplicationState state = report.getYarnApplicationState(); FinalApplicationStatus dsStatus = report.getFinalApplicationStatus(); if (YarnApplicationState.FINISHED = state) if (FinalApplicationStatus.SUCCEEDED = dsStatus) return
12、 true; else return false; else if (YarnApplicationState.KILLED = state| YarnApplicationState.FAILED = state) return false;3. Distributedshell ApplicationMaster 源码分析 Distributedshell ApplicationMaster 的实现方法与“如何编写 YARN 应用程序”所描述的步骤完全一致，它的过程如下：步骤 1 ApplicationMaster 由 ResourceManager 分配的一个 container 启用，
13、之后，它与 ResourceManager 通信，注册 自己，以告知自己所在的节点 (host ：port )，trackingurl (客 户端可通过该 url 直接查询 AM 运行状态)等。 RegisterApplicationMasterRequest appMasterRequest = Records.newRecord(RegisterApplicationMasterRequest.cl ass);appMasterRequest.setApplicationAttemptId(appAttemptID); appMasterRequest.setHost(appMasterH
14、ostname); appMasterRequest.setRpcPort(appMasterRpcPort); appMasterRequest.setTrackingUrl(appMasterTrackingUrl); return resourceManager.registerApplicationMaster(appMasterRe quest);步骤 2 ApplicationMaster 周期性向 ResourceManager 发 送心跳信息，以告知 ResourceManager 自己仍然活着，这 是通过周期性调用 AMRMProtocol#allocate 实现的。步骤 3
15、 为了完成计算任务， ApplicationMaster 需向 ResourceManage 发送一个 ResourceRequest 描述对资源 的需求，包括 container 个数、期望资源所在的节点、需要 的 CPU 和内存等，而 ResourceManager 则为 ApplicationMaster 返回一个 AllocateResponse 结构以告知 新分配到的 container 列表、 运行完成的 container 列表和当 前可用的资源量等信息。while (numCompletedContainers.get() < numTotalContainers&& !appDone) Thread.sleep(1000);List<ResourceRequest> resourceReq = newArrayList<ResourceRequest>();if (askCount > 0) ResourceRequest containerAsk = setupContainerAskForRM(askCount); resourceReq.add(containerAsk);/如果 resourceReq 为 null ，则可看做心跳信息，否则就是 申请资源AMResponse amR
