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 はワーカーコンテナ用の正確なシェルコマンドを送信します。

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 にアプリケーションを送信する一般的な手順 #

コメント