View a markdown version of this page

Crea e gestisci cluster Amazon EMR con Step Functions - AWS Step Functions
API supportateEsempiPolicy IAM

Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Crea e gestisci cluster Amazon EMR con Step Functions

Scopri come integrarti AWS Step Functions con Amazon EMR utilizzando le API di integrazione dei servizi Amazon EMR fornite. Le API di integrazione dei servizi sono simili alle corrispondenti API Amazon EMR, con alcune differenze nei campi trasmessi e nelle risposte restituite.

Per ulteriori informazioni sull'integrazione con i AWS servizi in Step Functions, consulta e. Integrazione dei servizi Passaggio di parametri a un'API di servizio in Step Functions

Caratteristiche principali dell'integrazione ottimizzata di Amazon EMR

  • L'integrazione del servizio Amazon EMR ottimizzato include un set personalizzato di API che racchiude le API Amazon EMR sottostanti, descritte di seguito. Per questo motivo, differisce in modo significativo dall'integrazione del servizio Amazon EMR SDK. AWS

  • Il modello di Esegui un processo (.sync) integrazione è supportato.

Step Functions non termina automaticamente un cluster Amazon EMR se l'esecuzione viene interrotta. Se il tuo sistema a stato si arresta prima che il cluster Amazon EMR sia terminato, il cluster può continuare a funzionare all'infinito e può comportare costi aggiuntivi. Per evitare ciò, assicurati che qualsiasi cluster Amazon EMR che crei sia terminato correttamente. Per ulteriori informazioni, consulta:

Nota

A partire daemr-5.28.0, è possibile specificare il parametro StepConcurrencyLevel durante la creazione di un cluster per consentire l'esecuzione di più passaggi in parallelo su un singolo cluster. È possibile utilizzare le funzioni Map e Parallel gli stati delle fasi per inviare il lavoro in parallelo al cluster.

La disponibilità dell'integrazione del servizio Amazon EMR è soggetta alla disponibilità delle API Amazon EMR. Consulta la documentazione di Amazon EMR per le limitazioni in regioni speciali.

Nota

Per l'integrazione con Amazon EMR, Step Functions ha una frequenza di job polling codificata di 60 secondi per i primi 10 minuti e 300 secondi dopo.

API Amazon EMR ottimizzate

La tabella seguente descrive le differenze tra ciascuna API di integrazione dei servizi Amazon EMR e le corrispondenti API Amazon EMR.

API di integrazione dei servizi Amazon EMR API EMR corrispondente Differenze
createCluster

Crea e avvia l'esecuzione di un cluster (flusso di lavoro).

Amazon EMR è collegato direttamente a un tipo unico di ruolo IAM noto come ruolo collegato ai servizi. Perché createCluster e createCluster.sync funzionino, è necessario aver configurato le autorizzazioni necessarie per creare il ruolo collegato al servizio AWSServiceRoleForEMRCleanup. Per ulteriori informazioni su questo argomento, inclusa una dichiarazione che puoi aggiungere alla tua politica di autorizzazioni IAM, consulta Using the Service-Linked Role for Amazon EMR.

esegui JobFlow createClusterutilizza la stessa sintassi di richiesta di run JobFlow, ad eccezione di quanto segue:

  • Il campo Instances.KeepJobFlowAliveWhenNoSteps è obbligatorio e deve avere il valore booleano TRUE.

  • Il campo Steps non è consentito.

  • Deve essere fornito il campo Instances.InstanceFleets[index].Name, che deve essere univoco se viene utilizzata l'API del connettore modifyInstanceFleetByName opzionale.

  • Deve essere fornito il campo Instances.InstanceGroups[index].Name, che deve essere univoco se viene utilizzata l'API modifyInstanceGroupByName opzionale.

La risposta è: 
{
  "ClusterId": "string"
}
Amazon EMR utilizza questo: 
{
  "JobFlowId": "string"
}
createCluster.sync

Crea e avvia l'esecuzione di un cluster (flusso di lavoro).

correre JobFlow Lo stesso di createCluster, ma attende che il cluster raggiunga lo stato WAITING.
impostare ClusterTerminationProtection

Blocca un cluster (flusso di lavoro) in modo che le istanze EC2 nel cluster non possano essere terminate dall'intervento dell'utente, da una chiamata API o in caso di errore del flusso di lavoro.

 set TerminationProtection La richiesta utilizza questo:
{
  "ClusterId": "string"
}
Amazon EMR utilizza questo:
{
  "JobFlowIds": ["string"]
}
terminateCluster

Arresta un cluster (flusso di lavoro).

 terminare JobFlows La richiesta utilizza questo:
{
  "ClusterId": "string"
}
Amazon EMR utilizza questo:
{
  "JobFlowIds": ["string"]
}
terminare Cluster.sync

Arresta un cluster (flusso di lavoro).

 terminare JobFlows Lo stesso di terminateCluster, ma attende che il cluster venga terminato.
addStep

Aggiunge un nuovo passaggio a un cluster in esecuzione.

Facoltativamente, puoi anche specificare il ExecutionRoleArn parametro durante l'utilizzo di questa API.

aggiungere JobFlowSteps

La richiesta utilizza la chiave"ClusterId". Amazon EMR utilizza"JobFlowId". La richiesta utilizza un singolo passaggio.
{
  "Step": <"StepConfig object">
}
Amazon EMR utilizza questo:
{
  "Steps": [<StepConfig objects>]
}
La risposta è:
{
  "StepId": "string"
}
Amazon EMR restituisce quanto segue:
{
  "StepIds": [<strings>]
}
aggiungi Step.sync

Aggiunge un nuovo passaggio a un cluster in esecuzione.

Facoltativamente, puoi anche specificare il ExecutionRoleArn parametro durante l'utilizzo di questa API.

aggiungere JobFlowSteps

Come addStep, ma attende il completamento del passaggio.
cancelStep

Annulla un passaggio in sospeso in un cluster in esecuzione.

 cancelSteps La richiesta utilizza questo:
{
  "StepId": "string"
}
Amazon EMR utilizza questo:
{
  "StepIds": [<strings>]
}
La risposta è:
{
  "CancelStepsInfo": <CancelStepsInfo object>
}
Amazon EMR utilizza questo:
{
  "CancelStepsInfoList": [<CancelStepsInfo objects>]
}
modificare InstanceFleetByName

Modifica le capacità Spot target On-Demand e target per il parco istanze con quanto specificato. InstanceFleetName

 modifica InstanceFleet La richiesta è la stessa di quella di modifyInstanceFleet, ad eccezione di quanto segue:

  • Il campo Instance.InstanceFleetId non è consentito.

  • In fase di esecuzione InstanceFleetId viene determinato automaticamente dall'integrazione del servizio chiamando ListInstanceFleets e analizzando il risultato.
modificare InstanceGroupByName

Modifica il numero di nodi e le impostazioni di configurazione di un gruppo di istanze.

 modificare InstanceGroups La richiesta è questa:
{
  "ClusterId": "string",
  "InstanceGroup": <InstanceGroupModifyConfig object>
}
Amazon EMR utilizza un elenco: 
{
  "ClusterId": ["string"],
  "InstanceGroups": [<InstanceGroupModifyConfig objects>]
}

All'interno dell'oggetto InstanceGroupModifyConfig il campo InstanceGroupId non è consentito.

È stato aggiunto un nuovo campo, InstanceGroupName. In fase di esecuzione InstanceGroupId viene determinato automaticamente dall'integrazione del servizio chiamando ListInstanceGroups e analizzando il risultato.

Esempio di workflow

L'esempio seguente include uno stato Task che crea 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
}

Di seguito è riportato uno stato Task che consente la protezione di terminazione. 

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

Di seguito è riportato uno stato Task che invia un passaggio a 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
}

Di seguito è riportato uno stato Task che annulla un passaggio. 

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

Di seguito è riportato uno stato Task che termina un cluster. 

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

Di seguito è riportato uno stato Task che consente di scalare un cluster verso l'alto o verso il basso per un gruppo di istanze.


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

Di seguito è riportato uno stato Task che consente di scalare un cluster verso l'alto o verso il basso per un parco di istanze.

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

Politiche IAM per chiamare Amazon EMR

I seguenti modelli di esempio mostrano come AWS Step Functions generare policy IAM in base alle risorse nella definizione della macchina a stati. Per ulteriori informazioni, consulta In che modo Step Functions genera policy IAM per servizi integrati e Scopri i modelli di integrazione dei servizi in Step Functions.

addStep

Risorse statiche

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

Risorse dinamiche

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

cancelStep

Risorse statiche

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

Risorse dinamiche

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

createCluster

Risorse statiche

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

impostato ClusterTerminationProtection

Risorse statiche

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

Risorse dinamiche

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

modificare InstanceFleetByName

Risorse statiche

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

Risorse dinamiche

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

modificare InstanceGroupByName

Risorse statiche

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

Risorse dinamiche

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

terminateCluster

Risorse statiche

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

Risorse dinamiche

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