Best Practices für die Spark-Leistungsoptimierung – IBM Developer

count() für ein Dataset ist eine Spark-Aktion. Es ist ein häufiges Problem, bei dem es in Spark-Anwendungen mehrere count() Aufrufe gibt, die während des Debuggens hinzugefügt und nicht entfernt werden. Es ist eine gute Idee, nach Spark-Aktionen zu suchen und alle zu entfernen, die nicht erforderlich sind, da wir keine CPU-Zyklen und andere Ressourcen verwenden möchten, wenn dies nicht erforderlich ist.

Dateiformate

Stellen Sie beim Entwerfen Ihrer Datasets für Ihre Anwendung sicher, dass Sie die mit Spark verfügbaren Dateiformate optimal nutzen. Einige Dinge zu beachten:

  • Spark ist für Apache Parquet und ORC für den Lesedurchsatz optimiert. Spark bietet Vektorisierungsunterstützung, die Festplatten-E / A reduziert.
  • Verwenden Sie das Parquet-Dateiformat und verwenden Sie die Komprimierung.
  • Es gibt verschiedene Dateiformate und integrierte Datenquellen, die in Apache Spark verwendet werden können.Verwenden Sie teilbare Dateiformate.
  • Stellen Sie sicher, dass nicht zu viele kleine Dateien vorhanden sind. Wenn Sie viele kleine Dateien haben, kann es sinnvoll sein, diese für eine bessere Leistung zu komprimieren.

Parallelität

  • Erhöhen Sie die Anzahl der Spark-Partitionen, um die Parallelität basierend auf der Größe der Daten zu erhöhen. Stellen Sie sicher, dass Cluster-Ressourcen optimal genutzt werden. Zu wenige Partitionen können dazu führen, dass einige Executoren inaktiv sind, während zu viele Partitionen zu einem Overhead bei der Aufgabenplanung führen können.
  • Optimieren Sie die Partitionen und Aufgaben. Spark kann Aufgaben von 100 ms + verarbeiten und empfiehlt mindestens 2-3 Aufgaben pro Kern für einen Executor.
  • Spark legt die Anzahl der Partitionen basierend auf der eingegebenen Dateigröße fest. Manchmal ist es sinnvoll, die Anzahl der Partitionen explizit anzugeben.
    • Die Lese-API akzeptiert eine optionale Anzahl von Partitionen.
    • Funken.SQL.Datei.maxPartitionBytes, verfügbar in Spark v2.0.0, für Parkett, ORC und JSON.
  • Die Shuffle-Partitionen können durch Einstellen von Spark eingestellt werden.SQL.mischen.partitionen, der Standardwert ist 200. Dies ist wirklich klein, wenn Sie große Datensatzgrößen haben.

Shuffle reduzieren

Shuffle ist ein teurer Vorgang, da Daten über die Knoten in Ihrem Cluster verschoben werden müssen. Hier sind einige Tipps, um Shuffle zu reduzieren:

  • Stimmen Sie den Spark ab.SQL.mischen.partitionen.
  • Partitionieren Sie das Eingabedatenset entsprechend, damit jede Aufgabengröße nicht zu groß ist.
  • Verwenden Sie die Spark-Benutzeroberfläche, um den Plan zu studieren und nach Möglichkeiten zu suchen, den Shuffle so weit wie möglich zu reduzieren.
  • Formelempfehlung für spark.SQL.mischen.partitionen:
    • Bei großen Datasets sollten Sie eine Zielgröße von 100 MB bis weniger als 200 MB für eine Partition anstreben (verwenden Sie beispielsweise die Zielgröße von 100 MB).
    • Funken.SQL.mischen.partitionen = Quotient (Shuffle (Eingabegröße / Zielgröße) / Gesamtkerne) * Gesamtkerne.

Datensatzgröße filtern /reduzieren

Suchen Sie nach Möglichkeiten, Daten so früh wie möglich in Ihrer Anwendungspipeline herauszufiltern. Wenn es einen Filtervorgang gibt und Sie nur an einer Analyse für eine Teilmenge der Daten interessiert sind, wenden Sie diesen Filter frühzeitig an. Wenn Sie die Datensatzgröße frühzeitig reduzieren können, tun Sie es. Verwenden Sie geeignete Filterprädikate in Ihrer SQL-Abfrage, damit Spark sie auf die zugrunde liegende Datenquelle übertragen kann. Verwenden Sie sie entsprechend. Verwenden Sie Partitionsfilter, falls zutreffend.

Cache>

Spark unterstützt das Zwischenspeichern von Datensätzen im Speicher. Es stehen verschiedene Optionen zur Verfügung:

  • Verwenden Sie Caching, wenn derselbe Vorgang mehrmals im Pipelinefluss berechnet wird.
  • Verwenden Sie das Caching mit der Persist-API, um die erforderliche Cache-Einstellung zu aktivieren (auf Festplatte persistieren oder nicht; serialisiert oder nicht).
  • Achten Sie auf Lazy Loading und Prime Cache, wenn nötig im Voraus. Einige APIs sind eifrig und andere nicht.
  • Auf der Registerkarte Speicher der Spark-Benutzeroberfläche finden Sie Informationen zu den von Ihnen zwischengespeicherten Datasets.
  • Es empfiehlt sich, das zwischengespeicherte Dataset aufzuheben, wenn Sie es nicht mehr verwenden, um Ressourcen freizugeben, insbesondere wenn andere Personen den Cluster ebenfalls verwenden.

Join

Join ist im Allgemeinen eine teure Operation, also achten Sie auf die Joins in Ihrer Anwendung, um sie zu optimieren. BroadcastHashJoin ist am leistungsfähigsten für Fälle, in denen eine der Relationen klein genug ist, dass sie übertragen werden kann. Im Folgenden finden Sie einige Tipps:

  • Join-Reihenfolge ist wichtig; Beginnen Sie mit dem selektivsten Join. Für Beziehungen kleiner als spark.SQL.autoBroadcastJoinThreshold, können Sie überprüfen, ob Broadcast HashJoin abgeholt wird.
  • Verwenden Sie bei Bedarf SQL-Hinweise, um einen bestimmten Join-Typ zu erzwingen.
    • Beispiel: Wenn Sie einen kleinen Datensatz mit einem großen Datensatz verbinden, wird ein Broadcast-Join möglicherweise gezwungen, den kleinen Datensatz zu übertragen.
    • Bestätigen Sie, dass Spark einen Hash-Join aufnimmt; Wenn nicht, kann man es mit dem SQL-Hinweis erzwingen.
  • Vermeiden Sie Kreuzverbindungen.
  • Broadcast HashJoin ist am leistungsfähigsten, aber möglicherweise nicht anwendbar, wenn beide Beziehungen in Join groß sind.
  • Sammeln Sie Statistiken zu Tabellen für Spark, um einen optimalen Plan zu berechnen.

Clusterressourcen optimieren

  • Optimieren Sie die Ressourcen im Cluster abhängig vom Ressourcenmanager und der Version von Spark.
  • Stimmen Sie den verfügbaren Speicher auf den Treiber ab: spark.Treiber.Speicher.
  • Optimieren Sie die Anzahl der Executoren sowie die Speicher- und Kernauslastung basierend auf den Ressourcen im Cluster: Executor-memory, num-executors und executor-cores.

Lesen Sie die Konfigurationsdokumentation für die Spark-Version, mit der Sie arbeiten, und verwenden Sie die entsprechenden Parameter.

Vermeiden Sie teure Operationen

  • Vermeiden Sie bestellen Sie nach, wenn es nicht benötigt wird.
  • Wenn Sie Ihre Abfragen schreiben, rufen Sie nicht select * , um alle Spalten abzurufen, sondern nur die für Ihre Abfrage relevanten Spalten.
  • Rufen Sie count nicht unnötig auf.

Datenverschiebung

  • Stellen Sie sicher, dass die Partitionen gleich groß sind, um Datenverschiebungen und Probleme mit der geringen CPU-Auslastung zu vermeiden.
    • Beispiel: Wenn Daten parallel von einer JDBC-Datenquelle eingehen und jede dieser Partitionen nicht eine ähnliche Anzahl von Datensätzen abruft, führt dies zu Aufgaben ungleicher Größe (eine Form von Datenverschiebung). Vielleicht ist eine Partition nur ein paar KB, während eine andere ein paar hundert MB ist. Einige Aufgaben sind größer als andere, und während die Executoren für größere Aufgaben beschäftigt sind, werden die anderen Executoren, die die kleinere Aufgabe erledigen, beendet und sind im Leerlauf.
    • Wenn die Daten an der Quelle nicht optimal partitioniert sind, können Sie auch die Kompromisse bei der Verwendung der Neupartitionierung bewerten, um eine ausgewogene Partition zu erhalten, und sie dann gegebenenfalls im Speicher zwischenspeichern.
  • Die Neupartitionierung führt zu einem Shuffle, und Shuffle ist eine teure Operation.
  • Verwenden Sie die Spark-Benutzeroberfläche, um nach den Partitionsgrößen und der Aufgabendauer zu suchen.

UDFs

Spark verfügt über eine Reihe integrierter benutzerdefinierter Funktionen (UDFs). Überprüfen Sie für die Leistung, ob Sie eine der integrierten Funktionen verwenden können, da diese für die Leistung gut sind. Benutzerdefinierte UDFs in der Scala-API sind leistungsfähiger als Python-UDFs. Wenn Sie die Python-API verwenden müssen, verwenden Sie die neu eingeführte Pandas-UDF in Python, die in Spark 2.3 veröffentlicht wurde. Die Pandas UDF-Unterstützung (Vectorized UDFs) in Spark bietet erhebliche Leistungsverbesserungen im Gegensatz zum Schreiben einer benutzerdefinierten Python-UDF. Weitere Informationen zum Schreiben einer Pandas-UDF.

Ich hoffe, dies war hilfreich für Sie beim Schreiben Ihrer Spark-Anwendungen. Glückliche Entwicklung! In einem kommenden Blog werde ich zeigen, wie Sie den Ausführungsplan für Ihren Spark-Job erhalten.

Schreibe einen Kommentar

Deine E-Mail-Adresse wird nicht veröffentlicht.