Call Amazon EMR with Step Functions - AWS Step Functions

Call Amazon EMR with Step Functions

Step Functions can control certain AWS services directly from the Amazon States Language. For more information, see the following:

To integrate AWS 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.

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 EMR service integration is subject to the availability of EMR APIs. Please check the Amazon EMR documentation for limitations in special regions.

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 createCluster and createCluster.sync to work, you must have configured the necessary permissions to create the service-linked role AWSServiceRoleForEMRCleanup. For more information about this, including a statement you can add to your IAM permissions policy, see Using the Service-Linked Role for Amazon EMR.

runJobFlow createCluster uses the same request syntax as runJobFlow, except for the following:
  • The field Instances.KeepJobFlowAliveWhenNoSteps is mandatory, and must have the Boolean value TRUE.

  • The field Steps is not allowed.

  • The field Instances.InstanceFleets[index].Name should be provided and must be unique if the optional modifyInstanceFleetByName connector API is used.

  • The field Instances.InstanceGroups[index].Name should be provided and must be unique if the optional modifyInstanceGroupByName API is used.

Response is this:
{ "ClusterId": "string" }
Amazon EMR uses this:
{ "JobFlowId": "string" }
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:
{ "ClusterId": "string" }
Amazon EMR uses this:
{ "JobFlowIds": ["string"] }
terminateCluster

Shuts down a cluster (job flow).

terminateJobFlows Request uses this:
{ "ClusterId": "string" }
Amazon EMR uses this:
{ "JobFlowIds": ["string"] }
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.

addJobFlowSteps Request uses the key "ClusterId". Amazon EMR uses "JobFlowId". Request uses a single step.
{ "Step": <"StepConfig object"> }
Amazon EMR uses this:
{ "Steps": [<StepConfig objects>] }
Response is this:
{ "StepId": "string" }
Amazon EMR returns this:
{ "StepIds": [<strings>] }
addStep.sync

Adds a new step to a running cluster.

addJobFlowSteps The same as addStep, but waits for the step to complete.
cancelStep

Cancels a pending step in a running cluster.

cancelSteps Request uses this:
{ "StepId": "string" }
Amazon EMR uses this:
{ "StepIds": [<strings>] }
Response is this:
{ "CancelStepsInfo": <CancelStepsInfo object> }
Amazon EMR uses this:
{ "CancelStepsInfoList": [<CancelStepsInfo objects>] }
modifyInstanceFleetByName

Modifies the target On-Demand and target Spot capacities for the instance fleet with the specified InstanceFleetName.

modifyInstanceFleet Request is the same as for modifyInstanceFleet, except for the following:
  • The field Instance.InstanceFleetId is not allowed.

  • At runtime the InstanceFleetId is determined automatically by the service integration by calling ListInstanceFleets and parsing the result.

modifyInstanceGroupByName

Modifies the number of nodes and configuration settings of an instance group.

modifyInstanceGroups Request is this:
{ "ClusterId": "string", "InstanceGroup": <InstanceGroupModifyConfig object> }
Amazon EMR uses a list:
{ "ClusterId": ["string"], "InstanceGroups": [<InstanceGroupModifyConfig objects>] }

Within the InstanceGroupModifyConfig object, the field InstanceGroupId is not allowed.

A new field, InstanceGroupName, has been added. At runtime the InstanceGroupId is determined automatically by the service integration by calling ListInstanceGroups and parsing the result.

The following includes a Task state that creates a cluster.

"Create_Cluster": { "Type": "Task", "Resource": "arn:aws: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: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:states:::elasticmapreduce:addStep.sync", "Parameters": { "ClusterId.$": "$.ClusterId", "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: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: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: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::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 AWS services, see IAM Policies for Integrated Services.