构建和运行YARN应用程序全解析

### 构建和运行 YARN 应用程序全解析 #### 1. YARN 基础与应用构建概述 在 YARN 中，NodeManager 负责从 HDFS 下载请求中指定的本地资源，如应用所需的库或分布式缓存中的文件，下载完成后启动容器进程。下面我们开始构建一个简单的 YARN 应用程序，该应用将启动一个容器来执行 `vmstat` Linux 命令。 #### 2. 构建 YARN 客户端 YARN 客户端的主要作用是与 ResourceManager 协商创建并启动 YARN 应用程序实例。在这个过程中，需要告知 ResourceManager 应用主程序（ApplicationMaster）的系统资源需求。客户端创建和提交 YARN 应用程序的步骤如下： - **创建 YARN 应用程序**： ```java YarnConfiguration conf = new YarnConfiguration(); YarnClient yarnClient = YarnClient.createYarnClient(); yarnClient.init(conf); yarnClient.start(); YarnClientApplication app = yarnClient.createApplication(); ``` 上述代码中，`createApplication` 方法会调用 ResourceManager，返回一个新的应用程序 ID。同时，`YarnClientApplication` 对象包含集群的相关信息，可用于预先确定容器资源属性。`YarnClient` 类有许多会触发对 ResourceManager 进行 RPC 调用的 API，部分方法如下： ```java package org.apache.hadoop.yarn.client.api; public abstract class YarnClient extends AbstractService { YarnClientApplication createApplication(); ApplicationId submitApplication(ApplicationSubmissionContext ctx); void killApplication(ApplicationId applicationId); ApplicationReport getApplicationReport(ApplicationId appId); List<ApplicationReport> getApplications(); YarnClusterMetrics getYarnClusterMetrics(); List<NodeReport> getNodeReports(NodeState... states); ... } ``` - **提交 YARN 应用程序**：提交 YARN 应用程序会在 YARN 集群的新容器中启动 ApplicationMaster。在提交之前，需要进行以下配置： - 应用程序名称 - 启动 ApplicationMaster 的命令、类路径和环境设置 - 应用程序所需的 JAR 文件、配置文件等 - ApplicationMaster 的资源需求（内存和 CPU） - 提交应用程序的调度队列和队列内的应用程序优先级 - 安全令牌 以下是启动基于 Java 的基本 ApplicationMaster 的代码： ```java ContainerLaunchContext container = Records.newRecord(ContainerLaunchContext.class); String amLaunchCmd = String.format( "$JAVA_HOME/bin/java -Xmx256M %s 1>%s/stdout 2>%s/stderr", ApplicationMaster.class.getName(), ApplicationConstants.LOG_DIR_EXPANSION_VAR, ApplicationConstants.LOG_DIR_EXPANSION_VAR); container.setCommands(Lists.newArrayList(amLaunchCmd)); String jar = ClassUtil.findContainingJar(Client.class); FileSystem fs = FileSystem.get(conf); Path src = new Path(jar); Path dest = new Path(fs.getHomeDirectory(), src.getName()); fs.copyFromLocalFile(src, dest); FileStatus jarStat = FileSystem.get(conf).getFileStatus(dest); LocalResource appMasterJar = Records.newRecord(LocalResource.class); appMasterJar.setResource(ConverterUtils.getYarnUrlFromPath(dest)); appMasterJar.setSize(jarStat.getLen()); appMasterJar.setTimestamp(jarStat.getModificationTime()); appMasterJar.setType(LocalResourceType.FILE); appMasterJar.setVisibility(LocalResourceVisibility.APPLICATION); container.setLocalResources( ImmutableMap.of("AppMaster.jar", appMasterJar)); Map<String, String> appMasterEnv = Maps.newHashMap(); for (String c : conf.getStrings( YarnConfiguration.YARN_APPLICATION_CLASSPATH, YarnConfiguration.DEFAULT_YARN_APPLICATION_CLASSPATH)) { Apps.addToEnvironment(appMasterEnv, Environment.CLASSPATH.name(), c.trim()); } Apps.addToEnvironment(appMasterEnv, Environment.CLASSPATH.name(), Environment.PWD.$() + File.separator + "*"); container.setEnvironment(appMasterEnv); Resource capability = Records.newRecord(Resource.class); capability.setMemory(256); capability.setVirtualCores(1); ApplicationSubmissionContext appContext = app.getApplicationSubmissionContext(); appContext.setApplicationName("basic-dshell"); appContext.setAMContainerSpec(container); appContext.setResource(capability); appContext.setQueue("default"); ApplicationId appId = appContext.getApplicationId(); yarnClient.submitApplication(appContext); ``` 所有发送到 ResourceManager 的容器请求都是异步处理的，因此 `submitApplication` 方法返回并不意味着 ApplicationMaster 已经启动并运行。要了解应用程序的状态，需要轮询 ResourceManager 获取应用程序状态。 - **等待 YARN 应用程序完成**：提交应用程序后，可以轮询 ResourceManager 获取 ApplicationMaster 的状态信息，包括应用程序的状态、ApplicationMaster 运行的主机、跟踪 URL 等。以下代码用于轮询直到 ApplicationMaster 完成： ```java ApplicationReport report = yarnClient.getApplicationReport(appId); YarnApplicationState state = report.getYarnApplicationState(); EnumSet<YarnApplicationState> terminalStates = EnumSet.of(YarnApplicationState.FINISHED, YarnApplicationState.KILLED, YarnApplicationState.FAILED); while (!terminalStates.contains(state)) { TimeUnit.SECONDS.sleep(1); report = yarnClient.getApplicationReport(appId); state = report.getYarnApplicationState(); } ``` #### 3. 构建 ApplicationMaster ApplicationMaster 是 YARN 应用程序的协调者，负责