View a markdown version of this page

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

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

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

Découvrez comment intégrer AWS Step Functions Amazon EMR à l'aide des API d'intégration des services Amazon EMR fournies. Les API d'intégration de services sont similaires aux API Amazon EMR correspondantes, avec quelques différences dans les champs transmis et dans les réponses renvoyées.

Pour en savoir plus sur l'intégration aux AWS services de Step Functions, reportez-vous Intégration des services aux sections etTransmission de paramètres à une API de service dans Step Functions.

Principales caractéristiques de l'intégration optimisée d'Amazon EMR
  • L'intégration optimisée du service Amazon EMR comporte un ensemble personnalisé d'API qui englobe les API Amazon EMR sous-jacentes, décrites ci-dessous. De ce fait, il est très différent de l'intégration du service Amazon EMR AWS SDK.

  • Le modèle Exécuter une tâche (.sync) d'intégration est pris en charge.

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

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 fonctions Map et les Parallel états des étapes pour soumettre des travaux en parallèle au cluster.

La disponibilité de l'intégration du service Amazon EMR dépend de la disponibilité des API Amazon EMR. Consultez la documentation Amazon EMR pour connaître les limites applicables à certaines régions.

Note

Pour l'intégration à Amazon EMR, Step Functions propose une fréquence d'interrogation des tâches codée en dur de 60 secondes pendant les 10 premières minutes et les 300 secondes suivantes.

API Amazon EMR optimisées

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

API d'intégration du service Amazon EMR API EMR correspondante Différences
createCluster

Crée et démarre l'exécution d'un cluster (flux de travail).

Amazon EMR est directement lié à un type unique de rôle IAM 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'autorisations IAM, consultez Utilisation du Service-Linked rôle pour Amazon EMR.

courir JobFlow createClusterutilise la même syntaxe de requête que run JobFlow, à l'exception des éléments suivants :
  • 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 l'API de connecteur modifyInstanceFleetByName facultative est utilisée.

  • Le champ Instances.InstanceGroups[index].Name doit être fourni et doit être unique si l'API facultative modifyInstanceGroupByName 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).

courir JobFlow Le même que createCluster, mais attend que le cluster atteigne l'état WAITING.
ensemble ClusterTerminationProtection

Verrouille un cluster (flux de travail) afin que les instances EC2 du cluster ne puissent pas être arrêtées par l'intervention de l'utilisateur, un appel d'API ou une erreur de flux de travail.

ensemble TerminationProtection La demande utilise ceci :
{ "ClusterId": "string" }
Amazon EMR utilise ceci :
{ "JobFlowIds": ["string"] }
terminateCluster

Arrête un cluster (flux de travail).

mettre fin JobFlows La demande utilise ceci :
{ "ClusterId": "string" }
Amazon EMR utilise ceci :
{ "JobFlowIds": ["string"] }
mettre fin Cluster.sync

Arrête un cluster (flux de travail).

mettre fin JobFlows Identique à terminateCluster, mais attend que le cluster ait terminé.
addStep

Ajoute une nouvelle étape à un cluster en cours d'exécution.

En option, vous pouvez également spécifier le ExecutionRoleArn paramètre lors de l'utilisation de cette API.

ajouter JobFlowSteps

La requête 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 ceci :
{ "StepIds": [<strings>] }
ajouter Step.sync

Ajoute une nouvelle étape à un cluster en cours d'exécution.

En option, vous pouvez également spécifier le ExecutionRoleArn paramètre lors de l'utilisation de cette API.

ajouter JobFlowSteps

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>] }
modifier InstanceFleetByName

Modifie les capacités Spot cible On-Demand et cible pour le parc d'instances en fonction des valeurs spécifiéesInstanceFleetName.

modifier InstanceFleet 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.

modifier InstanceGroupByName

Modifie le nombre de nœuds et de paramètres de configuration d'un groupe d'instances.

modifier InstanceGroups 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", "Arguments": { "Name": "MyWorkflowCluster", "VisibleToAllUsers": true, "ReleaseLabel": "emr-5.28.0", "Applications": [ { "Name": "Hive" } ], "ServiceRole": "EMR_DefaultRole", "JobFlowRole": "EMR_EC2_DefaultRole", "LogUri": "s3n://aws-logs-account-id-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", "Arguments": { "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", "Arguments": { "ClusterId": "{% $ClusterId %}", "ExecutionRoleArn": "arn:aws:iam::account-id:role/myEMR-execution-role", "Step": { "Name": "The first step", "ActionOnFailure": "TERMINATE_CLUSTER", "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", "Arguments": { "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", "Arguments": { "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", "Arguments": { "ClusterId": "j-account-id3", "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", "Arguments": { "ClusterId": "j-account-id3", "InstanceFleetName": "MyCoreFleet", "InstanceFleet": { "TargetOnDemandCapacity": 8, "TargetSpotCapacity": 0 } }, "End": true }

Politiques IAM pour les appels à Amazon EMR

Les exemples de modèles suivants montrent comment AWS Step Functions générer des politiques IAM en fonction des ressources de la définition de votre machine à états. Pour plus d’informations, consultez Comment Step Functions génère des politiques IAM 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:us-east-1:123456789012: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:us-east-1:123456789012:cluster/myCluster-id" ] } ] }

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::123456789012:role/myRoleName" ] } ] }

ensemble ClusterTerminationProtection

Ressources statiques

{ "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "elasticmapreduce:SetTerminationProtection", "Resource": [ "arn:aws:elasticmapreduce:us-east-1:123456789012:cluster/myCluster-id" ] } ] }

Ressources dynamiques

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

modifier InstanceFleetByName

Ressources statiques

{ "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "elasticmapreduce:ModifyInstanceFleet", "elasticmapreduce:ListInstanceFleets" ], "Resource": [ "arn:aws:elasticmapreduce:us-east-1:123456789012:cluster/myCluster-id" ] } ] }

Ressources dynamiques

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

modifier InstanceGroupByName

Ressources statiques

{ "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "elasticmapreduce:ModifyInstanceGroups", "elasticmapreduce:ListInstanceGroups" ], "Resource": [ "arn:aws:elasticmapreduce:us-east-1:123456789012:cluster/myCluster-id" ] } ] }

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:us-east-1:123456789012:cluster/myCluster-id" ] } ] }

Ressources dynamiques

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