Run Arbitrary Command on Yarn

Generic Steps to Submit an App to YARN #

1. Prepare application resources. Package the client and ApplicationMaster code, then make sure YARN can localize them.

   java Copy

   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));

2. Create a YARN client and allocate a new application ID. The client connects to the ResourceManager and asks for a fresh application handle.

   java Copy

   YarnConfiguration conf = new YarnConfiguration();
   YarnClient yarnClient = YarnClient.createYarnClient();
   yarnClient.init(conf);
   yarnClient.start();
   
   YarnClientApplication app = yarnClient.createApplication();
   ApplicationId appId = app.getNewApplicationResponse().getApplicationId();

3. Build the ApplicationMaster launch command. This is the exact Java command that YARN will use to start the AM container.

   java Copy

   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(" ");

4. Wrap the AM command in a ContainerLaunchContext. This bundles together the command, resources, and environment for the AM.

   java Copy

   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
   );

5. Submit the application to the ResourceManager. The submission includes the AM spec, app name, queue, and AM resource requirement.

   java Copy

   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());

6. Register the ApplicationMaster after YARN launches it. Once the AM starts inside its container, it announces itself to the ResourceManager.

   java Copy

   AMRMClient<AMRMClient.ContainerRequest> rmClient = AMRMClient.createAMRMClient();
   rmClient.init(conf);
   rmClient.start();
   
   rmClient.registerApplicationMaster("", 0, "");

7. Request a worker container from the ResourceManager. The AM asks for resources for the actual workload.

   java Copy

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

8. Wait for a container allocation. The AM polls the RM until a container becomes available.

   java Copy

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

9. Ask the owning NodeManager to launch the real payload. This is where the AM sends the exact shell command for the worker container.

   java Copy

   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);

10. Monitor completion and finish the application. The AM watches for the container result, reports final status, and exits cleanly.

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, "");