概述
Yarn是一个资源的管理系统，负责集群的管理及调度。如果想要一个应用运行在Yarn，需要编写两个组件：
-
Client（客户端）
- 客户端负责项目ResourceManager提交ApplicationMaster
- 查询应用的运行状态
-
ApplicationMaster
- 向ResourceManager申请资源（以Container的形式）
- 启动NodeManger上的Container
- 监控程序运行状态，并在失败是为其重新分配资源
编写一个Yarn Application涉及协议
- ApplicationClientProtocol：Client通过该协议提交Application到RM，查询Application的运行状态，杀死Application等
- ApplicationMasterProtocol：AM使用该协议向RM注册ApplicationMaster，申请资源，查询各个task的运行状况等
- ContainerManagementProtocol：AM使用该协议让NM去启动/撤销Container、查询Container状态等
客户端（Client）
如上图编写一个客户端程序，需要以上两个步骤：
- 通过ApplicationClientProtocol获取applicationId
- 通过ApplicationClientProtocol submitAppcation去提交Application
- 构造ApplicationSubmissionContext, 其包含application的Id，名称，所要提交的队列名，优先级，用户等
- 构造ContainerLaunchContext，其包含所要申请的资源信息（cpu，内存）jars，files等
ApplicationMaster
ApplicationMaster分为两部分：ApplicationMaster-ResouceManager， ApplicationMaster-NodeManager
1，ApplicationMaster-ResouceManager
1）主要接口
如上图，ApplicationMaster与ResouceManager主要涉及以上三个接口：
- 通过ApplicationMasterProtocol.registerApplicationMaster注册AM
- 如果1成功，会起一个线程池周期的调用通过ApplicationMasterProtocol.allocateContainer,获取该Application所需的全部资源
- 当Application执行完成后调用ApplicationMasterProtocol.finishApplicationMaster完成该App的执行
2）核心编程库
如上图AM和RM交互分为阻塞（AMRMCLientImpl）和非阻塞（AMRMCLientAsyncImpl）
非阻塞（AMRMCLientAsyncImpl）是基于阻塞（AMRMCLientImpl）实现，当AM触发任何操作，AMRMCLientAsyncImpl会将该操作封装成事件，通过专门的线程池去处理这些事件。但用使用非阻塞接口是需要AM实现AMRMCLientAsyncImpl.CallbackHandler接口
public interface CallbackHandler {
void onContainersCompleted(List<ContainerStatus> statuses);
void onContainersAllocated(List<Container> containers);
void onShutdownRequest();
void onNodesUpdated(List<NodeReport> updatedNodes);
float getProgress();
void onError(Throwable e);
}
2，ApplicationMaster-NodeManager
1) 主要接口
如上图，ApplicationMaster与NodeManager主要涉及以上三个接口：
- ContainerManangementProtocol.startContainer, 在NodeManage节点启动一个container
- ContainerManangementProtocol.getContainerStatus,周期的去产看已启动container的状态
- ContainerManangementProtocol.stopContainer，当task完成，或者失败等。调用该其停止Container
2）核心编程库
如上图AM和NM交互分为阻塞（NMCLientImpl）和非阻塞（NMCLientAsyncImpl）
非阻塞（NMCLientAsyncImpl）是基于阻塞（NMCLientImpl）实现，当AM触发任何操作，NMCLientAsyncImpl会将该操作封装成事件，通过专门的线程池去处理这些事件。但用使用非阻塞接口是需要AM实现NMCLientAsyncImpl.CallbackHandler接口
public static interface CallbackHandler {
void onContainerStarted(ContainerId containerId,
Map<String, ByteBuffer> allServiceResponse);
void onContainerStatusReceived(ContainerId containerId,
ContainerStatus containerStatus);
void onContainerStopped(ContainerId containerId);
void onStartContainerError(ContainerId containerId, Throwable t);
void onGetContainerStatusError(ContainerId containerId, Throwable t);
void onStopContainerError(ContainerId containerId, Throwable t);
}
Yarn应用实例
这里的应用实例，我们使用DistributedShell这个Yarn自带的demo。其功能就是并行执行一个shell脚本。
DistributedShell使用方式：
/bin/hadoop jar\
share/hadoop/yarn/hadoop-yarn-applications-distributedshell-*.jar\
org.apache.hadoop.yarn.applications.distributedshell.Client\
--jar share/hadoop/yarn/hadoop-yarn-applications-distributedshell-*.jar\
--shell_command ls\
--num_containers 10\
--container_memory 350\
--master_memory 350\
--priority 10
1. Client端核心代码
org.apache.hadoop.yarn.applications.distributedshell.Client.run()
// 获得application id
YarnClientApplication app = yarnClient.createApplication();
GetNewApplicationResponse appResponse = app.getNewApplicationResponse();
...
//构建ApplicationSubmissionContext给submit操作提供上下文
ApplicationSubmissionContext appContext = app.getApplicationSubmissionContext();
/**
* 构建ApplicationMaster的命令
* 命令如下：
* /bin/java -Xmx 350m \
* org.apache.hadoop.yarn.applications.distributedshell.ApplicationMaster \
* --container_memory 350 \
* --num_containers 10 \
* --priority 10 \
* --shell_command ls \
* 1> <LOG_DIR>/AppMaster.stdout \
* 2> <LOG_DIR>/AppMaster.stderr
*/
vargs.add("\"" + Environment.JAVA_HOME.$$() + "/bin/java\"");
// Set Xmx based on am memory size
vargs.add("-Xmx" + amMemory + "m");
// Set class name
vargs.add(appMasterMainClass);
...
ContainerLaunchContext amContainer = ContainerLaunchContext.newInstance(
localResources, env, commands, null, null, null);
...
appContext.setAMContainerSpec(amContainer);
...
//提交应用
yarnClient.submitApplication(appContext);
//监控 application
monitorApplication(appId);
2. ApplicationMaster端核心代码
org.apache.hadoop.yarn.applications.distributedshell.ApplicationMaster.run()
...
// 将自定义的RMCallbackHandler注入给AMRMClientAsync
AMRMClientAsync.AbstractCallbackHandler allocListener =
new RMCallbackHandler();
amRMClient = AMRMClientAsync.createAMRMClientAsync(1000, allocListener);
amRMClient.init(conf);
amRMClient.start();
...
// 将自定义的NMCallbackHandler注入给NMClientAsyncImpl
containerListener = createNMCallbackHandler();
nmClientAsync = new NMClientAsyncImpl(containerListener);
nmClientAsync.init(conf);
nmClientAsync.start();
...
// 注册ApplicationMaster
// 同时开启ApplicationMaster的heartbeat
RegisterApplicationMasterResponse response = amRMClient.registerApplicationMaster(appMasterHostname, appMasterRpcPort,
appMasterTrackingUrl, placementConstraintMap);
...
//去RM请求全部的所需资源
for (int i = 0; i < numTotalContainersToRequest; ++i) {
ContainerRequest containerAsk = setupContainerAskForRM();
amRMClient.addContainerRequest(containerAsk);
}
RMCallbackHandler和NMCallbackHandler具体实现请查看org.apache.hadoop.yarn.applications.distributedshell.ApplicationMaster中的实现
Unmanaged AM
在YARN中，一个ApplicationMaster需要占用一个Container，该Container可能位于任意一个NodeManager上，这给ApplicationMaster调试带来很大麻烦。为了解决该问题，YARN引入了一种新的ApplicationMaster—Unmanaged AM，这种AM运行在客户端，不再由ResourceManager启动和销毁。用户只需在提交应用程序时设置一个参数，YARN便允许用户将ApplicationMaster运行在客户端的一个单独进程中。
从而方便调试，具体实现请查看:org.apache.hadoop.yarn.applications.unmanagedamlauncher.UnmanagedAMLauncher
