本篇文章繼續介紹 Yarn Application 中 ApplicationMaster 部分的編寫方法。 一、Application Master 編寫方法 上一節講了 Client 提交任務給 RM 的全流程，RM 收到任務後，由 ApplicationsManager 向 NM 申請 Con ...
本篇文章繼續介紹 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 啟動、停止、更新等事件。
收到不同的事件時，會執行相應的回調函數。這裡僅給出兩個函數的實現。
