IT评测·应用市场-qidao123.com技术社区

标题: 【深入浅出 Yarn 架构与实现】3-3 Yarn Application Master 编写 [打印本页]
作者: 乌市泽哥    时间: 2022-11-18 20:32
标题: 【深入浅出 Yarn 架构与实现】3-3 Yarn Application Master 编写
本篇文章继续介绍 Yarn Application 中 ApplicationMaster 部分的编写方法。
一、Application Master 编写方法

上一节讲了 Client 提交任务给 RM 的全流程，RM 收到任务后，由 ApplicationsManager 向 NM 申请 Container，并根据 Client 提供的 ContainerLaunchContext 启动 ApplicationMaster。
本篇代码已上传 Github：
Github - MyApplicationMaster
一）整体流程


1&2、启动 NMClient 和 RMClient

在 AM 中需要分别启动 NMClient 和 RMClient 进行通信。
两个客户端中都注册了我们自定义的 eventHandler，将会在后面进行介绍。
在 amRMClient 中会定义 AM 向 RM 定时发送心跳的间隔。（在 RM 中会有心跳容忍时间，注意不要超过 RM 配置的时间）
  1. // logInformation();
  2. Configuration conf = new Configuration();
  4. // 1 create amRMClient
  5. // 第一个参数是心跳时间 ms
  6. amRMClient = AMRMClientAsync.createAMRMClientAsync(1000, new RMCallbackHandler());
  7. amRMClient.init(conf);
  8. amRMClient.start();
  10. // 2 Create nmClientAsync
  11. amNMClient = new NMClientAsyncImpl(new NMCallbackHandler());
  12. amNMClient.init(conf);
  13. amNMClient.start();
复制代码
3、向 RM 注册 ApplicationMaster
  1. // 3 register with RM and this will heart beating to RM
  2. RegisterApplicationMasterResponse response = amRMClient
  3.                 .registerApplicationMaster(NetUtils.getHostname(), -1, "");
复制代码
4、申请 Containers

首先需要从 response 中确认资源池剩余资源，然后再根据需求申请 container
  1. // 4 Request containers
  2. response.getContainersFromPreviousAttempts();
  4. // 4.1 check resource
  5. long maxMem = response.getMaximumResourceCapability().getMemorySize();
  6. int maxVCores = response.getMaximumResourceCapability().getVirtualCores();
  8. // 4.2 request containers base on avail resource
  9. for (int i = 0; i < numTotalContainers.get(); i++) {
  10.     ContainerRequest containerAsk = new ContainerRequest(
  11.             //100*10M + 1vcpu
  12.             Resource.newInstance(100, 1), null, null,
  13.             Priority.newInstance(0));
  14.     amRMClient.addContainerRequest(containerAsk);
  15. }
复制代码
5、运行任务

将在 RMCallbackHandler 中的 onContainersAllocated 回调函数中处理，并在其中调用 NMCallbackHandler 的方法，执行对应的 task。
（RMCallbackHandler、NMCallbackHandler将在后面进行详细介绍。）
  1. // RMCallbackHandler
  2. public void onContainersAllocated(List<Container> containers) {
  3.     for (Container c : containers) {
  4.         log.info("Container Allocated, id = " + c.getId() + ", containerNode = " + c.getNodeId());
  5.         // LaunchContainerTask 实现在下面
  6.         exeService.submit(new LaunchContainerTask(c));
  7.     }
  8. }
  10. private class LaunchContainerTask implements Runnable {
  11.     @Override
  12.     public void run() {
  13.         // ……
  14.         // 发送事件交给 nm 处理
  15.         amNMClient.startContainerAsync(container, ctx);
  16.     }
  17. }
复制代码
6、结束任务

当全部子任务完成后，需要做收尾工作，将 amNMClient 和 amRMClient 停止。
  1. while(numTotalContainers.get() != numCompletedContainers.get()){
  2.     try{
  3.         Thread.sleep(1000);
  4.         log.info("waitComplete" +
  5.                 ", numTotalContainers=" + numTotalContainers.get() +
  6.                 ", numCompletedConatiners=" + numCompletedContainers.get());
  7.     } catch (InterruptedException ex){}
  8. }
  9. log.info("ShutDown exeService Start");
  10. exeService.shutdown();
  11. log.info("ShutDown exeService Complete");
  12. amNMClient.stop();
  13. log.info("amNMClient stop Complete");
  14. amRMClient.unregisterApplicationMaster(FinalApplicationStatus.SUCCEEDED, "dummy Message", null);
  15. log.info("unregisterApplicationMaster Complete");
  16. amRMClient.stop();
  17. log.info("amRMClient stop Complete");
复制代码
二）NMClient 和 RMClient Callback Handler 编写

1、RMCallbackHandler

本质是个 eventHandler，对事件库不熟悉的同学可以翻之前的文章「2-3 Yarn 基础库 - 服务库与事件库」进行学习。
其会处理 Container 启动、停止、更新等事件。
收到不同的事件时，会执行相应的回调函数。这里仅给出两个函数的实现。
<blockquote>

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！



欢迎光临 IT评测·应用市场-qidao123.com技术社区 (https://dis.qidao123.com/) Powered by Discuz! X3.4