There are more AWS SDK examples available in the [AWS Doc SDK Examples](https://github.com/awsdocs/aws-doc-sdk-examples) GitHub repo.
# Amazon EMR examples using SDK for Python (Boto3)
The following code examples show you how to perform actions and implement common scenarios by using the AWS SDK for Python (Boto3) with Amazon EMR.
*Actions* are code excerpts from larger programs and must be run in context. While actions show you how to call individual service functions, you can see actions in context in their related scenarios.
*Scenarios* are code examples that show you how to accomplish specific tasks by calling multiple functions within a service or combined with other AWS services.
Each example includes a link to the complete source code, where you can find instructions on how to set up and run the code in context.
**Topics**
+ [Actions](#actions)
+ [Scenarios](#scenarios)
## Actions
### `AddJobFlowSteps`
The following code example shows how to use `AddJobFlowSteps`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/emr#code-examples).
Add a Spark step, which is run by the cluster as soon as it is added.
```
def add_step(cluster_id, name, script_uri, script_args, emr_client):
"""
Adds a job step to the specified cluster. This example adds a Spark
step, which is run by the cluster as soon as it is added.
:param cluster_id: The ID of the cluster.
:param name: The name of the step.
:param script_uri: The URI where the Python script is stored.
:param script_args: Arguments to pass to the Python script.
:param emr_client: The Boto3 EMR client object.
:return: The ID of the newly added step.
"""
try:
response = emr_client.add_job_flow_steps(
JobFlowId=cluster_id,
Steps=[
{
"Name": name,
"ActionOnFailure": "CONTINUE",
"HadoopJarStep": {
"Jar": "command-runner.jar",
"Args": [
"spark-submit",
"--deploy-mode",
"cluster",
script_uri,
*script_args,
],
},
}
],
)
step_id = response["StepIds"][0]
logger.info("Started step with ID %s", step_id)
except ClientError:
logger.exception("Couldn't start step %s with URI %s.", name, script_uri)
raise
else:
return step_id
```
Run an Amazon EMR File System (EMRFS) command as a job step on a cluster. This can be used to automate EMRFS commands on a cluster instead of running commands manually through an SSH connection.
```
import boto3
from botocore.exceptions import ClientError
def add_emrfs_step(command, bucket_url, cluster_id, emr_client):
"""
Add an EMRFS command as a job flow step to an existing cluster.
:param command: The EMRFS command to run.
:param bucket_url: The URL of a bucket that contains tracking metadata.
:param cluster_id: The ID of the cluster to update.
:param emr_client: The Boto3 Amazon EMR client object.
:return: The ID of the added job flow step. Status can be tracked by calling
the emr_client.describe_step() function.
"""
job_flow_step = {
"Name": "Example EMRFS Command Step",
"ActionOnFailure": "CONTINUE",
"HadoopJarStep": {
"Jar": "command-runner.jar",
"Args": ["/usr/bin/emrfs", command, bucket_url],
},
}
try:
response = emr_client.add_job_flow_steps(
JobFlowId=cluster_id, Steps=[job_flow_step]
)
step_id = response["StepIds"][0]
print(f"Added step {step_id} to cluster {cluster_id}.")
except ClientError:
print(f"Couldn't add a step to cluster {cluster_id}.")
raise
else:
return step_id
def usage_demo():
emr_client = boto3.client("emr")
# Assumes the first waiting cluster has EMRFS enabled and has created metadata
# with the default name of 'EmrFSMetadata'.
cluster = emr_client.list_clusters(ClusterStates=["WAITING"])["Clusters"][0]
add_emrfs_step(
"sync", "s3://elasticmapreduce/samples/cloudfront", cluster["Id"], emr_client
)
if __name__ == "__main__":
usage_demo()
```
+ For API details, see [AddJobFlowSteps](https://docs.aws.amazon.com/goto/boto3/elasticmapreduce-2009-03-31/AddJobFlowSteps) in *AWS SDK for Python (Boto3) API Reference*.
### `DescribeCluster`
The following code example shows how to use `DescribeCluster`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/emr#code-examples).
```
def describe_cluster(cluster_id, emr_client):
"""
Gets detailed information about a cluster.
:param cluster_id: The ID of the cluster to describe.
:param emr_client: The Boto3 EMR client object.
:return: The retrieved cluster information.
"""
try:
response = emr_client.describe_cluster(ClusterId=cluster_id)
cluster = response["Cluster"]
logger.info("Got data for cluster %s.", cluster["Name"])
except ClientError:
logger.exception("Couldn't get data for cluster %s.", cluster_id)
raise
else:
return cluster
```
+ For API details, see [DescribeCluster](https://docs.aws.amazon.com/goto/boto3/elasticmapreduce-2009-03-31/DescribeCluster) in *AWS SDK for Python (Boto3) API Reference*.
### `DescribeStep`
The following code example shows how to use `DescribeStep`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/emr#code-examples).
```
def describe_step(cluster_id, step_id, emr_client):
"""
Gets detailed information about the specified step, including the current state of
the step.
:param cluster_id: The ID of the cluster.
:param step_id: The ID of the step.
:param emr_client: The Boto3 EMR client object.
:return: The retrieved information about the specified step.
"""
try:
response = emr_client.describe_step(ClusterId=cluster_id, StepId=step_id)
step = response["Step"]
logger.info("Got data for step %s.", step_id)
except ClientError:
logger.exception("Couldn't get data for step %s.", step_id)
raise
else:
return step
```
+ For API details, see [DescribeStep](https://docs.aws.amazon.com/goto/boto3/elasticmapreduce-2009-03-31/DescribeStep) in *AWS SDK for Python (Boto3) API Reference*.
### `ListSteps`
The following code example shows how to use `ListSteps`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/emr#code-examples).
```
def list_steps(cluster_id, emr_client):
"""
Gets a list of steps for the specified cluster. In this example, all steps are
returned, including completed and failed steps.
:param cluster_id: The ID of the cluster.
:param emr_client: The Boto3 EMR client object.
:return: The list of steps for the specified cluster.
"""
try:
response = emr_client.list_steps(ClusterId=cluster_id)
steps = response["Steps"]
logger.info("Got %s steps for cluster %s.", len(steps), cluster_id)
except ClientError:
logger.exception("Couldn't get steps for cluster %s.", cluster_id)
raise
else:
return steps
```
+ For API details, see [ListSteps](https://docs.aws.amazon.com/goto/boto3/elasticmapreduce-2009-03-31/ListSteps) in *AWS SDK for Python (Boto3) API Reference*.
### `RunJobFlow`
The following code example shows how to use `RunJobFlow`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/emr#code-examples).
```
def run_job_flow(
name,
log_uri,
keep_alive,
applications,
job_flow_role,
service_role,
security_groups,
steps,
emr_client,
):
"""
Runs a job flow with the specified steps. A job flow creates a cluster of
instances and adds steps to be run on the cluster. Steps added to the cluster
are run as soon as the cluster is ready.
This example uses the 'emr-5.30.1' release. A list of recent releases can be
found here:
https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-release-components.html.
:param name: The name of the cluster.
:param log_uri: The URI where logs are stored. This can be an Amazon S3 bucket URL,
such as 's3://my-log-bucket'.
:param keep_alive: When True, the cluster is put into a Waiting state after all
steps are run. When False, the cluster terminates itself when
the step queue is empty.
:param applications: The applications to install on each instance in the cluster,
such as Hive or Spark.
:param job_flow_role: The IAM role assumed by the cluster.
:param service_role: The IAM role assumed by the service.
:param security_groups: The security groups to assign to the cluster instances.
Amazon EMR adds all needed rules to these groups, so
they can be empty if you require only the default rules.
:param steps: The job flow steps to add to the cluster. These are run in order
when the cluster is ready.
:param emr_client: The Boto3 EMR client object.
:return: The ID of the newly created cluster.
"""
try:
response = emr_client.run_job_flow(
Name=name,
LogUri=log_uri,
ReleaseLabel="emr-5.30.1",
Instances={
"MasterInstanceType": "m5.xlarge",
"SlaveInstanceType": "m5.xlarge",
"InstanceCount": 3,
"KeepJobFlowAliveWhenNoSteps": keep_alive,
"EmrManagedMasterSecurityGroup": security_groups["manager"].id,
"EmrManagedSlaveSecurityGroup": security_groups["worker"].id,
},
Steps=[
{
"Name": step["name"],
"ActionOnFailure": "CONTINUE",
"HadoopJarStep": {
"Jar": "command-runner.jar",
"Args": [
"spark-submit",
"--deploy-mode",
"cluster",
step["script_uri"],
*step["script_args"],
],
},
}
for step in steps
],
Applications=[{"Name": app} for app in applications],
JobFlowRole=job_flow_role.name,
ServiceRole=service_role.name,
EbsRootVolumeSize=10,
VisibleToAllUsers=True,
)
cluster_id = response["JobFlowId"]
logger.info("Created cluster %s.", cluster_id)
except ClientError:
logger.exception("Couldn't create cluster.")
raise
else:
return cluster_id
```
+ For API details, see [RunJobFlow](https://docs.aws.amazon.com/goto/boto3/elasticmapreduce-2009-03-31/RunJobFlow) in *AWS SDK for Python (Boto3) API Reference*.
### `TerminateJobFlows`
The following code example shows how to use `TerminateJobFlows`.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/emr#code-examples).
```
def terminate_cluster(cluster_id, emr_client):
"""
Terminates a cluster. This terminates all instances in the cluster and cannot
be undone. Any data not saved elsewhere, such as in an Amazon S3 bucket, is lost.
:param cluster_id: The ID of the cluster to terminate.
:param emr_client: The Boto3 EMR client object.
"""
try:
emr_client.terminate_job_flows(JobFlowIds=[cluster_id])
logger.info("Terminated cluster %s.", cluster_id)
except ClientError:
logger.exception("Couldn't terminate cluster %s.", cluster_id)
raise
```
+ For API details, see [TerminateJobFlows](https://docs.aws.amazon.com/goto/boto3/elasticmapreduce-2009-03-31/TerminateJobFlows) in *AWS SDK for Python (Boto3) API Reference*.
## Scenarios
### Create a short-lived Amazon EMR cluster and run a step
The following code example shows how to create a short-lived Amazon EMR cluster that runs a step and automatically terminates after the step completes.
**SDK for Python (Boto3)**
Create a short-lived Amazon EMR cluster that estimates the value of pi using Apache Spark to parallelize a large number of calculations. The job writes output to Amazon EMR logs and to an Amazon Simple Storage Service (Amazon S3) bucket. The cluster terminates itself after completing the job.
+ Create an Amazon S3 bucket and upload a job script.
+ Create AWS Identity and Access Management (IAM) roles.
+ Create Amazon Elastic Compute Cloud (Amazon EC2) security groups.
+ Create a short-lived cluster and run a single job step.
This example is best viewed on GitHub. For complete source code and instructions on how to set up and run, see the full example on [GitHub](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/emr).
**Services used in this example**
+ Amazon EMR
### Run a shell script to install libraries
The following code example shows how to use AWS Systems Manager to run a shell script on Amazon EMR instances that installs additional libraries. This way, you can automate instance management instead of running commands manually through an SSH connection.
**SDK for Python (Boto3)**
There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/emr#code-examples).
```
import argparse
import time
import boto3
def install_libraries_on_core_nodes(cluster_id, script_path, emr_client, ssm_client):
"""
Copies and runs a shell script on the core nodes in the cluster.
:param cluster_id: The ID of the cluster.
:param script_path: The path to the script, typically an Amazon S3 object URL.
:param emr_client: The Boto3 Amazon EMR client.
:param ssm_client: The Boto3 AWS Systems Manager client.
"""
core_nodes = emr_client.list_instances(
ClusterId=cluster_id, InstanceGroupTypes=["CORE"]
)["Instances"]
core_instance_ids = [node["Ec2InstanceId"] for node in core_nodes]
print(f"Found core instances: {core_instance_ids}.")
commands = [
# Copy the shell script from Amazon S3 to each node instance.
f"aws s3 cp {script_path} /home/hadoop",
# Run the shell script to install libraries on each node instance.
"bash /home/hadoop/install_libraries.sh",
]
for command in commands:
print(f"Sending '{command}' to core instances...")
command_id = ssm_client.send_command(
InstanceIds=core_instance_ids,
DocumentName="AWS-RunShellScript",
Parameters={"commands": [command]},
TimeoutSeconds=3600,
)["Command"]["CommandId"]
while True:
# Verify the previous step succeeded before running the next step.
cmd_result = ssm_client.list_commands(CommandId=command_id)["Commands"][0]
if cmd_result["StatusDetails"] == "Success":
print(f"Command succeeded.")
break
elif cmd_result["StatusDetails"] in ["Pending", "InProgress"]:
print(f"Command status is {cmd_result['StatusDetails']}, waiting...")
time.sleep(10)
else:
print(f"Command status is {cmd_result['StatusDetails']}, quitting.")
raise RuntimeError(
f"Command {command} failed to run. "
f"Details: {cmd_result['StatusDetails']}"
)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("cluster_id", help="The ID of the cluster.")
parser.add_argument("script_path", help="The path to the script in Amazon S3.")
args = parser.parse_args()
emr_client = boto3.client("emr")
ssm_client = boto3.client("ssm")
install_libraries_on_core_nodes(
args.cluster_id, args.script_path, emr_client, ssm_client
)
if __name__ == "__main__":
main()
```
+ For API details, see [ListInstances](https://docs.aws.amazon.com/goto/boto3/elasticmapreduce-2009-03-31/ListInstances) in *AWS SDK for Python (Boto3) API Reference*.