Create a Elastic MapReduce (EMR) Cluster with the Java API

In this post we will learn how to start creating and managing clusters using the Java API provided by Amazon for its Elastic MapReduce service. To start with, we need to download and extract the Amazon Web Services SDK. At the creation time of this post, the version 1.8.3 has been used (be aware that different versions may differ in several things implemented in this post). After having downloaded, extracted, and included the correspondent libraries into the chosen IDE (NetBeans IDE 7.4 in this case), we are ready to start creating the cluster.

The creation of a cluster consists, as in the CLI mode, on specifying several parameters to be understood by Amazon. Let's keep on analysing each parameter one by one:

AMI version

The AMI version determines the Amazon Machine Image to be used, which also determines the Hadoop version included in the images. Remember that clusters will automatically install Hadoop in every chosen node, so it is important to choose wisely. The main recommendation is to choose an AMI that bundles a Hadoop version that fits best with your needs.

In my project I have used a Hadoop app developed using its newest (at this time) version 2.4.0, so my selection has been the AMI version 3.1.0.

Logging

Amazon provides a useful logging infrastructure that stores nodes information in the desired location. Logs provide very useful information that can be used to investigate the cluster's behaviour and obtain information about different errors. The most common practice is to gather all the logging files inside the Amazon's Simple Storage Service or S3.

Bootstrap actions

This parameters makes the user able to execute scripts of any kind before starting to run Hadoop applications. Bootstrap actions are commonly used to install third party applications and built in applications such as Ganglia. It is also appropriate to insert libraries to be later consumed by the user's applications.

Step actions

Step actions are jobs ran once the cluster is running. These consist mostly on custom jobs such as Hadoop or Spark applications. As it is explained, there are two types 

Instance Configuration

Amazon offers a wide range of instances to be used within EMR. It is important to create an instance configuration that determines the type and quantity of instances to be used. which needs to fit to the user requirements. The following link lists all the instance types supported by EMR. 

Creating the cluster

The code needed to start a cluster is rather short. We just need to configure the parameters described above, and with a single command the cluster will start immediately. However, in order to be able to use Amazon's services, we first need to create an EMR client that will handle all the correspondent connections with the server.

Creating an EMR client:

The SDK provides a possibility to create clients off all kind. In our case, we will be creating an EMR client as shown at the following code:

AWSCredentials credentials = new BasicAWSCredentials(accessKey, secretKey); 
//where accessKey and secretKey are the keys given to the user created in the
IAM web interface  

AmazonElasticMapReduceClient emrClient;
emrClient = new AmazonElasticMapReduceClient(credentials);  
   
emrClient.setEndpoint("elasticmapreduce.eu-west-1.amazonaws.com"); 

It is important to notice how an access key and a secret key must be first created in Amazon's IAM user management web interface. Store the keys properly (specially the secret key), since you wont be able to download it again from the web.

Configure bootstrap:

For this example we have chosen to install ganglia on a single bootstrap action. Amazon provides an install script that makes all the installation for you, so the user does not need to worry about it. It is enough to specify the location of the script to be used. The following code shows how to create a bootstrap action:

String scriptPath = "s3://elasticmapreduce/bootstrap-actions/install-ganglia";  

ScriptBootstrapActionConfig bootstrapScriptConfig = new ScriptBootstrapActionConfig();  

bootstrapScriptConfig.setPath(scriptPath);  

BootstrapActionConfig gangliaBootstrap = new BootstrapActionConfig;

gangliaBootstrap.setName("Ganglia installation");

gangliaBootstrap.setScriptBootstrapAction(bootstrapScriptConfig);  

Configure steps:

One of the most common step is to run a debugging application that serves to obtain information about the applications ran afterwards. The SDK already provides a command to specify some pre-defined steps that do not need to be configured. The following two pieces of code show how to create a step with a pre-defined and a custom application.

Pre-defined debugging application

StepConfig debugging = new StepConfig()
      .withName("Enable debugging")  
      .withActionOnFailure("TERMINATE_JOB_FLOW")  
      .withHadoopJarStep(stepFactory.newEnableDebuggingStep());  

Custom user application (java .jar)

HadoopJarStepConfig customConfig = new HadoopJarStepConfig()  
     .withJar("s3://mybucket/myfolder/HadoopApp.jar")  
     .withMainClass("HadoopApp.class")  
     .withArgs(args);

StepConfig stepConfig = new StepConfig()
     .withHadoopJarStep(customConfig)  
     .withName("HadoopApp")  
     .withActionOnFailure(ActionOnFailure.CONTINUE); 

Configure instances:

The instance configuration is a parameter that the user can customise as he wants. Several parameters can be modified, being the instance type and quantity the most common ones. There are two ways to configure instances. The first one is indicating the parameters directly to the SDK, while the other consist on creating an instance group that contains all the information. For the sake of simplicity, the following code has been made using the first method and indicating the parameters directly to the system. The other method may appear in a different post when covering how to resize running clusters.

 JobFlowInstanceConfig instanceConfiguration = new JobFlowInstancesConfig()  
       .withEc2KeyName(keyPair) // Where keyPair is our custom Key Pair name
       .withInstanceCount(10)
       .withMasterInstanceType(InstanceType.C3xlarge.toString())
       .withKeepJobFlowAliveWhenNoSteps(true);

Configure job flow:

The Job Flow is the parameter that packages all the configurations previously done. It is used to tell Amazon what type of cluster we want to be created. Check the code below to understand how this process is made.

RunJobFlowRequest runJobFlowRequest = new RunJobFlowRequest()
     .withAmiVersion("3.1.0") // Hadoop 2.4.0
     .withBootstrapActions(gangliaBootstrap)
     .withName("My Cluster")
     .withSteps(debugging, stepConfig)
     .withLogUri("s3://mybucket/logs")
     .withInstances(instanceConfiguration);

Run job flow (and start the cluster):

At this point, the cluster is already configured and ready to be launch in the EMR  infraestructure. The only remaining step is to tell amazon to start running the cluster using the command runJobFlow. The exact code is the following:

RunJobFlowResult runJobFlowResult = emrClient.runJobFlow(runJobFlowRequest);

If we check the EMR web console at our preferred browser, we will be able to see a new cluster has been created with all the introduced custom parameters.

As a final note, we will look how to shut down the cluster once we have finished running all of our step and tasks. Be aware that if the parameter withKeepJobFlowAliveWhenNoSteps() is set to false, the cluster will shut down automatically after having finished all the steps. 

The following code shows how to tell EMR to shut down the cluster immediately:

String jobFlowId = runJobFlowResult.getJobFlowId();

TerminateJobFlowsRequest terminate;
terminate = new TerminateJobFlowsRequest().withJobFlowIds(Arrays.asList(new String[] {jobFlowId});

emrClient.terminateJobFlows(terminate);

This has been all for this post. On the next one we will learn how to check the clusters' status in order to be able to manage the clusters properly. If you have found the information interesting please leave a comment and share this post with people who could be interested in reading it.