Call Amazon EMR with Step Functions
Step Functions can control certain Amazon services directly from the Amazon States Language. For more information about working with Amazon Step Functions and its integrations, see the following:
How the Optimized Amazon EMR integration is different than the Amazon EMR Amazon SDK 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. In addition, the Run a Job (.sync) integration pattern is supported.
To integrate Amazon Step Functions with Amazon EMR, you use 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.
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. Please check the 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.
The following table describes the differences between each service integration API and its corresponding Amazon EMR API.
Amazon EMR Service Integration APIs 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, |
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://<mybucket>
/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
}
For information on how to configure IAM when using Step Functions with other Amazon services, see IAM Policies for integrated services.