Crie e gerencie EMR clusters da Amazon com Step Functions

Saiba como integrar AWS Step Functions com a Amazon EMR usando a integração de EMR serviços da Amazon fornecidaAPIs. A integração do serviço APIs é semelhante à Amazon correspondente EMRAPIs, com algumas diferenças nos campos que são passados e nas respostas que são retornadas.

Para saber mais sobre a integração com AWS serviços em Step Functions, consulte e. Integração de produtos da Passando parâmetros para um serviço API em Step Functions

Principais recursos da EMR integração otimizada com a Amazon

A integração otimizada do EMR serviço Amazon tem um conjunto personalizado APIs que envolve a Amazon subjacente EMRAPIs, descrito abaixo. Por causa disso, ele difere significativamente da Amazon EMR AWS SDKintegração de serviços.
O padrão de integração Executar um trabalho (.sync) é compatível.

O Step Functions não encerra automaticamente um EMR cluster da Amazon se a execução for interrompida. Se sua máquina de estado parar antes do EMR encerramento do cluster da Amazon, seu cluster poderá continuar funcionando indefinidamente e poderá acumular cobranças adicionais. Para evitar isso, certifique-se de que qualquer EMR cluster da Amazon que você criar seja encerrado corretamente. Para obter mais informações, consulte:

Controle o encerramento do cluster no Guia EMR do usuário da Amazon.
A seção Executar um trabalho (.sync) de padrões de integração de serviços.

nota

A partir de emr-5.28.0, você pode especificar o parâmetro StepConcurrencyLevel ao criar um cluster para permitir que várias etapas sejam executadas em paralelo em um único cluster. Você pode usar os estados de Map e Parallel do Step Functions para enviar trabalho em paralelo ao cluster.

A disponibilidade da integração de EMR serviços da Amazon está sujeita à disponibilidade da Amazon EMRAPIs. Consulte a EMR documentação da Amazon para ver as limitações em regiões especiais.

nota

Para integração com a AmazonEMR, o Step Functions tem uma frequência de pesquisa de trabalhos codificada de 60 segundos nos primeiros 10 minutos e 300 segundos depois disso.

Amazon compatível EMR APIs

A tabela a seguir descreve as diferenças entre cada integração de EMR serviços da Amazon API e a Amazon correspondente EMRAPIs.

Integração EMR de serviços da Amazon API	Correspondente EMR API	Diferenças
createCluster Cria e inicia a execução de um cluster (fluxo de trabalho). A Amazon EMR está vinculada diretamente a um tipo exclusivo de IAM função conhecida como função vinculada a serviços. Para que `createCluster` e `createCluster.sync` funcionem, você deve ter configurado as permissões necessárias para criar a função vinculada ao serviço `AWSServiceRoleForEMRCleanup`. Para obter mais informações sobre isso, incluindo uma declaração que você pode adicionar à sua política de IAM permissões, consulte Usando a função vinculada ao serviço para a Amazon. EMR	runJobFlow	`createCluster`usa a mesma sintaxe de solicitação que runJobFlow, exceto pelo seguinte: O campo `Instances.KeepJobFlowAliveWhenNoSteps` é obrigatório e deve ter o valor booliano `TRUE`. O campo `Steps` não é permitido. O campo `Instances.InstanceFleets[index].Name` deve ser fornecido e deve ser exclusivo se o `modifyInstanceFleetByName` conector opcional API for usado. O campo `Instances.InstanceGroups[index].Name` deve ser fornecido e deve ser exclusivo se o opcional `modifyInstanceGroupByName` API for usado. A resposta é a seguinte: `{ "ClusterId": "string" }` A Amazon EMR usa isso: `{ "JobFlowId": "string" }`
createCluster.sync Cria e inicia a execução de um cluster (fluxo de trabalho).	runJobFlow	O mesmo que `createCluster`, mas espera que o cluster atinja o estado `WAITING`.
setClusterTerminationProteção Bloqueia um cluster (fluxo de trabalho) para que as EC2 instâncias no cluster não possam ser encerradas pela intervenção do usuário, por uma API chamada ou por um erro de fluxo de trabalho.	setTerminationProtection	A solicitação usa o seguinte: `{ "ClusterId": "string" }` A Amazon EMR usa isso: `{ "JobFlowIds": ["string"] }`
terminateCluster Desliga um cluster (fluxo de trabalho).	terminateJobFlows	A solicitação usa o seguinte: `{ "ClusterId": "string" }` A Amazon EMR usa isso: `{ "JobFlowIds": ["string"] }`
terminateCluster.sync Desliga um cluster (fluxo de trabalho).	terminateJobFlows	O mesmo que `terminateCluster`, mas aguarda o encerramento do cluster.
addStep Adiciona uma nova etapa a um cluster em execução. Opcionalmente, você também pode especificar o `ExecutionRoleArn` parâmetro ao usá-lo. API	addJobFlowEtapas	A solicitação usa a chave `"ClusterId"`. A Amazon EMR usa`"JobFlowId"`. A solicitação usa uma única etapa. `{ "Step": <"StepConfig object"> }` A Amazon EMR usa isso: `{ "Steps": [<StepConfig objects>] }` A resposta é a seguinte: `{ "StepId": "string" }` A Amazon EMR devolve isso: `{ "StepIds": [<strings>] }`
addStep.sync Adiciona uma nova etapa a um cluster em execução. Opcionalmente, você também pode especificar o `ExecutionRoleArn` parâmetro ao usá-lo. API	addJobFlowEtapas	O mesmo que `addStep`, mas aguarda a etapa ser concluída.
cancelStep Cancela uma etapa pendente em um cluster em execução.	cancelSteps	A solicitação usa o seguinte: `{ "StepId": "string" }` A Amazon EMR usa isso: `{ "StepIds": [<strings>] }` A resposta é a seguinte: `{ "CancelStepsInfo": <CancelStepsInfo object> }` A Amazon EMR usa isso: `{ "CancelStepsInfoList": [<CancelStepsInfo objects>] }`
modifyInstanceFleetByName Modifica as capacidades sob demanda e spot de destino para a frota de instâncias com o `InstanceFleetName` especificado.	modifyInstanceFleet	O pedido é o mesmo que para `modifyInstanceFleet`, exceto pelo seguinte: O campo `Instance.InstanceFleetId` não é permitido. Em tempo de execução, o `InstanceFleetId` é determinado automaticamente pela integração do serviço chamando `ListInstanceFleets` e analisando o resultado.
modifyInstanceGroupByName Modifica o número de nós e as configurações de um grupo de instâncias.	modifyInstanceGroups	O pedido é o seguinte: `{ "ClusterId": "string", "InstanceGroup": <InstanceGroupModifyConfig object> }` A Amazon EMR usa uma lista: `{ "ClusterId": ["string"], "InstanceGroups": [<InstanceGroupModifyConfig objects>] }` Dentro do objeto `InstanceGroupModifyConfig`, o campo `InstanceGroupId` não é permitido. Um novo campo, `InstanceGroupName`, foi adicionado. Em tempo de execução, o `InstanceGroupId` é determinado automaticamente pela integração do serviço chamando `ListInstanceGroups` e analisando o resultado.

Exemplo de fluxo de trabalho

Veja a seguir um estado Task que cria um 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
}

Veja a seguir um estado Task que habilita a proteção contra encerramento.


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

Veja a seguir um estado Task que envia uma etapa para um 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
}

Veja a seguir um estado Task que cancela uma etapa.


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

Veja a seguir um estado Task que encerra um cluster.


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

Veja a seguir um estado Task que expande ou reduz um cluster para um grupo de instâncias.



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

Veja a seguir um estado Task que expande ou reduz um cluster para uma frota de instâncias.


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

IAMpolíticas para ligar para a Amazon EMR

Os modelos de exemplo a seguir mostram como AWS Step Functions gera IAM políticas com base nos recursos em sua definição de máquina de estado. Para ter mais informações, consulte Como o Step Functions gera IAM políticas para serviços integrados e Descubra padrões de integração de serviços em Step Functions.

`addStep`

Recursos estáticos


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

Recursos dinâmicos


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

`cancelStep`

Recursos estáticos


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

Recursos dinâmicos


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

`createCluster`

Recursos estáticos


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

Recursos estáticos


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

Recursos dinâmicos


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

`modifyInstanceFleetByName`

Recursos estáticos


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

Recursos dinâmicos


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

`modifyInstanceGroupByName`

Recursos estáticos


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

Recursos dinâmicos


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

`terminateCluster`

Recursos estáticos


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

Recursos dinâmicos


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

Atenção O Javascript está desativado ou não está disponível no seu navegador.

Para usar a documentação da AWS, o Javascript deve estar ativado. Consulte as páginas de Ajuda do navegador para obter instruções.

Convenções do documento

Amazon EKS

Amazon EMR em EKS