Tipps zur Abstimmung von Spark-Jobs

In den folgenden Abschnitten finden Sie Tipps zur Feinabstimmung Ihrer Spark-Anwendungen in Managed Service for Apache Spark.

Sitzungsspezifische Cluster verwenden

Wenn Sie das Modell für temporäre Cluster von Managed Service for Apache Spark verwenden, erstellen Sie für jeden Job einen dedizierten Cluster und löschen den Cluster, wenn der Job abgeschlossen ist. Mit dem sitzungsspezifischen Modell können Sie Speicher und Computing separat behandeln. Sie speichern die Eingabe- und Ausgabedaten des Jobs in Cloud Storage oder BigQuery und verwenden den Cluster nur für Computing und temporäre Datenspeicherung.

Probleme mit nichtflüchtigen Clustern

Durch die Verwendung sitzungsspezifischer Cluster mit einem Job werden die folgenden Fallstricke und potenziellen Probleme vermieden, die mit der Verwendung freigegebener und langlebiger „persistenter“ Cluster verbunden sind:

  • Single Points of Failure: Ein Fehlerstatus des gemeinsam genutzten Clusters kann dazu führen, dass alle Jobs fehlschlagen und eine gesamte Datenpipeline blockiert wird. Die Untersuchung und Behebung eines Fehlers kann Stunden dauern. Da sitzungsspezifische Cluster nur temporäre clusterinterne Status beibehalten, können sie bei einem Fehler schnell gelöscht und neu erstellt werden.
  • Schwierigkeiten bei der Wartung und Migration von Clusterstatus in HDFS, MySQL oder lokalen Dateisystemen
  • Ressourcenkonflikte zwischen Jobs, die sich negativ auf SLOs auswirken
  • Nicht reagierende Dienst-Daemons aufgrund von Speicherdruck
  • Ansammlung von Protokollen und temporären Dateien, die die Laufwerkskapazität überschreiten können
  • Fehler beim Hochskalieren aufgrund von fehlendem Bestand in der Clusterzone
  • Keine Unterstützung für veraltete Cluster-Image-Versionen.

Vorteile von sitzungsspezifischen Clustern

Kurzlebige Cluster bieten folgende Vorteile:

  • Konfigurieren Sie verschiedene IAM-Berechtigungen für verschiedene Jobs mit unterschiedlichen VM-Dienstkonten für Managed Service for Apache Spark.
  • Hardware- und Softwarekonfigurationen eines Clusters für jeden Job optimieren und Clusterkonfigurationen nach Bedarf ändern.
  • Führen Sie ein Upgrade der Image-Versionen in neuen Clustern durch, um die neuesten Sicherheitspatches, Fehlerkorrekturen und Optimierungen zu erhalten.
  • Probleme in einem isolierten Cluster mit nur einem Job schneller beheben.
  • Sie sparen Kosten, da Sie nur für die Laufzeit des sitzungsspezifischen Clusters bezahlen, nicht für die Leerlaufzeit zwischen Jobs in einem freigegebenen Cluster.

Spark SQL nutzen

Die Spark SQL DataFrame API ist eine erhebliche Optimierung der RDD API. Wenn Sie mit Code interagieren, der RDDs verwendet, sollten Sie Daten als DataFrame lesen, bevor Sie einen RDD im Code übergeben. In Java- oder Scala-Code ist eine sinnvolle Verwendung der Dataset API Spark SQL als Obermenge von RDDs und DataFrames.

Apache Spark 3 nutzen

Managed Service for Apache Spark 2.0 installiert Spark 3, das die folgenden Funktionen und Leistungsverbesserungen enthält:

  • GPU-Unterstützung
  • Möglichkeit, Binärdateien zu lesen
  • Leistungsverbesserungen
  • Dynamische Partitionsbereinigung
  • Adaptive Abfrageausführung: Spark-Jobs werden in Echtzeit optimiert

Dynamische Zuordnung nutzen

Apache Spark enthält ein Feature zur dynamischen Zuordnung, das die Anzahl der Spark-Executors auf Workern in einem Cluster skaliert. Mit dieser Funktion kann ein Job den gesamten Managed Service for Apache Spark-Cluster verwenden, auch wenn der Cluster hochskaliert wird. Dieses Feature ist in Managed Service for Apache Spark standardmäßig aktiviert (spark.dynamicAllocation.enabled ist auf true festgelegt). Weitere Informationen finden Sie unter Dynamische Spark-Zuweisung.

Autoscaling von Managed Service for Apache Spark verwenden

Mit dem Autoscaling von Managed Service for Apache Spark werden Managed Service for Apache Spark-Worker dynamisch zu einem Cluster hinzugefügt und daraus entfernt, damit Spark-Jobs die erforderlichen Ressourcen haben, um schnell abgeschlossen zu werden.

Als Best Practice wird empfohlen, die Autoscaling-Richtlinie so zu konfigurieren, dass nur sekundäre Worker skaliert werden.

Erweiterten Flexibilitätsmodus von Managed Service for Apache Spark verwenden

Cluster mit VMs auf Abruf oder einer Autoscaling-Richtlinie können FetchFailed-Ausnahmen erhalten, wenn Worker vorzeitig beendet oder entfernt werden, bevor sie die Bereitstellung von Shuffle-Daten an Reducer beenden. Diese Ausnahme kann zu Wiederholungen von Aufgaben und längeren Abschlusszeiten führen.

Empfehlung: Verwenden Sie den Enhanced Flexibility Mode von Managed Service for Apache Spark, bei dem keine Zwischen-Shuffle-Daten auf sekundären Workern gespeichert werden, sodass sekundäre Worker sicher vorzeitig beendet oder herunterskaliert werden können.

Partitionierung und Daten nach dem Zufallsprinzip konfigurieren

Spark speichert Daten in temporären Partitionen im Cluster. Wenn Ihre Anwendung DataFrames gruppiert oder zusammenführt, werden die Daten gemäß der Gruppierung und Konfiguration auf niedriger Ebene in neuen Partitionen nach dem Zufallsprinzip verteilt.

Die Datenpartitionierung beeinträchtigt die Anwendungsleistung erheblich: Zu wenige Partitionen begrenzen die Jobparallelität und die Nutzung von Clusterressourcen. Zu viele Partitionen verlangsamen den Job aufgrund zusätzlicher Partitionsverarbeitung und Shuffling.

Partitionen konfigurieren

Die folgenden Attribute bestimmen die Anzahl und Größe Ihrer Partitionen:

  • spark.sql.files.maxPartitionBytes: Die maximale Größe der Partitionen, wenn Sie Daten aus Cloud Storage lesen. Der Standardwert ist 128 MB. Dieser ist für die meisten Anwendungen, die weniger als 100 TB verarbeiten, ausreichend.

  • spark.sql.shuffle.partitions: Die Anzahl der Partitionen nach dem Shuffle-Vorgang. Der Standardwert ist 1000 für Cluster mit der Image-Version 2.2 und höher. Empfehlung: Legen Sie diesen Wert auf das Dreifache der Anzahl der vCPUs in Ihrem Cluster fest.

  • spark.default.parallelism: Die Anzahl der Partitionen, die nach der Ausführung von RDD-Transformationen zurückgegeben werden, die Shuffles erfordern, z. B. join, reduceByKey und parallelize. Der Standardwert ist die Gesamtzahl der vCPUs in Ihrem Cluster. Wenn Sie RDDs in Spark-Jobs verwenden, können Sie diesen Wert auf das Dreifache Ihrer vCPUs festlegen.

Anzahl der Dateien begrenzen

Wenn Spark eine große Anzahl kleiner Dateien liest, kommt es zu einem Leistungsverlust. Speichern Sie Daten in größeren Dateien, z. B. in Dateien mit einer Größe von 256 MB bis 512 MB. Begrenzen Sie auch die Anzahl der Ausgabedateien (Informationen zum Erzwingen eines Shuffle finden Sie unter Unnötige Shuffles vermeiden).

Adaptive Abfrageausführung konfigurieren (Spark 3)

Die adaptive Abfrageausführung (standardmäßig in Managed Service for Apache Spark-Image-Version 2.0 aktiviert) bietet Leistungsverbesserungen für Spark-Jobs, darunter:

Obwohl die Standardkonfigurationseinstellungen für die meisten Anwendungsfälle klingen, kann es sinnvoll sein, spark.sql.adaptive.advisoryPartitionSizeInBytes auf spark.sql.files.maxPartitionBytes (Standard: 128 MB) zu setzen.

Unnötige Shuffles vermeiden

Mit Spark können Nutzer ein Shuffle manuell auslösen, um ihre Daten mit der Funktion repartition neu auszubalancieren. Shuffle sind teuer, daher sollten Daten neu gemischt werden. Die entsprechende Konfiguration der Partitionen sollte ausreichen, damit Spark die Daten automatisch partitionieren kann.

Ausnahme: Beim Schreiben von spaltenpartitionierten Daten in Cloud Storage wird beim Neupartitionieren einer bestimmten Spalte vermieden, dass viele kleine Dateien geschrieben werden, um schnellere Schreibzeiten zu erzielen. 

df.repartition("col_name").write().partitionBy("col_name").save("gs://...")

Daten in Parquet oder Avro speichern

Spark SQL verwendet standardmäßig Lese- und Schreibzugriff auf Daten in komprimierten Snappy-Parquet-Dateien. Parquet hat ein effizientes Dateiformat, das es Spark ermöglicht, nur die Daten zu lesen, die zum Ausführen einer Anwendung benötigt werden. Dies ist ein wichtiger Vorteil bei der Arbeit mit großen Datasets. Andere Spaltenformate wie Apache ORC sind ebenfalls gut geeignet.

Für nicht spaltenorientierte Daten bietet Apache Avro ein effizientes Dateiformat für binäre Zeilen. Obwohl Avro in der Regel langsamer als Parquet ist, ist die Leistung besser als bei textbasierten Formaten wie CSV oder JSON.

Laufwerkgröße optimieren

Der Durchsatz nichtflüchtiger Speicher hängt von der Laufwerkgröße ab, was sich auf die Leistung des Spark-Jobs auswirken kann, da Jobs Metadaten schreiben und Daten auf das Laufwerk verschieben. Bei Verwendung von nichtflüchtigen Standardspeichern sollte die Laufwerkgröße mindestens 1 Terabyte pro Worker betragen (siehe Leistung nach Größe des nichtflüchtigen Speichers).

So überwachen Sie den Laufwerksdurchsatz von Workern in derGoogle Cloud Console:

  1. Klicken Sie auf der Seite Cluster auf den Clusternamen.
  2. Klicken Sie auf den Tab VM-INSTANZEN.
  3. Klicken Sie auf den Namen eines beliebigen Workers.
  4. Klicken Sie auf den Tab MONITORING und scrollen Sie dann zu „Laufwerkdurchsatz“, um den Worker-Durchsatz aufzurufen.

Hinweise zu Laufwerken

Sitzungsspezifische Managed Service for Apache Spark-Cluster, die nicht von nichtflüchtigem Speicher profitieren, können lokale SSDs verwenden. Lokale SSDs sind physisch mit dem Cluster verbunden und bieten einen höheren Durchsatz als nichtflüchtige Speicher (siehe Leistungstabelle). Lokale SSDs sind mit einer festen Größe von 375 Gigabyte verfügbar. Sie können jedoch mehrere SSDs hinzufügen, um die Leistung zu erhöhen.

Auf lokalen SSDs werden Daten nach dem Herunterfahren eines Clusters nicht beibehalten. Wenn Sie nichtflüchtigen Speicher benötigen, können Sie nichtflüchtige SSD-Speicher verwenden, die für ihre Größe einen höheren Durchsatz als nichtflüchtige Standardspeicher bieten. Nichtflüchtige SSD-Speicher sind auch eine gute Wahl, wenn die Partitionsgröße kleiner als 8 KB ist. Kleine Partitionen sollten jedoch vermieden werden.

GPUs Ihrem Cluster hinzufügen

Spark 3 unterstützt GPUs. Verwenden Sie GPUs mit der CUDA-X-Initialisierungsaktion, um Spark-Jobs mit der NVIDIA cuDF for Apache Spark-Bibliothek zu beschleunigen. Die GPU-Treiberinitialisierungsaktion zum Konfigurieren eines Clusters mit GPUs.

Häufige Jobfehler und -korrekturen

Im folgenden Abschnitt werden häufige Ursachen für Jobfehler und Lösungen dafür erläutert.

Nicht genügend Arbeitsspeicher

Beispiele:

  • "Lost Executor"
  • "java.lang.OutOfMemoryError: GC-Overhead überschritten"
  • "Container, der von YARN beendet wurde, um Speicherlimits zu überschreiten"

Mögliche Korrekturen:

Shuffle-Abruffehler

Beispiele:

  • "FetchFailedException" (Spark-Fehler)
  • "Verbindung konnte nicht hergestellt werden..." (Spark-Fehler)
  • "Fehler beim Abrufen" (MapReduce-Fehler)

Normalerweise verursacht durch das vorzeitige Entfernen von Workern, die noch Shuffle-Daten für die Bereitstellung haben.

Mögliche Ursachen und Lösungen:

  • Worker-VMs auf Abruf wurden zurückgefordert oder nicht auf Abruf verwendete Worker-VMs wurden vom Autoscaling entfernt. Lösung: Verwenden Sie den Enhanced Flexibility Mode, um sekundäre Worker sicher auf Abruf oder skalierbar zu machen.
  • Der Executor oder Mapper ist aufgrund eines OutOfMemory-Fehlers abgestürzt. Lösung: Erhöhen Sie den Arbeitsspeicher des Executors oder des Mappers.
  • Der Spark-Shuffle-Dienst ist möglicherweise überlastet. Lösung: Reduzieren Sie die Anzahl der Jobpartitionen.

YARN-Knoten sind FEHLERHAFT

Beispiele (aus YARN-Logs): 

...reported UNHEALTHY with details: 1/1 local-dirs usable space is below
configured utilization percentage/no more usable space
[ /hadoop/yarn/nm-local-dir : used space above threshold of 90.0% ]

Häufig mit unzureichendem Speicherplatz für Shuffle-Daten verbunden. Diagnose durch Aufrufen der Logdateien:

  • Öffnen Sie in der Google Cloud Console die Seite Cluster Ihres Projekts und klicken Sie dann auf den Namen des Clusters.
  • Klicken Sie auf LOGS ANSEHEN.
  • Logs nach hadoop-yarn-nodemanager filtern.
  • Suchen Sie nach "FEHLERHAFT".

Mögliche Lösungen:

  • Der Nutzercache wird in dem Verzeichnis gespeichert, das durch das Attribut yarn.nodemanager.local-dirs in der yarn-site.xml file angegeben ist. Diese Datei befindet sich unter /etc/hadoop/conf/yarn-site.xml. Sie können den freien Speicherplatz im Pfad /hadoop/yarn/nm-local-dir prüfen und Speicherplatz freigeben, indem Sie den Nutzer-Cache-Ordner /hadoop/yarn/nm-local-dir/usercache löschen.
  • Wenn im Log der Status „UNHEALTHY“ angezeigt wird, erstellen Sie den Cluster mit mehr Speicherplatz neu. Dadurch wird die Durchsatzobergrenze erhöht.

Nutzer-Cache belegt Speicherplatz

In Managed Service for Apache Spark wird der Nutzercache in dem Verzeichnis gespeichert, das durch das Attribut yarn.nodemanager.local-dirs in yarn-site.xml (/etc/hadoop/conf/yarn-site.xml) angegeben ist. In den meisten Fällen ist der Speicherort /hadoop/yarn/nm-local-dir.

Mögliche Lösungen:

  • Manuelle Bereinigung: Prüfen Sie den verfügbaren Speicherplatz im Pfad /hadoop/yarn/nm-local-dir und schaffen Sie Speicherplatz, indem Sie den Cache-Ordner des Nutzers /hadoop/yarn/nm-local-dir/usercache löschen.

  • Langfristige Lösung: Verwalten Sie den Bereinigungsprozess, indem Sie die folgenden Clusterattribute festlegen:

    • yarn.nodemanager.localizer.cache.cleanup.interval-ms
    • yarn.nodemanager.localizer.cache.target-size-mb

    Wenn Sie den Cluster nicht neu erstellen möchten, können Sie diese Attribute in yarn-site.xml auf allen Knoten aktualisieren und dann den NodeManager neu starten. Führen Sie dazu den folgenden Befehl auf den Worker-Knoten aus:

    sudo systemctl restart hadoop-yarn-nodemanager.service

  • Größe des Clusters anpassen: Wenn im Log der Status „UNHEALTHY“ gemeldet wird, erstellen Sie den Cluster mit mehr Speicherplatz neu. Dadurch wird die Durchsatzobergrenze erhöht.

Job schlägt aufgrund von unzureichendem Treiberspeicher fehl

Wenn Sie Jobs im Clustermodus ausführen, schlägt der Job fehl, wenn die Arbeitsspeichergröße des Worker-Knotens überschritten wird.

Beispiel aus Treiberlogs: 

'Exception in thread "main" java.lang.IllegalArgumentException:
Required AM memory (32768+3276 MB) is above the max threshold (12288 MB) of this cluster!
Please check the values of 'yarn.scheduler.maximum -allocation-mb' and/or 'yarn.nodemanager.resource.memory-mb'.'

Mögliche Lösungen:

  • Legen Sie für spark:spark.driver.memory einen Wert fest, der kleiner als yarn:yarn.scheduler.maximum-allocation-mb ist.
  • Verwenden Sie für Master- und Worker-Knoten denselben Maschinentyp.

Nächste Schritte