在 YARN 上运行任意命令

向 YARN 提交应用的通用步骤 #

准备应用资源。 打包客户端和 ApplicationMaster 代码，然后确保 YARN 可以将其本地化。

java

FileSystem fs = FileSystem.get(conf);
Path stagingDir = new Path(fs.getHomeDirectory(), ".yarn-binary-demo/" + appId);
fs.mkdirs(stagingDir);

Map<String, LocalResource> amLocalResources = new HashMap<>();
Path appJarDst = new Path(stagingDir, YarnBinaryDemoSupport.APP_JAR_NAME);
amLocalResources.put(
        YarnBinaryDemoSupport.APP_JAR_NAME,
        YarnBinaryDemoSupport.copyToHdfsAsLocalResource(fs, new Path(appJar), appJarDst));

创建 YARN 客户端并分配新的应用 ID。 客户端连接 ResourceManager，并请求一个新的应用句柄。

java

YarnConfiguration conf = new YarnConfiguration();
YarnClient yarnClient = YarnClient.createYarnClient();
yarnClient.init(conf);
yarnClient.start();

YarnClientApplication app = yarnClient.createApplication();
ApplicationId appId = app.getNewApplicationResponse().getApplicationId();

构建 ApplicationMaster 启动命令。 这是 YARN 用于启动 AM 容器的确切 Java 命令。

java

StringBuilder amCommand = new StringBuilder();
amCommand.append(ApplicationConstants.Environment.JAVA_HOME.$$())
        .append("/bin/java -Xmx256m ")
        .append("demo.yarn.YarnBinaryApplicationMaster ")
        .append("--command ")
        .append(YarnBinaryDemoSupport.shellQuote(command))
        .append(" ");

将 AM 命令封装在 ContainerLaunchContext 中。 这会为 AM 汇总命令、资源和环境。

java

Map<String, String> env = new HashMap<>();
env.put("CLASSPATH", YarnBinaryDemoSupport.buildClasspath(conf));

ContainerLaunchContext amContext = ContainerLaunchContext.newInstance(
        amLocalResources,
        env,
        Collections.singletonList(amCommand.toString()),
        null,
        null,
        null
);

向 ResourceManager 提交应用。 提交内容包括 AM 规范、应用名称、队列和 AM 资源需求。

java

Resource amResource = Resource.newInstance(512, 1);
app.getApplicationSubmissionContext().setApplicationName("yarn-binary-demo");
app.getApplicationSubmissionContext().setAMContainerSpec(amContext);
app.getApplicationSubmissionContext().setResource(amResource);
app.getApplicationSubmissionContext().setQueue("default");

yarnClient.submitApplication(app.getApplicationSubmissionContext());

YARN 启动 ApplicationMaster 后注册它。 AM 在其容器内启动后，会向 ResourceManager 通告自身。

java

AMRMClient<AMRMClient.ContainerRequest> rmClient = AMRMClient.createAMRMClient();
rmClient.init(conf);
rmClient.start();

rmClient.registerApplicationMaster("", 0, "");

向 ResourceManager 请求工作容器。 AM 为实际工作负载请求资源。

java

Resource capability = Resource.newInstance(256, 1);
Priority priority = Priority.newInstance(0);
AMRMClient.ContainerRequest request =
        new AMRMClient.ContainerRequest(capability, null, null, priority);
rmClient.addContainerRequest(request);

等待容器分配。 AM 会持续轮询 RM，直到容器可用。

java

Container allocatedContainer = null;
while (allocatedContainer == null) {
    AllocateResponse response = rmClient.allocate(0.1f);
    for (Container container : response.getAllocatedContainers()) {
        allocatedContainer = container;
        break;
    }
}

请求所属的 NodeManager 启动实际负载。 此时 AM 会向工作容器发送确切的 shell 命令。

java

String containerCommand =
        "/bin/bash -lc " + YarnBinaryDemoSupport.shellQuote(payload)
                + " 1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout"
                + " 2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stderr";

ContainerLaunchContext launchContext = ContainerLaunchContext.newInstance(
        localResources,
        new HashMap<String, String>(),
        Collections.singletonList(containerCommand),
        null,
        null,
        null
);

nmClient.startContainer(allocatedContainer, launchContext);

监控完成情况并结束应用。 AM 监视容器结果，报告最终状态，然后正常退出。

java

while (!finished) {
    AllocateResponse response = rmClient.allocate(0.5f);
    for (ContainerStatus status : response.getCompletedContainersStatuses()) {
        if (status.getContainerId().equals(allocatedContainer.getId())) {
            exitCode = status.getExitStatus();
            diagnostics = status.getDiagnostics();
            finished = true;
            break;
        }
    }
}

FinalApplicationStatus finalStatus =
        exitCode == 0 ? FinalApplicationStatus.SUCCEEDED : FinalApplicationStatus.FAILED;
rmClient.unregisterApplicationMaster(finalStatus, diagnostics, "");

在 YARN 上运行任意命令

向 YARN 提交应用的通用步骤 #

评论