【深入浅出 Yarn 架构与实现】3-3 Yarn Application Master 编写

一、Application Master icode9 编写方法

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

一）整体流程#

1&2、启动 NMClient 和 RMClient

在 AM 中需要分别启动 NMClient 和 RMClient 进行通信。
两个客户端中都注册了我们自定义的 eventHandler，将会在后面进行介绍。
在 amRMClient 中会定义 AM 向 RM 定时发送心跳的间隔。（在 RM 中会有心跳容忍时间，注意不要超过 RM 配置的时间）

// logInformation();
Configuration conf = new Configuration();

// 1 create amRMClient
// 第一个参数是心跳时间 ms
amRMClient = AMRMClientAsync.createAMRMClientAsync(1000, new RMCallbackHandler());
amRMClient.init(conf);
amRMClient.start();

// 2 Create nmClientAsync
amNMClient = new NMClientAsyncImpl(new NMCallbackHandler());
amNMClient.init(conf);
amNMClient.start();

3、向 RM 注册 ApplicationMaster

// 3 register with RM and this will heart beating to RM
RegisterApplicationMasterResponse response = amRMClient
                .registerApplicationMaster(NetUtils.getHostname(), -1, "");

4、申请 Containers

首先需要从 response 中确认资源池剩余资源，然后再根据需求申请 container

// 4 Request containers
response.getContainersFromPreviousAttempts();

// 4.1 check resource
long maxMem = response.getMaximumResourceCapability().getMemorySize();
int maxVCores = response.getMaximumResourceCapability().getVirtualCores();

// 4.2 request containers base on avail resource
for (int i = 0; i < numTotalContainers.get(); i++) {
    ContainerRequest containerAsk = new ContainerRequest(
            //100*10M + 1vcpu
            Resource.newInstance(100, 1), null, null,
            Priority.newInstance(0));
    amRMClient.addContainerRequest(containerAsk);
}

5、运行任务

将在 RMCallbackHandler 中的 onContainersAllocated 回调函数中处理，并在其中调用 NMCallbackHandler 的方法，执行对应的 task。
（RMCallbackHandler、NMCallbackHandler将在后面进行详细介绍。）

// RMCallbackHandler
public void onContainersAllocated(List<Container> containers) {
    for (Container c : containers) {
        log.info("Container Allocated, id = " + c.getId() + ", containerNode = " + c.getNodeId());
        // LaunchContainerTask 实现在下面
        exeService.submit(new LaunchContainerTask(c));
    }
}

private class LaunchContainerTask implements Runnable {
    @Override
    public void run() {
        // ……
        // 发送事件交给 nm 处理
        amNMClient.startContainerAsync(container, ctx);
    }
}

本站声明：
1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；

2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；

3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；

4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；

5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。