Créez et gérez des EMR clusters Amazon avec Step Functions

Découvrez comment intégrer AWS Step Functions avec Amazon EMR en utilisant l'intégration du EMR service Amazon fournieAPIs. L'intégration APIs des services est similaire à celle d'Amazon correspondante EMRAPIs, avec quelques différences dans les champs transmis et dans les réponses renvoyées.

Pour en savoir plus sur l'intégration avec AWS services dans Step Functions, voir Intégration des services et. Transmission de paramètres à un service API dans Step Functions

Principales fonctionnalités de l'EMRintégration optimisée avec Amazon

L'intégration optimisée du EMR service Amazon comprend un ensemble personnalisé APIs qui englobe l'Amazon sous-jacent EMRAPIs, décrit ci-dessous. De ce fait, il diffère considérablement de l'Amazon EMR AWS SDKintégration des services.
Le modèle Exécuter une tâche (.sync) d'intégration est pris en charge.

Step Functions ne met pas automatiquement fin à un EMR cluster Amazon si l'exécution est arrêtée. Si votre machine d'état s'arrête avant la fin de votre EMR cluster Amazon, celui-ci peut continuer à fonctionner indéfiniment et entraîner des frais supplémentaires. Pour éviter cela, assurez-vous que tout EMR cluster Amazon que vous créez est correctement résilié. Pour plus d’informations, consultez :

Contrôlez la résiliation du cluster dans le guide de EMR l'utilisateur Amazon.
La Exécuter une tâche (.sync) section Modèles d'intégration des services.

Note

À partir deemr-5.28.0, vous pouvez spécifier le paramètre StepConcurrencyLevel lors de la création d'un cluster pour permettre à plusieurs étapes de s'exécuter en parallèle sur un seul cluster. Vous pouvez utiliser les Step Functions Map et Parallel les états pour soumettre des travaux en parallèle au cluster.

La disponibilité de l'intégration des EMR services Amazon dépend de la disponibilité d'Amazon EMRAPIs. Consultez EMR la documentation Amazon pour connaître les restrictions applicables dans certaines régions.

Note

Pour l'intégration à AmazonEMR, Step Functions dispose d'une fréquence d'interrogation des offres d'emploi codée en dur de 60 secondes pendant les 10 premières minutes, puis de 300 secondes par la suite.

Amazon pris en charge EMR APIs

Le tableau suivant décrit les différences entre chaque intégration EMR de services Amazon API et Amazon correspondant EMRAPIs.

Intégration des EMR services Amazon API	Correspondant EMR API	Différences
createCluster Crée et démarre l'exécution d'un cluster (flux de travail). Amazon EMR est directement lié à un type de IAM rôle unique appelé rôle lié à un service. Pour que `createCluster` et `createCluster.sync` fonctionnent, vous devez avoir configuré les autorisations nécessaires pour créer le rôle lié au service `AWSServiceRoleForEMRCleanup`. Pour plus d'informations à ce sujet, y compris une déclaration que vous pouvez ajouter à votre politique d'IAMautorisations, consultez Utiliser le rôle lié à un service pour Amazon. EMR	runJobFlow	`createCluster`utilise la même syntaxe de demande que runJobFlow, à l'exception de ce qui suit : Le champ `Instances.KeepJobFlowAliveWhenNoSteps` est obligatoire et doit avoir la valeur booléenne `TRUE`. Le champ `Steps` n'est pas autorisé. Le champ `Instances.InstanceFleets[index].Name` doit être fourni et doit être unique si le `modifyInstanceFleetByName` connecteur optionnel API est utilisé. Le champ `Instances.InstanceGroups[index].Name` doit être fourni et doit être unique si l'option `modifyInstanceGroupByName` API est utilisée. La réponse est la suivante : `{ "ClusterId": "string" }` Amazon EMR utilise ceci : `{ "JobFlowId": "string" }`
createCluster.sync Crée et démarre l'exécution d'un cluster (flux de travail).	runJobFlow	Le même que `createCluster`, mais attend que le cluster atteigne l'état `WAITING`.
setClusterTerminationProtection Verrouille un cluster (flux de travail) afin que les EC2 instances du cluster ne puissent pas être résiliées par une intervention de l'utilisateur, un API appel ou une erreur de flux de travail.	setTerminationProtection	La demande utilise ceci : `{ "ClusterId": "string" }` Amazon EMR utilise ceci : `{ "JobFlowIds": ["string"] }`
terminateCluster Arrête un cluster (flux de travail).	terminateJobFlows	La demande utilise ceci : `{ "ClusterId": "string" }` Amazon EMR utilise ceci : `{ "JobFlowIds": ["string"] }`
terminateCluster.sync Arrête un cluster (flux de travail).	terminateJobFlows	Identique à `terminateCluster`, mais attend que le cluster ait terminé.
addStep Ajoute une nouvelle étape à un cluster en cours d'exécution. Facultativement, vous pouvez également spécifier le `ExecutionRoleArn` paramètre lors de son utilisationAPI.	addJobFlowÉtapes	La demande utilise la clé`"ClusterId"`. Amazon EMR utilise`"JobFlowId"`. La demande utilise une seule étape. `{ "Step": <"StepConfig object"> }` Amazon EMR utilise ceci : `{ "Steps": [<StepConfig objects>] }` La réponse est la suivante : `{ "StepId": "string" }` Amazon EMR renvoie ce qui suit : `{ "StepIds": [<strings>] }`
addStep.sync Ajoute une nouvelle étape à un cluster en cours d'exécution. Facultativement, vous pouvez également spécifier le `ExecutionRoleArn` paramètre lors de son utilisationAPI.	addJobFlowÉtapes	Identique à `addStep`, mais attend que l'étape se termine.
cancelStep Annule une étape en attente dans un cluster en cours d'exécution	cancelSteps	La demande utilise ceci : `{ "StepId": "string" }` Amazon EMR utilise ceci : `{ "StepIds": [<strings>] }` La réponse est la suivante : `{ "CancelStepsInfo": <CancelStepsInfo object> }` Amazon EMR utilise ceci : `{ "CancelStepsInfoList": [<CancelStepsInfo objects>] }`
modifyInstanceFleetByName Modifie les capacités d'instances à la demande et d'instances ponctuelles cibles pour le parc d'instances avec l'`InstanceFleetName` spécifié.	modifyInstanceFleet	La demande est la même que pour `modifyInstanceFleet`, sauf pour ce qui suit : Le champ `Instance.InstanceFleetId` n'est pas autorisé. Lors de l'exécution, le `InstanceFleetId` est déterminé automatiquement par l'intégration du service en appelant `ListInstanceFleets` et en analysant le résultat.
modifyInstanceGroupByName Modifie le nombre de nœuds et de paramètres de configuration d'un groupe d'instances.	modifyInstanceGroups	La demande est la suivante : `{ "ClusterId": "string", "InstanceGroup": <InstanceGroupModifyConfig object> }` Amazon EMR utilise une liste : `{ "ClusterId": ["string"], "InstanceGroups": [<InstanceGroupModifyConfig objects>] }` Dans l'objet `InstanceGroupModifyConfig`, le champ `InstanceGroupId` n'est pas autorisé. Un nouveau champ, `InstanceGroupName`, a été ajouté. Lors de l'exécution, le `InstanceGroupId` est déterminé automatiquement par l'intégration du service en appelant `ListInstanceGroups` et en analysant le résultat.

Exemple de flux de travail

L'exemple suivant inclut un état Task qui crée un 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
}

Ce qui suit inclut un état Task qui permet la protection contre la résiliation.


"Enable_Termination_Protection": {
    "Type": "Task",
    "Resource": "arn:aws:states:::elasticmapreduce:setClusterTerminationProtection",
    "Parameters": {
        "ClusterId.$": "$.ClusterId",
        "TerminationProtected": true
    },
    "End": true
}

Ce qui suit inclut un état Task qui soumet une étape à un cluster.


"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
}

Ce qui suit inclut un état Task qui annule une étape.


"Cancel_Step_One": {
    "Type": "Task",
    "Resource": "arn:aws:states:::elasticmapreduce:cancelStep",
    "Parameters": {
        "ClusterId.$": "$.ClusterId",
        "StepId.$": "$.AddStepsResult.StepId"
    },
    "End": true
}

Ce qui suit inclut un état Task qui met fin à un cluster.


"Terminate_Cluster": {
    "Type": "Task",
    "Resource": "arn:aws:states:::elasticmapreduce:terminateCluster.sync",
    "Parameters": {
        "ClusterId.$": "$.ClusterId"
    },
    "End": true
}

Ce qui suit inclut un état Task qui met à l'échelle un cluster vers le haut ou vers le bas pour un groupe d'instances.



"ModifyInstanceGroupByName": {
    "Type": "Task",
    "Resource": "arn:aws:states:::elasticmapreduce:modifyInstanceGroupByName",
    "Parameters": {
        "ClusterId": "j-1234567890123",
        "InstanceGroupName": "MyCoreGroup",
        "InstanceGroup": {
            "InstanceCount": 8
        }
    },
    "End": true
}

Ce qui suit inclut un état Task qui met à l'échelle un cluster vers le haut ou vers le bas pour une flotte d'instances.


"ModifyInstanceFleetByName": {
    "Type": "Task",
    "Resource": "arn:aws:states:::elasticmapreduce:modifyInstanceFleetByName",
    "Parameters": {
        "ClusterId": "j-1234567890123",
        "InstanceFleetName": "MyCoreFleet",
        "InstanceFleet": {
            "TargetOnDemandCapacity": 8,
            "TargetSpotCapacity": 0
        }
    },
    "End": true
}

IAMpolitiques relatives aux appels d'Amazon EMR

Les exemples de modèles suivants montrent comment AWS Step Functions génère IAM des politiques basées sur les ressources contenues dans la définition de votre machine à états. Pour plus d’informations, consultez Comment Step Functions génère IAM des politiques pour les services intégrés et Découvrez les modèles d'intégration des services dans Step Functions.

`addStep`

Ressources statiques


{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "elasticmapreduce:AddJobFlowSteps",
        "elasticmapreduce:DescribeStep",
        "elasticmapreduce:CancelSteps"
      ],
      "Resource": [
        "arn:aws:elasticmapreduce:[[region]]:[[accountId]]:cluster/[[clusterId]]"
      ]
    }
  ]
}

Ressources dynamiques


{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "elasticmapreduce:AddJobFlowSteps",
        "elasticmapreduce:DescribeStep",
        "elasticmapreduce:CancelSteps"
      ],
      "Resource": "arn:aws:elasticmapreduce:*:*:cluster/*"
    }
  ]
}

`cancelStep`

Ressources statiques


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "elasticmapreduce:CancelSteps",
            "Resource": [
                "arn:aws:elasticmapreduce:[[region]]:[[accountId]]:cluster/[[clusterId]]"
            ]
        }
    ]
}

Ressources dynamiques


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "elasticmapreduce:CancelSteps",
            "Resource": "arn:aws:elasticmapreduce:*:*:cluster/*"
        }
    ]
}

`createCluster`

Ressources statiques


{
  "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`

Ressources statiques


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "elasticmapreduce:SetTerminationProtection",
            "Resource": [
                "arn:aws:elasticmapreduce:[[region]]:[[accountId]]:cluster/[[clusterId]]"
            ]
        }
    ]
}

Ressources dynamiques


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "elasticmapreduce:SetTerminationProtection",
            "Resource": "arn:aws:elasticmapreduce:*:*:cluster/*"
        }
    ]
}

`modifyInstanceFleetByName`

Ressources statiques


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "elasticmapreduce:ModifyInstanceFleet",
                "elasticmapreduce:ListInstanceFleets"
            ],
            "Resource": [
                "arn:aws:elasticmapreduce:[[region]]:[[accountId]]:cluster/[[clusterId]]"
            ]
        }
    ]
}

Ressources dynamiques


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "elasticmapreduce:ModifyInstanceFleet",
                "elasticmapreduce:ListInstanceFleets"
            ],
            "Resource": "arn:aws:elasticmapreduce:*:*:cluster/*"
        }
    ]
}

`modifyInstanceGroupByName`

Ressources statiques


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "elasticmapreduce:ModifyInstanceGroups",
                "elasticmapreduce:ListInstanceGroups"
            ],
            "Resource": [
                "arn:aws:elasticmapreduce:[[region]]:[[accountId]]:cluster/[[clusterId]]"
            ]
        }
    ]
}

Ressources dynamiques


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "elasticmapreduce:ModifyInstanceGroups",
                "elasticmapreduce:ListInstanceGroups"
            ],
            "Resource": "*"
        }
    ]
}

`terminateCluster`

Ressources statiques


{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "elasticmapreduce:TerminateJobFlows",
        "elasticmapreduce:DescribeCluster"
      ],
      "Resource": [
        "arn:aws:elasticmapreduce:[[region]]:[[accountId]]:cluster/[[clusterId]]"
      ]
    }
  ]
}

Ressources dynamiques


{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "elasticmapreduce:TerminateJobFlows",
        "elasticmapreduce:DescribeCluster"
      ],
      "Resource": "arn:aws:elasticmapreduce:*:*:cluster/*"
    }
  ]
}

Avertissement JavaScript est désactivé ou n'est pas disponible dans votre navigateur.

Pour que vous puissiez utiliser la documentation AWS, Javascript doit être activé. Vous trouverez des instructions sur les pages d'aide de votre navigateur.

Conventions de rédaction

Amazon EKS

Amazon EMR sur EKS