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 `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. Optionally, you can also specify the `ExecutionRoleArn` parameter while using this API.	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. Optionally, you can also specify the `ExecutionRoleArn` parameter while using this API.	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-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.

Javascript is disabled or is unavailable in your browser.

To use the Amazon Web Services Documentation, Javascript must be enabled. Please refer to your browser's Help pages for instructions.

SageMaker

Amazon EMR on EKS