Create and manage Amazon EMR clusters with Step Functions
Learn how to integrate Amazon Step Functions with Amazon EMR using the provided Amazon EMR service integration APIs. The service integration APIs are similar to the corresponding Amazon EMR APIs, with some differences in the fields that are passed and in the responses that are returned.
To learn about integrating with Amazon services in Step Functions, see Integrating services and Passing parameters to a service API in Step Functions.
Key features of Optimized Amazon EMR integration
The Optimized Amazon EMR service integration has a customized set of APIs that wrap the underlying Amazon EMR APIs, described below. Because of this, it differs significantly from the Amazon EMR Amazon SDK service integration.
-
The Run a Job (.sync) integration pattern is supported.
Step Functions does not terminate an Amazon EMR cluster automatically if execution is stopped. If your state machine stops before your Amazon EMR cluster has terminated, your cluster may continue running indefinitely, and can accrue additional charges. To avoid this, ensure that any Amazon EMR cluster you create is terminated properly. For more information, see:
-
Control Cluster Termination in the Amazon EMR User Guide.
-
The Service Integration Patterns Run a Job (.sync) section.
Note
As of emr-5.28.0
, you can specify the parameter
StepConcurrencyLevel
when creating a cluster to allow multiple steps to
run in parallel on a single cluster. You can use the Step Functions Map
and
Parallel
states to submit work in parallel to the cluster.
The availability of Amazon EMR service integration is subject to the availability of Amazon EMR APIs. See Amazon EMR documentation for limitations in special regions.
Note
For integration with Amazon EMR, Step Functions has a hard-coded 60 seconds job polling frequency for the first 10 minutes and 300 seconds after that.
Supported Amazon EMR APIs
The following table describes the differences between each Amazon EMR service integration API and corresponding Amazon EMR APIs.
Amazon EMR Service Integration API | Corresponding EMR API | Differences |
---|---|---|
createCluster
Creates and starts running a cluster (job flow). Amazon EMR
is linked directly to a unique type of IAM role known as a
service-linked role. For |
runJobFlow | createCluster uses the same request syntax as runJobFlow, except for the following:
Amazon EMR uses this:
|
createCluster.sync
Creates and starts running a cluster (job flow). |
runJobFlow | The same as createCluster , but waits for the cluster to
reach the WAITING state. |
setClusterTerminationProtection
Locks a cluster (job flow) so the EC2 instances in the cluster cannot be terminated by user intervention, an API call, or a job-flow error. |
setTerminationProtection | Request uses
this:
Amazon EMR uses
this:
|
terminateCluster
Shuts down a cluster (job flow). |
terminateJobFlows | Request uses
this:
Amazon EMR uses
this:
|
terminateCluster.sync Shuts down a cluster (job flow). |
terminateJobFlows | The same as terminateCluster , but waits for the cluster to
terminate. |
addStep
Adds a new step to a running cluster. Optionally, you can also specify the
|
Request uses the key "ClusterId" . Amazon EMR uses
"JobFlowId" . Request uses a single
step.
Amazon EMR uses
this:
Response is
this:
Amazon EMR returns
this:
|
|
addStep.sync
Adds a new step to a running cluster. Optionally, you can also specify the
|
The same as addStep , but waits for the step to
complete. |
|
cancelStep
Cancels a pending step in a running cluster. |
cancelSteps | Request uses
this:
Amazon EMR uses
this:
Response is
this:
Amazon EMR uses
this:
|
modifyInstanceFleetByName
Modifies the target On-Demand and target Spot capacities for the
instance fleet with the specified
|
modifyInstanceFleet | Request is the same as for modifyInstanceFleet , except for
the following:
|
modifyInstanceGroupByName
Modifies the number of nodes and configuration settings of an instance group. |
modifyInstanceGroups | Request is
this:
Amazon EMR uses a list:
Within the A new field, |
Workflow example
The following includes a Task
state that creates a cluster.
"Create_Cluster": {
"Type": "Task",
"Resource": "arn:aws-cn:states:::elasticmapreduce:createCluster.sync",
"Parameters": {
"Name": "MyWorkflowCluster",
"VisibleToAllUsers": true,
"ReleaseLabel": "emr-5.28.0",
"Applications": [
{
"Name": "Hive"
}
],
"ServiceRole": "EMR_DefaultRole",
"JobFlowRole": "EMR_EC2_DefaultRole",
"LogUri": "s3n://aws-logs-123456789012-us-east-1/elasticmapreduce/",
"Instances": {
"KeepJobFlowAliveWhenNoSteps": true,
"InstanceFleets": [
{
"InstanceFleetType": "MASTER",
"Name": "MASTER",
"TargetOnDemandCapacity": 1,
"InstanceTypeConfigs": [
{
"InstanceType": "m4.xlarge"
}
]
},
{
"InstanceFleetType": "CORE",
"Name": "CORE",
"TargetOnDemandCapacity": 1,
"InstanceTypeConfigs": [
{
"InstanceType": "m4.xlarge"
}
]
}
]
}
},
"End": true
}
The following includes a Task
state that enables termination protection.
"Enable_Termination_Protection": {
"Type": "Task",
"Resource": "arn:aws-cn:states:::elasticmapreduce:setClusterTerminationProtection",
"Parameters": {
"ClusterId.$": "$.ClusterId",
"TerminationProtected": true
},
"End": true
}
The following includes a Task
state that submits a step to a cluster.
"Step_One": {
"Type": "Task",
"Resource": "arn:aws-cn:states:::elasticmapreduce:addStep.sync",
"Parameters": {
"ClusterId.$": "$.ClusterId",
"ExecutionRoleArn": "arn:aws-cn:iam::123456789012
:role/myEMR-execution-role
",
"Step": {
"Name": "The first step",
"ActionOnFailure": "CONTINUE",
"HadoopJarStep": {
"Jar": "command-runner.jar",
"Args": [
"hive-script",
"--run-hive-script",
"--args",
"-f",
"s3://<region>
.elasticmapreduce.samples/cloudfront/code/Hive_CloudFront.q",
"-d",
"INPUT=s3://<region>
.elasticmapreduce.samples",
"-d",
"OUTPUT=s3://<amzn-s3-demo-bucket>
/MyHiveQueryResults/"
]
}
}
},
"End": true
}
The following includes a Task
state that cancels a step.
"Cancel_Step_One": {
"Type": "Task",
"Resource": "arn:aws-cn:states:::elasticmapreduce:cancelStep",
"Parameters": {
"ClusterId.$": "$.ClusterId",
"StepId.$": "$.AddStepsResult.StepId"
},
"End": true
}
The following includes a Task
state that terminates a cluster.
"Terminate_Cluster": {
"Type": "Task",
"Resource": "arn:aws-cn:states:::elasticmapreduce:terminateCluster.sync",
"Parameters": {
"ClusterId.$": "$.ClusterId"
},
"End": true
}
The following includes a Task
state that scales a cluster up or down for an
instance group.
"ModifyInstanceGroupByName": {
"Type": "Task",
"Resource": "arn:aws-cn:states:::elasticmapreduce:modifyInstanceGroupByName",
"Parameters": {
"ClusterId": "j-1234567890123",
"InstanceGroupName": "MyCoreGroup",
"InstanceGroup": {
"InstanceCount": 8
}
},
"End": true
}
The following includes a Task
state that scales a cluster up or down for an
instance fleet.
"ModifyInstanceFleetByName": {
"Type": "Task",
"Resource": "arn:aws-cn:states:::elasticmapreduce:modifyInstanceFleetByName",
"Parameters": {
"ClusterId": "j-1234567890123",
"InstanceFleetName": "MyCoreFleet",
"InstanceFleet": {
"TargetOnDemandCapacity": 8,
"TargetSpotCapacity": 0
}
},
"End": true
}
IAM policies for calling Amazon EMR
The following example templates show how Amazon Step Functions generates IAM policies based on the resources in your state machine definition. For more information, see How Step Functions generates IAM policies for integrated services and Discover service integration patterns in Step Functions.
addStep
Static resources
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:AddJobFlowSteps",
"elasticmapreduce:DescribeStep",
"elasticmapreduce:CancelSteps"
],
"Resource": [
"arn:aws:elasticmapreduce:[[region]]:[[accountId]]:cluster/[[clusterId]]"
]
}
]
}
Dynamic resources
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:AddJobFlowSteps",
"elasticmapreduce:DescribeStep",
"elasticmapreduce:CancelSteps"
],
"Resource": "arn:aws-cn:elasticmapreduce:*:*:cluster/*"
}
]
}
cancelStep
Static resources
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "elasticmapreduce:CancelSteps",
"Resource": [
"arn:aws-cn:elasticmapreduce:[[region]]
:[[accountId]]
:cluster/[[clusterId]]
"
]
}
]
}
Dynamic resources
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "elasticmapreduce:CancelSteps",
"Resource": "arn:aws-cn:elasticmapreduce:*:*:cluster/*"
}
]
}
createCluster
Static resources
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:RunJobFlow",
"elasticmapreduce:DescribeCluster",
"elasticmapreduce:TerminateJobFlows"
],
"Resource": "*"
},
{
"Effect": "Allow",
"Action": "iam:PassRole",
"Resource": [
"arn:aws-cn:iam::{{account}}
:role/[[roleName]]
"
]
}
]
}
setClusterTerminationProtection
Static resources
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "elasticmapreduce:SetTerminationProtection",
"Resource": [
"arn:aws-cn:elasticmapreduce:[[region]]
:[[accountId]]
:cluster/[[clusterId]]
"
]
}
]
}
Dynamic resources
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "elasticmapreduce:SetTerminationProtection",
"Resource": "arn:aws-cn:elasticmapreduce:*:*:cluster/*"
}
]
}
modifyInstanceFleetByName
Static resources
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:ModifyInstanceFleet",
"elasticmapreduce:ListInstanceFleets"
],
"Resource": [
"arn:aws-cn:elasticmapreduce:[[region]]
:[[accountId]]
:cluster/[[clusterId]]
"
]
}
]
}
Dynamic resources
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:ModifyInstanceFleet",
"elasticmapreduce:ListInstanceFleets"
],
"Resource": "arn:aws-cn:elasticmapreduce:*:*:cluster/*"
}
]
}
modifyInstanceGroupByName
Static resources
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:ModifyInstanceGroups",
"elasticmapreduce:ListInstanceGroups"
],
"Resource": [
"arn:aws-cn:elasticmapreduce:[[region]]
:[[accountId]]
:cluster/[[clusterId]]
"
]
}
]
}
Dynamic resources
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:ModifyInstanceGroups",
"elasticmapreduce:ListInstanceGroups"
],
"Resource": "*"
}
]
}
terminateCluster
Static resources
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:TerminateJobFlows",
"elasticmapreduce:DescribeCluster"
],
"Resource": [
"arn:aws-cn:elasticmapreduce:[[region]]
:[[accountId]]
:cluster/[[clusterId]]
"
]
}
]
}
Dynamic resources
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:TerminateJobFlows",
"elasticmapreduce:DescribeCluster"
],
"Resource": "arn:aws-cn:elasticmapreduce:*:*:cluster/*"
}
]
}