Risposta agli eventi di timeout di ridimensionamento della flotta di istanze del EMR cluster Amazon Amazon

Panoramica

EMRI cluster Amazon emettono eventi durante l'esecuzione dell'operazione di ridimensionamento, ad esempio i cluster della flotta. Gli eventi di timeout del provisioning vengono emessi quando Amazon EMR interrompe il provisioning di capacità Spot o On-demand per la flotta dopo la scadenza del timeout. La durata del timeout può essere configurata dall'utente come parte delle specifiche di ridimensionamento per i parchi istanze. In scenari di ridimensionamento consecutivo per la stessa flotta di istanze, Amazon EMR emette On-Demand provisioning timeout - continuing resize gli eventi Spot provisioning timeout - continuing resize or alla scadenza del timeout per l'operazione di ridimensionamento corrente. Quindi inizia il provisioning della capacità per la successiva operazione di ridimensionamento del parco istanze.

Risposta agli eventi di timeout del ridimensionamento del parco istanze

Ti consigliamo di rispondere a un evento di timeout del provisioning in uno dei seguenti modi:

• Rivedi le specifiche di ridimensionamento e riprova a eseguire l'operazione di ridimensionamento. Poiché la capacità cambia frequentemente, i cluster verranno ridimensionati correttamente non appena la EC2 capacità di Amazon sarà disponibile. Consigliamo ai clienti di configurare valori inferiori per la durata del timeout per i lavori che richiedono requisiti più rigorosi. SLAs

• In alternativa, puoi:

  • Avviare un nuovo cluster con tipi di istanze diversificati in base alle best practice per la flessibilità dell'istanza e della zona di disponibilità oppure

  • Avviare un cluster con la capacità On-demand

• Per il timeout del provisioning, l'evento di ridimensionamento continua, puoi attendere anche l'elaborazione delle operazioni di ridimensionamento. Amazon EMR continuerà a elaborare in sequenza le operazioni di ridimensionamento attivate per la flotta, rispettando le specifiche di ridimensionamento configurate.

Puoi anche impostare regole o risposte automatiche a questo evento come descritto nella sezione successiva.

Ripristino automatico da un evento di timeout del provisioning

Puoi creare automazione in risposta agli EMR eventi di Amazon con il codice Spot Provisioning timeout evento. Ad esempio, la seguente AWS Lambda funzione chiude un EMR cluster con un parco istanze che utilizza istanze Spot per i nodi Task, quindi crea un nuovo EMR cluster con un parco istanze che contiene tipi di istanze più diversificati rispetto alla richiesta originale. In questo esempio, l'evento Spot Provisioning timeout emesso per i nodi attività attiverà l'esecuzione della funzione Lambda.

Esempio Funzione di esempio per rispondere all'evento Spot Provisioning timeout

// Lambda code with Python 3.10 and handler is lambda_function.lambda_handler
// Note: related IAM role requires permission to use Amazon EMR
 
import json
import boto3
import datetime
from datetime import timezone
 
SPOT_PROVISIONING_TIMEOUT_EXCEPTION_DETAIL_TYPE = "EMR Instance Fleet Resize"
SPOT_PROVISIONING_TIMEOUT_EXCEPTION_EVENT_CODE = (
    "Spot Provisioning timeout"
)
 
CLIENT = boto3.client("emr", region_name="us-east-1")
 
# checks if the incoming event is 'EMR Instance Fleet Resize' with eventCode 'Spot provisioning timeout'
def is_spot_provisioning_timeout_event(event):
    if not event["detail"]:
        return False
    else:
        return (
            event["detail-type"] == SPOT_PROVISIONING_TIMEOUT_EXCEPTION_DETAIL_TYPE
            and event["detail"]["eventCode"]
            == SPOT_PROVISIONING_TIMEOUT_EXCEPTION_EVENT_CODE
        )
 
 
# checks if the cluster is eligible for termination
def is_cluster_eligible_for_termination(event, describeClusterResponse):
    # instanceFleetType could be CORE, MASTER OR TASK
    instanceFleetType = event["detail"]["instanceFleetType"]
 
    # Check if instance fleet receiving Spot provisioning timeout event is TASK
    if (instanceFleetType == "TASK"):
        return True
    else:
        return False
 
 
# create a new cluster by choosing different InstanceType.
def create_cluster(event):
    # instanceFleetType cloud be CORE, MASTER OR TASK
    instanceFleetType = event["detail"]["instanceFleetType"]
 
    # the following two lines assumes that the customer that created the cluster already knows which instance types they use in original request
    instanceTypesFromOriginalRequestMaster = "m5.xlarge"
    instanceTypesFromOriginalRequestCore = "m5.xlarge"
   
    # select new instance types to include in the new createCluster request
    instanceTypesForTask = [
        "m5.xlarge",
        "m5.2xlarge",
        "m5.4xlarge",
        "m5.8xlarge",
        "m5.12xlarge"
    ]
    
    print("Starting to create cluster...")
    instances = {
        "InstanceFleets": [
            {
                "InstanceFleetType":"MASTER",
                "TargetOnDemandCapacity":1,
                "TargetSpotCapacity":0,
                "InstanceTypeConfigs":[
                    {
                        'InstanceType': instanceTypesFromOriginalRequestMaster,
                        "WeightedCapacity":1,
                    }
                ]
            },
            {
                "InstanceFleetType":"CORE",
                "TargetOnDemandCapacity":1,
                "TargetSpotCapacity":0,
                "InstanceTypeConfigs":[
                    {
                        'InstanceType': instanceTypesFromOriginalRequestCore,
                        "WeightedCapacity":1,
                    }
                ]
            },
            {
                "InstanceFleetType":"TASK",
                "TargetOnDemandCapacity":0,
                "TargetSpotCapacity":100,
                "LaunchSpecifications":{},
                "InstanceTypeConfigs":[
                    {
                        'InstanceType': instanceTypesForTask[0],
                        "WeightedCapacity":1,
                    },
                    {
                        'InstanceType': instanceTypesForTask[1],
                        "WeightedCapacity":2,
                    },
                    {
                        'InstanceType': instanceTypesForTask[2],
                        "WeightedCapacity":4,
                    },
                    {
                        'InstanceType': instanceTypesForTask[3],
                        "WeightedCapacity":8,
                    },
                    {
                        'InstanceType': instanceTypesForTask[4],
                        "WeightedCapacity":12,
                    }
                ],
                "ResizeSpecifications": {
                    "SpotResizeSpecification": {
                        "TimeoutDurationMinutes": 30
                    }
                }
            }
        ]
    }
    response = CLIENT.run_job_flow(
        Name="Test Cluster",
        Instances=instances,
        VisibleToAllUsers=True,
        JobFlowRole="EMR_EC2_DefaultRole",
        ServiceRole="EMR_DefaultRole",
        ReleaseLabel="emr-6.10.0",
    )
 
    return response["JobFlowId"]
 
 
# terminated the cluster using clusterId received in an event
def terminate_cluster(event):
    print("Trying to terminate cluster, clusterId: " + event["detail"]["clusterId"])
    response = CLIENT.terminate_job_flows(JobFlowIds=[event["detail"]["clusterId"]])
    print(f"Terminate cluster response: {response}")
 
 
def describe_cluster(event):
    response = CLIENT.describe_cluster(ClusterId=event["detail"]["clusterId"])
    return response
 
 
def lambda_handler(event, context):
    if is_spot_provisioning_timeout_event(event):
        print(
            "Received spot provisioning timeout event for instanceFleet, clusterId: "
            + event["detail"]["clusterId"]
        )
 
        describeClusterResponse = describe_cluster(event)
 
        shouldTerminateCluster = is_cluster_eligible_for_termination(
            event, describeClusterResponse
        )
        if shouldTerminateCluster:
            terminate_cluster(event)
 
            clusterId = create_cluster(event)
            print("Created a new cluster, clusterId: " + clusterId)
        else:
            print(
                "Cluster is not eligible for termination, clusterId: "
                + event["detail"]["clusterId"]
            )
 
    else:
        print("Received event is not spot provisioning timeout event, skipping")