I have written a yarn client which asks for some containers from YARN
- one container for application master on which I run my application master
- application master asks from another container for running my application
the problem is even after my application fails on the second container the application is marked as successful.
Can anyone please help me figure out a way to mark my application as failed if my application fails on the container
Client.java
import java.io.File;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
import org.apache.hadoop.yarn.api.records.*;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.client.api.YarnClientApplication;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.util.Apps;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;
public class Client {
Configuration conf = new YarnConfiguration();
public void run(String[] args) throws Exception {
// final String command = args[0];
final Path jarPath = new Path(args[0]);
// Create yarnClient
YarnConfiguration conf = new YarnConfiguration();
YarnClient yarnClient = YarnClient.createYarnClient();
yarnClient.init(conf);
yarnClient.start();
// Create application via yarnClient
YarnClientApplication app = yarnClient.createApplication();
ApplicationSubmissionContext appContext =
app.getApplicationSubmissionContext();
ApplicationId appId = appContext.getApplicationId();
// Set up the container launch context for the application master
ContainerLaunchContext amContainer =
Records.newRecord(ContainerLaunchContext.class);
amContainer.setCommands(
Collections.singletonList(
"$JAVA_HOME/bin/java" +
" -Xmx256M" +
" com.paytm.lending.datamart.yarn.v2.ApplicationMaster" +
" " + appId.toString() +
" 1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout" +
" 2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stderr"
)
);
// Setup jar for ApplicationMaster
LocalResource appMasterJar = Records.newRecord(LocalResource.class);
setupAppMasterJar(jarPath, appMasterJar);
amContainer.setLocalResources(
Collections.singletonMap("simpleapp.jar", appMasterJar));
// Setup CLASSPATH for ApplicationMaster
Map<String, String> appMasterEnv = new HashMap<String, String>();
setupAppMasterEnv(appMasterEnv,jarPath.toUri().getRawPath());
amContainer.setEnvironment(appMasterEnv);
// Set up resource type requirements for ApplicationMaster
Resource capability = Records.newRecord(Resource.class);
capability.setMemorySize(256);
capability.setVirtualCores(1);
// Finally, set-up ApplicationSubmissionContext for the application
appContext.setApplicationName("simple-yarn-app"); // application name
appContext.setAMContainerSpec(amContainer);
appContext.setResource(capability);
appContext.setQueue("default"); // queue
// Submit application
System.out.println("Submitting application " + appId);
yarnClient.submitApplication(appContext);
ApplicationReport appReport = yarnClient.getApplicationReport(appId);
YarnApplicationState appState = appReport.getYarnApplicationState();
while (appState != YarnApplicationState.FINISHED &&
appState != YarnApplicationState.KILLED &&
appState != YarnApplicationState.FAILED) {
Thread.sleep(100);
appReport = yarnClient.getApplicationReport(appId);
appState = appReport.getYarnApplicationState();
}
System.out.println(
"Application " + appId + " finished with" +
" state " + appState +
" at " + appReport.getFinishTime());
}
private void setupAppMasterJar(Path jarPath, LocalResource appMasterJar) throws IOException {
FileStatus jarStat = FileSystem.get(conf).getFileStatus(jarPath);
appMasterJar.setResource(URL.fromPath(jarPath));
appMasterJar.setSize(jarStat.getLen());
appMasterJar.setTimestamp(jarStat.getModificationTime());
appMasterJar.setType(LocalResourceType.FILE);
appMasterJar.setVisibility(LocalResourceVisibility.PUBLIC);
}
private void setupAppMasterEnv(Map<String, String> appMasterEnv,String jarPath) {
for (String c : conf.getStrings(
YarnConfiguration.YARN_APPLICATION_CLASSPATH,
YarnConfiguration.DEFAULT_YARN_APPLICATION_CLASSPATH)) {
Apps.addToEnvironment(appMasterEnv, Environment.CLASSPATH.name(),
c.trim(),File.pathSeparator);
}
Apps.addToEnvironment(appMasterEnv,
Environment.CLASSPATH.name(),
Environment.PWD.$() + File.separator + "*",File.pathSeparator);
Apps.addToEnvironment(appMasterEnv,"JAR_PATH",jarPath, File.pathSeparator);
}
public static void main(String[] args) throws Exception {
Client c = new Client();
c.run(args);
}
}
ApplicationMaster.java
import java.io.File;
import java.util.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
import org.apache.hadoop.yarn.api.records.*;
import org.apache.hadoop.yarn.client.api.AMRMClient;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.client.api.NMClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.util.Apps;
import org.apache.hadoop.yarn.util.Records;
public class ApplicationMaster {
public static void main(String[] args) throws Exception {
final String appId = args[0];
// Initialize clients to ResourceManager and NodeManagers
Configuration conf = new YarnConfiguration();
AMRMClient<ContainerRequest> rmClient = AMRMClient.createAMRMClient();
rmClient.init(conf);
rmClient.start();
NMClient nmClient = NMClient.createNMClient();
nmClient.init(conf);
nmClient.start();
// Register with ResourceManager
System.out.println("registerApplicationMaster 0");
rmClient.registerApplicationMaster("", 0, "");
System.out.println("registerApplicationMaster 1");
// Priority for worker containers - priorities are intra-application
Priority priority = Records.newRecord(Priority.class);
priority.setPriority(0);
// Resource requirements for worker containers
Resource capability = Records.newRecord(Resource.class);
capability.setMemorySize(128);
capability.setVirtualCores(1);
// Make container requests to ResourceManager
for (int i = 0; i < 2; ++i) {
ContainerRequest containerAsk = new ContainerRequest(capability, null, null, priority);
System.out.println("Making resource request " + i);
rmClient.addContainerRequest(containerAsk);
}
// Obtain allocated containers, launch and check for responses
int responseId = 0;
int completedContainers = 0;
while (completedContainers < 2) {
AllocateResponse response = rmClient.allocate(responseId++);
System.out.println("response.getAllocatedContainers().size() "+response.getAllocatedContainers().size());
for (Container container : response.getAllocatedContainers()) {
List<String> commands = new ArrayList<String>(Arrays.asList("$JAVA_HOME/bin/java " +
" -Xmx256m " +
" com.paytm.lending.datamart.yarn.DummyApplication " +1 +" "+
" 1> " + "/var/log/hadoop-yarn/containers/"+appId+"/"+container.getId()+"/"+ApplicationConstants.STDOUT +
" 2> " + "/var/log/hadoop-yarn/containers/"+appId+"/"+container.getId()+"/"+ApplicationConstants.STDERR
,"exit 100"));
// List<String> commands = Collections.singletonList(
// "$JAVA_HOME/bin/java " +
// " -Xmx256m " +
// " com.paytm.lending.datamart.yarn.DummyApplication " +1 +" "+
// " 1> " + "/var/log/hadoop-yarn/containers/"+appId+"/"+container.getId()+"/"+ApplicationConstants.STDOUT +
// " 2> " + "/var/log/hadoop-yarn/containers/"+appId+"/"+container.getId()+"/"+ApplicationConstants.STDERR
// );
// Launch container by create ContainerLaunchContext
ContainerLaunchContext ctx =
Records.newRecord(ContainerLaunchContext.class);
Map<String,LocalResource> test = new HashMap<>();
LocalResource test1 = Records.newRecord(LocalResource.class);
FileStatus status = FileSystem.get(conf).getFileStatus(new Path(System.getenv("JAR_PATH")));
test1.setResource(URL.fromPath(status.getPath()));
test1.setSize(status.getLen());
test1.setTimestamp(status.getModificationTime());
test1.setType(LocalResourceType.FILE);
test1.setVisibility(LocalResourceVisibility.PUBLIC);
Map<String, String> env = new HashMap<String, String>();
for (String c : conf.getStrings(
YarnConfiguration.YARN_APPLICATION_CLASSPATH,
YarnConfiguration.DEFAULT_YARN_APPLICATION_CLASSPATH)) {
Apps.addToEnvironment(env, ApplicationConstants.Environment.CLASSPATH.name(),
c.trim(), File.pathSeparator);
}
Apps.addToEnvironment(env,
ApplicationConstants.Environment.CLASSPATH.name(),
ApplicationConstants.Environment.PWD.$() + File.separator + "*",File.pathSeparator);
test.put("simpleapp.jar",test1);
ctx.setLocalResources(test);
ctx.setEnvironment(env);
ctx.setCommands(commands);
System.out.println("New Launching container " + container + " with command "+commands);
nmClient.startContainer(container, ctx);
}
for (ContainerStatus status : response.getCompletedContainersStatuses()) {
++completedContainers;
System.out.println("Completed container " + status.getContainerId());
System.out.println("Container state "+ status.getState());
}
Thread.sleep(100);
}
// Un-register with ResourceManager
rmClient.unregisterApplicationMaster(
FinalApplicationStatus.SUCCEEDED, "", "");
}
}
DummyApplication.java
import com.amazonaws.regions.Regions
import com.amazonaws.services.s3.AmazonS3ClientBuilder
object DummyApplication extends App {
val s3Client = AmazonS3ClientBuilder.standard().withRegion(Regions.AP_SOUTH_1).build()
val bucketName = "lending-data-science"
val objectKey = s"27059/test${args(0)}.txt"
val objectContent = "This is the file content."
println(s"creating file with name: ${objectKey}")
s3Client.putObject(bucketName, objectKey, objectContent)
throw new RuntimeException("Dummy Exception")
}
I need a way using which I can fail my application if exception occur in DummyApplication.java class