Amazon EMR - how to set a timeout for a step

9
votes

is there a way to set a timeout for a step in Amazon Aws EMR?

I'm running a batch Apache Spark job on EMR and I would like the job to stop with a timeout if it doesn't end within 3 hours.

I cannot find a way to set a timeout not in Spark, nor in Yarn, nor in EMR configuration.

Thanks for your help!

2
apache-sparkhadoop-yarnemramazon-emr
To my knowledge you cannot set a timeout. But you can always create a Lambda function that will kill your job or cluster if it runs for longer than 3 hours ;-) - Glennie Helles Sindholt
Thanks! That's what I thought, but I was hoping to find an easier way just setting some configuration parameter either during the creation of the step or directly in YARN. - nicola
@GlennieHellesSindholt, which is the easiest way to kill a step from a lambda function ? (I just want to kill the step but I DON'T want to terminate the full cluster) - nicola
Lambda supports a variety of programming languages (node.js, Java, Scala, Python, C#) and by including the EMR SDK, you can do whatever you want with the cluster (see the documentation for details and examples). Once you have written a Lambda that can check the running time and terminate a step, you schedule the Lambda function to run at an interval that makes sense depending on how often you run jobs that need to be monitored. If you only run them on Wednesday, schedule the Lambda to run only on Wednesdays. If you run them all the time, schedule the Lambda to run every 5 minutes or so. - Glennie Helles Sindholt
EMR add step needs a timeout argument, the rest is just complex workarounds. I hope somebody from AWS EMR is listening. - gae123

2 Answers

1
votes

I would like to offer an alternative approach, without any timeout/shutdown logic making application itself more complex than needed - although I am obviously quite late to the party. Maybe it proves useful for someone in the future.

You can:

  • write a Python script and use it as a wrapper around regular Yarn commands
  • execute those Yarn commands via subprocess lib
  • parse their output according to your will
  • decide which Yarn applications should be killed

More details about what I am talking about follow...

Python wrapper script and running the Yarn commands via subprocess lib

import subprocess

running_apps = subprocess.check_output(['yarn', 'application', '--list', '--appStates', 'RUNNING'], universal_newlines=True)

This snippet would give you an output similar to something like this:

Total number of applications (application-types: [] and states: [RUNNING]):1
                Application-Id      Application-Name                                Application-Type          User       Queue               State         Final-State         Progress                        Tracking-URL
application_1554703852869_0066      HIVE-645b9a64-cb51-471b-9a98-85649ee4b86f       TEZ                       hadoop     default             RUNNING       UNDEFINED           0%                              http://ip-xx-xxx-xxx-xx.eu-west-1.compute.internal:45941/ui/

You can than parse this output (beware there might be more than one app running) and extract application-id values.

Then, for each of those application ids, you can invoke another yarn command to get more details about the specific application:

app_status_string = subprocess.check_output(['yarn', 'application', '--status', app_id], universal_newlines=True)

Output of this command should be something like this:

Application Report :
  Application-Id : application_1554703852869_0070
  Application-Name : com.organization.YourApp
  Application-Type : HIVE
  User : hadoop
  Queue : default
  Application Priority : 0
  Start-Time : 1554718311926
  Finish-Time : 0
  Progress : 10%
  State : RUNNING
  Final-State : UNDEFINED
  Tracking-URL : http://ip-xx-xxx-xxx-xx.eu-west-1.compute.internal:40817
  RPC Port : 36203
  AM Host : ip-xx-xxx-xxx-xx.eu-west-1.compute.internal
  Aggregate Resource Allocation : 51134436 MB-seconds, 9284 vcore-seconds
  Aggregate Resource Preempted : 0 MB-seconds, 0 vcore-seconds
  Log Aggregation Status : NOT_START
  Diagnostics :
  Unmanaged Application : false
  Application Node Label Expression : <Not set>
  AM container Node Label Expression : CORE

Having this you can also extract application's start time, compare it with current time and see for how long it is running. If it is running for more than some threshold number of minutes, for example you kill it.

How do you kill it? Easy.

kill_output = subprocess.check_output(['yarn', 'application', '--kill', app_id], universal_newlines=True)

This should be it, from the killing of the step/application perspective.

Automating the approach

AWS EMR has a wonderful feature called "bootstrap actions". It runs a set of actions on EMR cluster creation and can be utilized for automating this approach.

Add a bash script to bootstrap actions which is going to:

  • download the python script you just wrote to the cluster (master node)
  • add the python script to a crontab

That should be it.

P.S. I assumed Python3 is at our disposal for this purpose.

0
votes

Well, as many have already answered, an EMR step cannot be killed/stopped/terminated via an API call at this moment.

But to achieve your goals, you can introduce a timeout as part of your application code itself. When you submit EMR steps, a child process is created to run your application - be it MapReduce Application, Spark Application, etc. and the step completion is determined by the exit code this child process (which is your application) returns.

For example, if you are submitting a MapReduce Application, you can use something like below :

FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));

final Runnable stuffToDo = new Thread() {
  @Override 
  public void run() { 
    job.submit();
  }
};

final ExecutorService executor = Executors.newSingleThreadExecutor();
final Future future = executor.submit(stuffToDo);
executor.shutdown(); // This does not cancel the already-scheduled task.

try { 
  future.get(180, TimeUnit.MINUTES); 
}
catch (InterruptedException ie) { 
  /* Handle the interruption. Or ignore it. */ 
}
catch (ExecutionException ee) { 
  /* Handle the error. Or ignore it. */ 
}
catch (TimeoutException te) { 
  /* Handle the timeout. Or ignore it. */ 
}
System.exit(job.waitForCompletion(true) ? 0 : 1);

Reference - Java: set timeout on a certain block of code?.

Hope this helps.