Reaktion auf Timeout-Ereignisse zur Größenänderung der Amazon-EMR-Cluster-Instance-Flotte

Übersicht

Amazon-EMR-Cluster senden Ereignisse aus, während sie die Größenänderung für z. B. Flottencluster ausführen. Die Timeout-Ereignisse für die Bereitstellung werden ausgelöst, wenn Amazon EMR die Bereitstellung von Spot- oder On-Demand-Kapazität für die Flotte nach Ablauf des Timeouts beendet. Die Dauer des Timeouts kann vom Benutzer im Rahmen der Größenänderungsspezifikationen für die Instance-Flotten konfiguriert werden. In Szenarien mit aufeinanderfolgenden Größenänderungen für dieselbe Instance-Flotte gibt Amazon EMR die Ereignisse Spot provisioning timeout - continuing resize oder On-Demand provisioning timeout - continuing resize aus, wenn das Timeout für den aktuellen Größenänderungsvorgang abläuft. Dann beginnt es mit der Bereitstellung von Kapazität für die nächste Größenänderung der Flotte.

Reagieren auf Timeout-Ereignisse zur Größenänderung der Instanceflotte

Es wird empfohlen, dass Sie auf ein Bereitstellungs-Timeout-Ereignis mit einer der folgenden Methoden reagieren:

• GreifenSie die Größenänderungsspezifikationen wieder auf und versuchen Sie erneut, die Größe zu ändern. Da sich die Kapazität häufig ändert, wird die Größe Ihrer Cluster erfolgreich angepasst, sobald EC2 Amazon-Kapazität verfügbar ist. Wir empfehlen unseren Kunden, niedrigere Werte für die Timeout-Dauer für Jobs zu konfigurieren, die strengere Anforderungen stellen. SLAs

• Alternativ können Sie entweder:

  • Einen neuen Cluster mit diversifizierten Instance-Typen auf der Grundlage von bewährten Methoden wie der Flexibilität von Instances und Availability Zones starten oder

  • Einen Cluster mit On-Demand-Kapazität starten

• Für das Ereignis „Timeout bei der Bereitstellung – Fortsetzung der Größenänderung“ können Sie zusätzlich warten, bis die Größenänderungsvorgänge verarbeitet sind. Amazon EMR verarbeitet weiterhin sequentiell die für die Flotte ausgelösten Größenänderungsvorgänge, wobei die konfigurierten Größenänderungsspezifikationen eingehalten werden.

Sie können auch Regeln oder automatische Reaktionen auf dieses Ereignis einrichten, wie im nächsten Abschnitt beschrieben.

Automatisierte Wiederherstellung nach einem Bereitstellungs-Timeout-Ereignis

Mit dem Spot Provisioning timeout-Ereigniscode können Sie als Reaktion auf Amazon-EMR-Ereignisse eine Automatisierung erstellen. Die folgende AWS Lambda -Funktion fährt beispielsweise einen EMR-Cluster mit einer Instance-Flotte herunter, die Spot Instances für Aufgabenknoten verwendet, und erstellt dann einen neuen EMR-Cluster mit einer Instance-Flotte, die vielfältigere Instance-Typen als die ursprüngliche Anfrage enthält. In diesem Beispiel löst das für Aufgabenknoten ausgegebene Spot Provisioning timeout Ereignis die Ausführung der Lambda-Funktion aus.

Beispielfunktion zur Reaktion auf ein Spot Provisioning timeout-Ereignis

// 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")