AWS Step Functions を使用して EKS で Amazon EMR クラスターを作成および管理する - AWS Step Functions

AWS Step Functions を使用して EKS で Amazon EMR クラスターを作成および管理する

AWS Step Functions を Amazon EMR on EKS と統合するには、EKS サービス統合 API で Amazon EMR を使用します。サービス統合 API は EKS API の対応する Amazon EMR と同じですが、次の表で示すとおり、すべての API があらゆる統合パターンをサポートしているわけではありません。

Step Functions で AWS サービスと統合することについては、「 サービスとの統合」および「Step Functions でサービス API にパラメータを渡す」を参照してください。

EKS 統合に対する最適化された Amazon EMR と EKS AWS SDDK 統合に対する Amazon EMR との違い
注記

Amazon EMR との統合のため、Step Functions は最初の 10 分とその後の 300 秒間、ジョブポーリング頻度をハードコーディングして 60 秒に設定しています。

API リクエストレスポンス ジョブの実行 (.sync)
CreateVirtualCluster サポート サポートされていません
DeleteVirtualCluster サポート サポート
StartJobRun サポート サポート

EKS API 上でサポートされる Amazon EMR は次のとおりです。

入力または結果データのクォータ

サービス間でデータを送受信する際、タスクの入力または結果の最大サイズは、UTF-8 でエンコードされた文字列で 256 KiB です。「ステートマシンの実行に関連するクォータ」を参照してください。

以下には仮想クラスターを作成する Task 状態が含まれています。

"Create_Virtual_Cluster": {
  "Type": "Task",
  "Resource": "arn:aws:states:::emr-containers:createVirtualCluster",
  "Arguments": {
    "Name": "MyVirtualCluster",
    "ContainerProvider": {
      "Id": "EKSClusterName",
      "Type": "EKS",
      "Info": {
        "EksInfo": {
          "Namespace": "Namespace"
        }
      }
    }
  },
  "End": true
}

以下には、 ジョブを仮想クラスターに送信し、完了するまで待機する Task 状態が含まれます。

"Submit_Job": {
    "Type": "Task",
    "Resource": "arn:aws:states:::emr-containers:startJobRun.sync",
    "Arguments": {
      "Name": "MyJobName",
      "VirtualClusterId": "{% $VirtualClusterId %}",
      "ExecutionRoleArn": "arn:aws:iam::<accountId>:role/job-execution-role",
      "ReleaseLabel": "emr-6.2.0-latest",
      "JobDriver": {
        "SparkSubmitJobDriver": {
          "EntryPoint": "s3://<amzn-s3-demo-bucket>/jobs/trip-count.py",
          "EntryPointArguments": [
            "60"
          ],
          "SparkSubmitParameters": "--conf spark.driver.cores=2 --conf spark.executor.instances=10 --conf spark.kubernetes.pyspark.pythonVersion=3 --conf spark.executor.memory=10G --conf spark.driver.memory=10G --conf spark.executor.cores=1 --conf spark.dynamicAllocation.enabled=false"
        }
      },
      "ConfigurationOverrides": {
        "ApplicationConfiguration": [
          {
            "Classification": "spark-defaults",
            "Properties": {
              "spark.executor.instances": "2",
              "spark.executor.memory": "2G"
            }
          }
        ],
        "MonitoringConfiguration": {
          "PersistentAppUI": "ENABLED",
          "CloudWatchMonitoringConfiguration": {
            "LogGroupName": "MyLogGroupName",
            "LogStreamNamePrefix": "MyLogStreamNamePrefix"
          },
          "S3MonitoringConfiguration": {
            "LogUri": "s3://<amzn-s3-demo-logging-bucket1>"
          }
        }
      },
      "Tags": {
        "taskType": "jobName"
      }
    },
    "End": true
}

以下には、仮想クラスターを削除し、削除が完了するまで待機する Task 状態が含まれます。

"Delete_Virtual_Cluster": {
  "Type": "Task",
  "Resource": "arn:aws:states:::emr-containers:deleteVirtualCluster.sync",
  "Arguments": {
    "Id": "{% $states.input.VirtualClusterId %}",
  },
  "End": true
}

他の AWS のサービスで Step Functions を使用して IAM アクセス許可を設定する方法については、「Step Functions が統合サービスの IAM ポリシーを生成する方法」を参照してください。