在 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 的一般步驟 #

評論