【深入淺出 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 配置的時間）

// 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);
    }
}

6、結束任務

當全部子任務完成後，需要做收尾工作，將 amNMClient 和 amRMClient 停止。

while(numTotalContainers.get() != numCompletedContainers.get()){
    try{
        Thread.sleep(1000);
        log.info("waitComplete" +
                ", numTotalContainers=" + numTotalContainers.get() +
                ", numCompletedConatiners=" + numCompletedContainers.get());
    } catch (InterruptedException ex){}
}
log.info("ShutDown exeService Start");
exeService.shutdown();
log.info("ShutDown exeService Complete");
amNMClient.stop();
log.info("amNMClient stop Complete");
amRMClient.unregisterApplicationMaster(FinalApplicationStatus.SUCCEEDED, "dummy Message", null);
log.info("unregisterApplicationMaster Complete");
amRMClient.stop();
log.info("amRMClient stop Complete");

二）NMClient 和 RMClient Callback Handler 編寫

1、RMCallbackHandler

本質是個 eventHandler，對事件庫不熟悉的同學可以翻之前的文章「2-3 Yarn 基礎庫 – 服務庫與事件庫」進行學習。
其會處理 Container 啟動、停止、更新等事件。
收到不同的事件時，會執行相應的回調函數。這裡僅給出兩個函數的實現。

💡 思考：之前版本中（2.6之前）還是實現 CallbackHandler 接口，為何後面改為了抽象類？
A：對原接口有了擴展增加了方法 onContainersUpdated。推測是因為避免使用接口繼承。

private class RMCallbackHandler extends AMRMClientAsync.AbstractCallbackHandler {
    @Override
    public void onContainersCompleted(List<ContainerStatus> statuses) {
        for (ContainerStatus status : statuses) {
            log.info("Container completed: " + status.getContainerId().toString()
                    + " exitStatus=" + status.getExitStatus());
            if (status.getExitStatus() != 0) {
                log.error("Container return error status: " + status.getExitStatus());
                log.warn("Need rerun container!");
                // do something restart container
                continue;
            }
            ContainerId containerId = status.getContainerId();
            runningContainers.remove(containerId);
            numCompletedContainers.addAndGet(1);
        }
    }
    
    @Override
    // 這裡在 container 中啟動相應的 task
    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 {
    Container container;
    public LaunchContainerTask(Container container) {
        this.container = container;
    }
    
    @Override
    public void run() {
        LinkedList<String> commands = new LinkedList<>();
        commands.add("sleep " + sleepSeconds.addAndGet(1));
        ContainerLaunchContext ctx = ContainerLaunchContext.newInstance(null, null, commands, null, null, null);
        // 這裡去執行 amNMClient 的回調
        amNMClient.startContainerAsync(container, ctx);
    }
}

2、NMCallbackHandler

定義 nm container 需要執行的各種事件處理。

private class NMCallbackHandler extends NMClientAsync.AbstractCallbackHandler {
    @Override
    public void onContainerStarted(ContainerId containerId, Map<String, ByteBuffer> allServiceResponse) {
        log.info("Container Stared " + containerId.toString());
    }
    
    // ……

三）涉及的通信協議

AM 與 RM

AM 與 NM

二、小結

至此我們學習了編寫 Yarn Application 的整體流程和實現方法，相信各位同學對其有了更深的認識。之後可以從 hadoop 提供的 DistributedShell 入手，再到其他框架（Hive、Flink）等探究工業級框架是如何提交 Application 的。

參考文章：
Hadoop Doc: Writing an ApplicationMaster (AM)
《Hadoop 技術內幕 – 深入解析 Yarn 結構設計與實現原理》第四章

Tags: hadoop yarn 大數據深入淺出 Yarn 架構與實現