本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。
使用 Step Functions 调用 Amazon EMR
Step Functions 可以直接从 Amazon States Language (ASL) 控制某些 Amazon 服务。要了解更多信息,请参阅使用其他服务和将参数传递给服务 API。
优化的亚马逊 EMR 集成与亚马逊 EMR SDK 集成有何不同 Amazon
优化的 Amazon EMR 服务集成有一组自定义的 API,用于封装底层的 Amazon EMR API,如下所述。因此,它与 Amazon EMR Amazon SDK 服务集成有很大不同。此外,它还支持运行作业 (.sync) 集成模式。
要 Amazon Step Functions 与亚马逊 EMR 集成,您可以使用提供的亚马逊 EMR 服务集成 API。服务集成 API 与相应的 Amazon EMR API 相似,但在传递的字段和返回的响应方面有所不同。
如果执行停止,Step Functions 不会自动终止 Amazon EMR 集群。如果您的状态机在 Amazon EMR 集群终止之前停止,则您的集群可能会无限期地继续运行,并且可能会产生额外费用。为避免这种情况,请确保您创建的任何 Amazon EMR 集群都已正确终止。有关更多信息,请参阅:
-
在 Amazon EMR 用户指南中控制集群终止。
-
服务集成模式运行作业 (.sync) 部分。
注意
自 emr-5.28.0
起,您可以在创建集群时指定参数 StepConcurrencyLevel
,以允许在单个集群上并行运行多个步骤。您可以使用 Step Functions Map
和 Parallel
状态将工作并行提交到集群。
Amazon EMR 服务集成的可用性取决于 Amazon EMR API 的可用性。请查看 Amazon EMR 文档,了解特殊区域的限制。
注意
为了与 Amazon EMR 集成,Step Functions 在前 10 分钟具有硬编码的 60 秒作业轮询频率,10 分钟后为 300 秒作业轮询频率。
下表描述了每个服务集成 API 及其相应 Amazon EMR API 之间的差异。
Amazon EMR 服务集成 API 和相应的 Amazon EMR API | |||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
Amazon EMR 服务集成 API | 相应的 EMR API | 差异 | |||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
CreateCluster 创建并开始运行集群(作业流程)。 Amazon EMR 与一种独特类型的 IAM 角色(称为服务相关角色)直接关联。要使 |
runJobFlow | createCluster 使用与相同的请求语法 runJobFlow,但以下语法除外:
Amazon EMR 使用以下信息:
|
|||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
createCluster.sync 创建并开始运行集群(作业流程)。 |
runJobFlow | 与 createCluster 相同,但等待集群达到 WAITING 状态。 |
|||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
setClusterTermination保护 锁定集群(作业流程),以使集群中的 EC2 实例无法通过用户干预、API 调用或作业流程错误终止。 |
setTerminationProtection | 请求使用: Amazon EMR 使用以下信息:
|
|||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
terminateCluster 关闭集群(作业流程)。 |
terminateJobFlows | 请求使用: Amazon EMR 使用以下信息:
|
|||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
terminateCluster.sync 关闭集群(作业流程)。 |
terminateJobFlows | 与 terminateCluster 相同,但等待集群终止。 |
|||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
addStep 向正在运行的集群添加新步骤。 另外,使用此 API 时,还能指定 |
请求使用密钥 "ClusterId" 。Amazon EMR 使用 "JobFlowId" 。请求使用单一步骤。 Amazon EMR 使用以下信息: 响应如下: Amazon EMR 返回以下内容:
|
||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
addStep.sync 向正在运行的集群添加新步骤。 另外,使用此 API 时,还能指定 |
与 addStep 相同,但等待步骤完成。 |
||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
cancelStep 取消正在运行的集群中的一个待处理步骤。 |
cancelSteps | 请求使用: Amazon EMR 使用以下信息: 响应如下: Amazon EMR 使用以下信息:
|
|||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
modifyInstanceFleetByName
使用指定的 |
modifyInstanceFleet | 请求与 modifyInstanceFleet 相同,但以下情况除外:
|
|||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||||
modifyInstanceGroupByName
修改实例组的节点数和配置设置。 |
modifyInstanceGroups | 请求如下: Amazon EMR 使用以下列表:
在 已添加一个新字段 |
以下内容包含一个创建集群的 Task
状态。
"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
}
以下内容包括启用终止保护的 Task
状态。
"Enable_Termination_Protection": {
"Type": "Task",
"Resource": "arn:aws:states:::elasticmapreduce:setClusterTerminationProtection",
"Parameters": {
"ClusterId.$": "$.ClusterId",
"TerminationProtected": true
},
"End": true
}
以下内容包括向集群提交步骤的 Task
状态。
"Step_One": {
"Type": "Task",
"Resource": "arn:aws:states:::elasticmapreduce:addStep.sync",
"Parameters": {
"ClusterId.$": "$.ClusterId",
"ExecutionRoleArn": "arn:aws: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
}
以下内容包括取消步骤的 Task
状态。
"Cancel_Step_One": {
"Type": "Task",
"Resource": "arn:aws:states:::elasticmapreduce:cancelStep",
"Parameters": {
"ClusterId.$": "$.ClusterId",
"StepId.$": "$.AddStepsResult.StepId"
},
"End": true
}
以下内容包括终止集群的 Task
状态。
"Terminate_Cluster": {
"Type": "Task",
"Resource": "arn:aws:states:::elasticmapreduce:terminateCluster.sync",
"Parameters": {
"ClusterId.$": "$.ClusterId"
},
"End": true
}
以下内容包括为实例组向上或向下扩展集群的 Task
状态。
"ModifyInstanceGroupByName": {
"Type": "Task",
"Resource": "arn:aws:states:::elasticmapreduce:modifyInstanceGroupByName",
"Parameters": {
"ClusterId": "j-1234567890123",
"InstanceGroupName": "MyCoreGroup",
"InstanceGroup": {
"InstanceCount": 8
}
},
"End": true
}
以下内容包括为实例队列向上或向下扩展集群的 Task
状态。
"ModifyInstanceFleetByName": {
"Type": "Task",
"Resource": "arn:aws:states:::elasticmapreduce:modifyInstanceFleetByName",
"Parameters": {
"ClusterId": "j-1234567890123",
"InstanceFleetName": "MyCoreFleet",
"InstanceFleet": {
"TargetOnDemandCapacity": 8,
"TargetSpotCapacity": 0
}
},
"End": true
}
有关在与其他 Amazon 服务Step Functions一起使用时如何配置IAM权限的信息,请参阅集成服务的 IAM 策略。