翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
Step Functions を使用した Amazon EMR クラスターの作成と管理
が提供する Amazon EMR サービス統合 APIs を使用して Amazon EMR AWS Step Functions と統合する方法について説明します。サービス統合 API は対応する Amazon EMR API に似ていますが、渡されるフィールドと返される応答にいくつかの違いがあります。
Step Functions での AWS サービスとの統合については、 サービスとの統合「」および「」を参照してくださいStep Functions でサービス API にパラメータを渡す。
最適化された Amazon EMR 統合の主な機能
最適化された Amazon EMR サービス統合には、以下で説明するような基になる Amazon EMR API をラップするカスタマイズされた一連の API があります。このため、Amazon EMR AWS SDK サービス統合とは大きく異なります。
-
ジョブの実行 (.sync) 統合パターンがサポートされています。
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 分とその後の 300 秒間、ジョブポーリング頻度をハードコーディングして 60 秒に設定しています。
最適化された Amazon EMR APIs
各 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 状態になるまで待機します。 |
setClusterTerminationProtection クラスター (ジョブフロー) をロックして、クラスター内の EC2 インスタンスをユーザーの介入、API コール、またはジョブフローエラーが発生した場合に終了できないようにします。 |
setTerminationProtection | リクエストは以下を使用します。 Amazon EMR は以下を使用します。
|
terminateCluster クラスター (ジョブフロー) をシャットダウンします。 |
terminateJobFlows | リクエストは以下を使用します。 Amazon EMR は以下を使用します。
|
terminateCluster.sync クラスター (ジョブフロー) をシャットダウンします。 |
terminateJobFlows | terminateCluster と同じですが、クラスターが終了するまで待機します。 |
addStep 実行中のクラスターに新しいステップを追加します。 オプションで、この API |
リクエストはキー "ClusterId" を使用します。Amazon EMR は "JobFlowId" を使用します。リクエストは 1 つのステップを使用します。 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://<amzn-s3-demo-bucket>
/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 EMR を呼び出すための IAM ポリシー
次のサンプルテンプレートは、 がステートマシン定義のリソースに基づいて IAM ポリシー AWS Step Functions を生成する方法を示しています。詳細については、Step Functions が統合サービスの IAM ポリシーを生成する方法およびStep Functions でサービス統合パターンを検出するを参照してください。
addStep
静的リソース
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:AddJobFlowSteps",
"elasticmapreduce:DescribeStep",
"elasticmapreduce:CancelSteps"
],
"Resource": [
"arn:aws:elasticmapreduce:[[region]]:[[accountId]]:cluster/[[clusterId]]"
]
}
]
}
動的リソース
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:AddJobFlowSteps",
"elasticmapreduce:DescribeStep",
"elasticmapreduce:CancelSteps"
],
"Resource": "arn:aws:elasticmapreduce:*:*:cluster/*"
}
]
}
cancelStep
静的リソース
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "elasticmapreduce:CancelSteps",
"Resource": [
"arn:aws:elasticmapreduce:[[region]]
:[[accountId]]
:cluster/[[clusterId]]
"
]
}
]
}
動的リソース
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "elasticmapreduce:CancelSteps",
"Resource": "arn:aws:elasticmapreduce:*:*:cluster/*"
}
]
}
createCluster
静的リソース
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:RunJobFlow",
"elasticmapreduce:DescribeCluster",
"elasticmapreduce:TerminateJobFlows"
],
"Resource": "*"
},
{
"Effect": "Allow",
"Action": "iam:PassRole",
"Resource": [
"arn:aws:iam::{{account}}
:role/[[roleName]]
"
]
}
]
}
setClusterTerminationProtection
静的リソース
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "elasticmapreduce:SetTerminationProtection",
"Resource": [
"arn:aws:elasticmapreduce:[[region]]
:[[accountId]]
:cluster/[[clusterId]]
"
]
}
]
}
動的リソース
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "elasticmapreduce:SetTerminationProtection",
"Resource": "arn:aws:elasticmapreduce:*:*:cluster/*"
}
]
}
modifyInstanceFleetByName
静的リソース
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:ModifyInstanceFleet",
"elasticmapreduce:ListInstanceFleets"
],
"Resource": [
"arn:aws:elasticmapreduce:[[region]]
:[[accountId]]
:cluster/[[clusterId]]
"
]
}
]
}
動的リソース
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:ModifyInstanceFleet",
"elasticmapreduce:ListInstanceFleets"
],
"Resource": "arn:aws:elasticmapreduce:*:*:cluster/*"
}
]
}
modifyInstanceGroupByName
静的リソース
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:ModifyInstanceGroups",
"elasticmapreduce:ListInstanceGroups"
],
"Resource": [
"arn:aws:elasticmapreduce:[[region]]
:[[accountId]]
:cluster/[[clusterId]]
"
]
}
]
}
動的リソース
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:ModifyInstanceGroups",
"elasticmapreduce:ListInstanceGroups"
],
"Resource": "*"
}
]
}
terminateCluster
静的リソース
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:TerminateJobFlows",
"elasticmapreduce:DescribeCluster"
],
"Resource": [
"arn:aws:elasticmapreduce:[[region]]
:[[accountId]]
:cluster/[[clusterId]]
"
]
}
]
}
動的リソース
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"elasticmapreduce:TerminateJobFlows",
"elasticmapreduce:DescribeCluster"
],
"Resource": "arn:aws:elasticmapreduce:*:*:cluster/*"
}
]
}