Arbeitslastressourcen automatisch skalieren

Mit dem Autoscaling von Managed Service for Apache Spark können Sie Arbeitslastressourcen wie die Anzahl der Executors dynamisch skalieren, um Spark-Jobs effizient auszuführen. Das standardmäßig aktivierte Autoscaling verwendet die dynamische Ressourcenzuweisung von Spark, um zu bestimmen, ob, wie und wann Ihre Arbeitslast skaliert werden soll.

Managed Service for Apache Spark – Autoscaling V2

Version 2 (V2) des Autoscalings für Managed Service for Apache Spark bietet zusätzliche Funktionen und Verbesserungen gegenüber der Standardversion 1 (V1). Damit können Sie Managed Service for Apache Spark-Arbeitslasten verwalten, die Leistung von Arbeitslasten verbessern und Kosten sparen:

  • Asynchrones Herunterskalieren von Knoten: Autoscaling V2 ersetzt das synchrone Herunterskalieren von V1 durch asynchrones Herunterskalieren. Beim asynchronen Downscaling werden die Arbeitslastressourcen von Managed Service for Apache Spark herunterskaliert, ohne dass darauf gewartet wird, dass alle Knoten die Shuffle-Migration abgeschlossen haben. Das bedeutet, dass Longtail-Knoten, die langsam herunterskaliert werden, das Hochskalieren nicht blockieren.
  • Intelligente Knotenauswahl für das Herunterskalieren: Autoscaling V2 ersetzt die zufällige Knotenauswahl von V1 durch einen intelligenten Algorithmus, der die besten Knoten für das Herunterskalieren zuerst identifiziert. Dieser Algorithmus berücksichtigt Faktoren wie die Größe der Shuffle-Daten des Knotens und die Inaktivitätszeit.
  • Konfigurierbares Verhalten für die ordnungsgemäße Außerbetriebnahme und Shuffle-Migration von Spark: Mit Autoscaling V2 können Sie Standard-Spark-Attribute verwenden, um die ordnungsgemäße Außerbetriebnahme und Shuffle-Migration von Spark zu konfigurieren. Mit dieser Funktion können Sie die Migrationskompatibilität mit Ihren benutzerdefinierten Spark-Properties aufrechterhalten.

Autoscaling-Funktionen von Managed Service for Apache Spark

Funktion Managed Service for Apache Spark Autoscaling V1 Managed Service for Apache Spark Autoscaling V2
Knoten herunterskalieren Synchron Asynchron
Knotenauswahl für die Herunterskalierung Zufällig Intelligent
Ordnungsgemäße Außerbetriebnahme und Shuffle-Migration von Spark Nicht konfigurierbar Konfigurierbar

Spark-Attribute für die dynamische Zuordnung

In der folgenden Tabelle sind die Attribute für die dynamische Spark-Zuweisung aufgeführt, die Sie beim Senden einer Batcharbeitslast festlegen können, um das Autoscaling zu steuern (siehe Spark-Attribute festlegen).

Attribut Beschreibung Standard
spark.dataproc.scaling.version Die Autoscaling-Version von Managed Service for Apache Spark. Geben Sie die Version 1 oder 2 an. 1
spark.dynamicAllocation.enabled Gibt an, ob die dynamische Ressourcenzuweisung verwendet werden soll, bei der die Anzahl der Executors je nach Arbeitslast hoch- und herunterskaliert wird. Wenn Sie den Wert auf false festlegen, wird das Autoscaling für die Arbeitslast deaktiviert. Standardeinstellung: true true
spark.dynamicAllocation.initialExecutors Die anfängliche Anzahl der Executors, die der Arbeitslast zugewiesen sind. Nach dem Start der Arbeitslast kann die Anzahl der aktiven Executors durch Autoscaling geändert werden. Der Mindestwert beträgt 2, der Höchstwert 2000. 2
spark.dynamicAllocation.minExecutors Die Mindestanzahl von Executors, auf die die Arbeitslast herunterskaliert werden soll. Der Mindestwert beträgt 2. 2
spark.dynamicAllocation.maxExecutors Die maximale Anzahl von Executors, auf die die Arbeitslast hochskaliert werden soll. Der Höchstwert beträgt 2000. 1000
spark.dynamicAllocation.executorAllocationRatio Passt die Hochskalierung der Spark-Arbeitslast an. Akzeptiert einen Wert von 0 bis 1. Der Wert 1.0 bietet die maximale Hochskalierungsfunktion und hilft, die maximale Parallelität zu erreichen. Bei einem Wert von 0.5 werden die Skalierungsfunktion und die Parallelität auf die Hälfte des Maximalwerts festgelegt. 0.3
spark.dynamicAllocation.diagnosis.enabled Wenn true, werden Diagnoseinformationen protokolliert, wenn die Anzahl der ausgeführten Executors die maximale Anzahl der erforderlichen Executors für den durch spark.dynamicAllocation.diagnosis.interval angegebenen Zeitraum überschreitet. Die Diagnose umfasst eine Executor-Zusammenfassung mit der Anzahl der inaktiven Executors und Perzentilen für die Inaktivitätszeit, der Verteilung aktiver Aufgaben, der Größe der Shuffle-Daten und der Größe des im Cache gespeicherten RDD. Verwenden Sie spark.dynamicAllocation.diagnosis.logLevel, um die Logebene der Ausgabe zu steuern. false
spark.dynamicAllocation.profile Legen Sie performance oder cost fest, um einen vordefinierten Satz von Konfigurationen anzuwenden, die für Leistung oder Kosteneffizienz optimiert sind. Benutzerdefinierte Eigenschaften überschreiben die Standardeinstellungen des Profils. Weitere Informationen finden Sie unter Profile für die dynamische Spark-Zuweisung. none
spark.dynamicAllocation.shuffleTracking.dynamicTimeout.enabled Wenn true, wird die dynamische Zeitüberschreitungsberechnung für Executors aktiviert, die Shuffle-Daten enthalten. Anstatt den statischen spark.dynamicAllocation.shuffleTracking.timeout zu verwenden, wird das Zeitlimit anhand der Menge der auf dem Executor gespeicherten Shuffle-Daten berechnet. So können Executors mit kleinen Shuffles schneller freigegeben werden, während Executors mit großen Shuffles länger aktiv bleiben. false
spark.reducer.fetchMigratedShuffle.enabled Wenn dieser Wert auf true gesetzt ist, wird der Speicherort der Shuffle-Ausgabe vom Spark-Treiber abgerufen, nachdem ein Abruf von einem Executor fehlgeschlagen ist, der aufgrund der dynamischen Spark-Zuweisung außer Betrieb genommen wurde. Dadurch werden ExecutorDeadException-Fehler reduziert, die durch die Migration von Shuffle-Blöcken von außer Betrieb genommenen Executors zu aktiven Executors verursacht werden. Außerdem werden Wiederholungsversuche von Phasen reduziert, die durch FetchFailedException-Fehler verursacht werden (siehe FetchFailedException verursacht durch ExecutorDeadException). Dieses Attribut ist in den Spark-Laufzeitversionen 1.1.12 und höher sowie 2.0.20 und höher von Managed Service for Apache Spark verfügbar. false
spark.scheduler.excludeShuffleSkewExecutors Wenn true, wird vermieden, dass Aufgaben auf Executors mit Shuffle-Skew geplant werden. Das sind Executors mit einer großen Menge an Shuffle-Daten oder einer großen Anzahl abgeschlossener Map-Aufgaben. Dadurch kann die Leistung verbessert werden, da die Shuffle-Verzerrung verringert wird. false

Profile für die dynamische Zuordnung von Spark

Sie können das Attribut spark.dynamicAllocation.profile auf performance oder cost festlegen, um einen vordefinierten Satz von Spark-Konfigurationen anzuwenden, die für Leistung oder Kosteneffizienz optimiert sind. Wenn Sie zusätzlich zum Attribut spark.dynamicAllocation.profile Spark-Attribute festlegen, werden die Standardwerte des Profils für diese Attribute durch Ihre Einstellungen überschrieben.

Leistung: Dieses Profil optimiert die Ausführungszeit durch Anwendung der folgenden Standardeinstellungen:

  • spark.scheduler.excludeShuffleSkewExecutors: true
  • spark.dynamicAllocation.executorIdleTimeout: 300s
  • spark.dynamicAllocation.initialExecutors: 10

cost: Dieses Profil optimiert den Ressourcenverbrauch durch die folgenden Standardeinstellungen:

  • spark.dynamicAllocation.executorIdleTimeout: 120s
  • spark.dynamicAllocation.cachedExecutorIdleTimeout: 120s
  • spark.dynamicAllocation.shuffleTracking.dynamicTimeout.enabled: true
  • spark.dynamicAllocation.diagnosis.enabled: true

Messwerte für die dynamische Zuordnung von Spark

Spark-Batcharbeitslasten generieren die folgenden Messwerte im Zusammenhang mit der dynamischen Ressourcenzuweisung von Spark. Weitere Informationen zu Spark-Messwerten finden Sie unter Monitoring und Instrumentierung.

Messwert Beschreibung
maximum-needed Die maximale Anzahl von Executors, die unter der aktuellen Last erforderlich sind, um alle laufenden und ausstehenden Aufgaben zu erfüllen.
running Die Anzahl der laufenden Executors, die Aufgaben ausführen.

Probleme und Lösungen bei der dynamischen Zuordnung von Spark

  • FetchFailedException verursacht durch ExecutorDeadException

    Ursache: Wenn die dynamische Spark-Zuweisung einen Executor herunterskaliert, wird die Shuffle-Datei zu Live-Executoren migriert. Da die Spark-Reducer-Aufgabe auf einem Executor jedoch die Shuffle-Ausgabe von dem Speicherort abruft, der vom Spark-Treiber beim Start der Reducer-Aufgabe festgelegt wurde, kann der Reducer, wenn eine Shuffle-Datei migriert wird, weiterhin versuchen, die Shuffle-Ausgabe von einem außer Betrieb genommenen Executor abzurufen. Dies führt zu ExecutorDeadException- und FetchFailedException-Fehlern.

    Lösung: Aktivieren Sie das erneute Abrufen des Shuffle-Speicherorts, indem Sie spark.reducer.fetchMigratedShuffle.enabled auf true setzen, wenn Sie Ihre Managed Service for Apache Spark-Batcharbeitslast ausführen (siehe Spark-Batcharbeitslast-Attribute festlegen). Wenn dieses Attribut aktiviert ist, ruft die Reducer-Aufgabe den Shuffle-Ausgabespeicherort noch einmal vom Treiber ab, nachdem ein Abruf von einem außer Betrieb genommenen Executor fehlgeschlagen ist.