Optimierungsstrategien für Apache Spark verstehen und anwenden

Motivatoren für das Lesen dieses Artikels.

• Eigene Erfahrung in Momenten des Chaos und Momenten ruhiger Analyse.
• Wonach ich gesucht habe, um tiefer einzutauchen.
• Was ich darüber gelernt habe, wie Spark zur Optimierung funktioniert.
• Was ist das „Plus“ an Databricks zur Optimierung?
• Gute Praktiken, die die Notwendigkeit einer Optimierung und Umgestaltung vermeiden können.

Einführung

Ich hatte schon immer guten Kontakt zu relationalen Datenbanken und später zu verteilten Systemen wie Spark. Zunächst beschäftigte ich mich tiefer mit dem DBMS, sowohl um komplexe Abfragen einzurichten, zu verwalten als auch vor allem, wie man ein performatives Skript für das DBMS zusammenstellt. Als ich anfing, mehr mit Spark und später mit Databricks zu arbeiten, hatte ich zunächst keine Leistungsprobleme für die Szenarien, die ich erstellen musste, aber als der BigData-Bereich wirklich zu BigData wurde, bekam ich Leistungsprobleme in Routinen, die jeweils um 30 % zunahmen Woche und das brachte mich dazu, nach der Funktionsweise von Spark „unter der Haube“ zu suchen, vor allem, weil ich bereits wusste, wie ein DBMS funktioniert, und dies half mir, einige Konzepte zu verstehen, die ich hier vorstellen werde.

Kurze Erläuterung der Apache Spark-Komponenten

Lassen Sie es uns kurz fassen, da ich möchte, dass sich dieser Artikel auf Leistungsanalyseszenarien, -techniken und Best Practices konzentriert.

SparkCore:

Diese Komponente ist die Basis von Spark. Sie ist für die Speicherverwaltung, Aufgaben, Fehlerbehebung und E/A-Verwaltung verantwortlich, mit anderen Worten, sie manipuliert das RDD. Daher ist er ein Typ, der einen großen Teil des Clusters hat.

Ausführende:

Diese Komponente ist der eigentliche Arbeiter des Spark-Ökosystems (Clusters), sie ist derjenige, der die Schreib- oder Lesebefehle (Aufgaben) erhält, die auf der Festplatte, im Speicher oder auf beiden sein können (ich werde später erklären, warum das ins Spiel kommt). spielen). Aufführungsszenarien).

Arbeiter:

Worker sind im wahrsten Sinne des Wortes das, was sie für diejenigen sind, die sich bereits mit verteiltem Computing auskennen. Sie sind die Knoten des Clusters, also „hosten“ sie die Executoren, die ich oben erwähnt habe. Jeder Worker kann einen oder mehrere Executoren enthalten. Es ist für die Verwaltung der den Testamentsvollstreckern zugewiesenen Ressourcen verantwortlich, als ob der Testamentsvollstrecker ein Assistent und der Arbeiter ein Lagerarbeiter wäre. Was ist, wenn er der Lagerarbeiter ist, dem er Bericht erstattet?

Clustermanager:

Dies ist der Manager. Er verwaltet die Ressourcen (Speicher und CPU) für die Arbeiter. Er ist derjenige, der entscheidet, wie viele Ausführende für jede Anwendung vorhanden sind und wie viele Ressourcen zugewiesen werden. Er verwaltet die von ihm gesendeten Aufgaben. „Chef“, den ich weiter unten erläutern werde, und da es sich um eine Position mit höherer Verantwortung handelt, überwacht er auch den Zustand des Clusters, um sich nach Ausfällen zu erholen und Aufgaben bei Bedarf neu zu verteilen. (HINWEIS: Es gibt verschiedene Arten von Cluster-Managern: Yarn, Mesos, Kubernetes und die einfachste, die eigenständig ist).

SparkContext:

Nun, das ist der Boss oder das Gateway, ich sage Gateway, weil jede Spark-Anwendung ihn durchläuft, es ist das, was der Anwendung ermöglicht, mit dem Cluster, also den Workern und Ausführenden, zu interagieren, es ermöglicht und verwaltet Es verteilt Aufgaben zwischen Workern und verwaltet auf diese Weise die gesamte Anwendung hinsichtlich Konfiguration, Anzahl der Ausführenden und Ressourcen wie Speicher. Möchten Sie wissen, wie Aufgaben ausgeführt werden? Sprechen Sie hier mit diesem Chef.

Also mal anschaulich:

Lassen Sie uns nun über Leistung, Tuning, Geschwindigkeit, Geschwindigkeit und alles, was Sie aus verschiedenen Positionen hören, sprechen.

Als ich mit der relationalen Banking-Seite gearbeitet habe und es Leistungsprobleme gab, hauptsächlich bei Prozeduren oder Funktionen oder einer Abfrage in einer Anwendung, habe ich die folgenden Aspekte analysiert:

1. Wann läuft dieses Skript und wie ist der Server derzeit?
2. Konkurriert jemand um Ressourcen oder Tabellensperren?
3. Alles läuft reibungslos, niemand blockiert (blockiert) die Serverressourcen, alles ist gut, großartig...
4. Lassen Sie mich nun das Drehbuch sehen. Ist seine Logik performativ? Mit anderen Worten: Wer hat jemals über das gemeinsame Lesen/Schreiben nachgedacht oder Zeile für Zeile darüber nachgedacht (Programmiersucht), konsultiert zu viele Spalten, die er nicht brauchte, monströse Abfragen mit Unterabfragen, CTE usw.? Ich habe alle diese Punkte modifiziert (Refactoring) und sowohl die Geschwindigkeit der Antwort als auch die Nutzung der Serverressourcen getestet. Warum erkläre ich das alles, wenn wir über Apache Spark sprechen? Also ... das gilt auch für Spark und ist meiner Meinung nach sogar noch komplexer, aber wir werden es schaffen.
5. Ich denke, wenn das Skript zum Schluss gut wäre, würde ich den „Weg der Steine“ analysieren, also den geschätzten Ausführungsplan und den tatsächlichen Ausführungsplan. Daraus konnte ich verstehen, was das DBMS mit seinen Statistiken (Histogramm) machte und welchen Weg es mit seinen Informationen einschlagen wollte und wie die Realität aussah, welcher Weg eingeschlagen wurde. Und dann könnten Sie Punkte identifizieren wie: einen zusätzlichen Filter in der Abfrage, einen leistungsfähigeren JOIN und sogar die Erstellung eines Index oder temporärer Tabellen.

Nun, ich denke, das war's. Was haben diese Punkte nun mit Apache Spark gemeinsam?

• Skript ist nicht für die Manipulation verteilter Mengen konzipiert (ich sagte, dass Spark ein „Plus“ an Schwierigkeitsgrad hat, lol).
• Zeit, in der eine bestimmte Routine ausgeführt wird, wenn ein einfacher Spark-Job im selben Cluster ausgeführt wird wie ein anderer ausführender Job (oder auch nicht), der alle Ressourcen verbraucht. (Sehen Sie sich hier eine Art berühmte DBMS-Sperre an).
• Und schließlich, ja, Apache Spark hat einen Ausführungsplan, genauer gesagt, er besteht aus den folgenden Phasen:

1. Logischer Plan.
2. Physische Ebene.
3. Ausführungsstrategie.
4. Zeigt manchmal die geschätzten Kosten an.

Um zusammenzufassen, was jedes einzelne ist, trotz des Namens, können Sie sich bereits eine Vorstellung davon machen:

Logischer Plan:
Stellt die ursprüngliche Abfrage als eine Reihe logischer Operationen dar. Es ist die abstrakte Form der Abfrage, ohne zu berücksichtigen, wie sie tatsächlich ausgeführt wird. Enthält Informationen zu den Vorgängen, die ausgeführt werden, wie Filterung, Auswahl, Zusammenführung, Aggregation und auch die falschen „Kleinigkeiten“, lol.

Physische Ebene:
Einzelheiten dazu, wie Spark die Abfrage tatsächlich ausführt. Dazu gehört die Reihenfolge der Operationen und welche Algorithmen verwendet werden (z. B. DBMS-Algorithmen). Es kann Details darüber enthalten, wie Daten partitioniert und unter den Ausführenden verteilt werden.

Ausführungsstrategien:
Die physische Ebene kann je nach Vorgang und Datengröße verschiedene Ausführungsstrategien anzeigen, die Spark verwenden kann, z. B. „Broadcast Join“ oder „Shuffle Hash Join“. Ich werde auch die wichtigsten Algorithmen des Ausführungsplans erklären, beruhigen Sie sich ...

Geschätzte Kosten:
Obwohl dies nicht immer angezeigt wird, können einige Pläne Kostenschätzungen für verschiedene Teile des Plans enthalten, damit Sie besser verstehen, welcher Teil der Bearbeitung möglicherweise am kostspieligsten ist.

Möglichkeiten zum Anzeigen des Apache Spark-Ausführungsplans

Wir haben das „Stamm“-Formular, das mit dem Befehl „explain()“ textuell wäre und eine Ausgabe ähnlich der folgenden mit einem einfachen Filter und einem Datenrahmen haben würde:

== Physischer Plan ==
*(2) Filter (Wert > 1)
- *(2) Projekt [Name#0, Wert#1]
- *(1) Vorhandenes RDD scannen[Name#0, Wert#1]

Und objektiv können wir es über die Schnittstelle analysieren, über die Spark-Benutzeroberfläche, in Databricks können wir darauf zugreifen, sei es in Zellenausführungen, im Job oder im Cluster. In Apache Spark ist es direkt die IP auf dem Standardport 4040.

Spark-Benutzeroberfläche ist in mehrere nützliche Abschnitte unterteilt:

• Jobs: Zeigt eine Liste aller in der Anwendung ausgeführten Jobs an. Jeder Job entspricht einer Aktion in Ihrem Code.

• Stufen: Zeigt die Stufen an, aus denen jeder Auftrag besteht. Phasen sind Unterteilungen von Arbeiten, die parallel ausgeführt werden können.

• Aufgaben: Einzelheiten zu den einzelnen Aufgaben in jeder Phase, einschließlich Informationen zur Ausführungszeit und zum Status der Aufgabe.

• Speicher: Bietet Informationen zur Speicher- und Speichernutzung von RDDs (Resilient Distributed Datasets).

• Umgebung: Zeigt Eigenschaften der Laufzeitumgebung an, einschließlich Spark-Konfigurationen und Systemvariablen.

• Executors: Zeigt Informationen zu den für die Anwendung erstellten Executors an, einschließlich Speichernutzung, Festplattennutzung und Leistungsstatistiken.

Hier war ich hierarchisch, okay? Dies ist die Reihenfolge, in der die Dinge funktionieren.

Ich möchte, dass Bilder auf dem Bildschirm angezeigt werden!!

Spark-Algorithmen und wie man die Tuning-Sünder erkennt:

Zuerst erkläre ich die Hauptalgorithmen, die sowohl in der Spark-UI-Schnittstelle als auch im Ausführungsplan demonstriert werden, sei es der logische oder der physische Plan:

HINWEIS: Denken Sie daran, dass Datensätze hier mit einer Spark-Tabelle identisch sind ;)

1. Beginnen wir mit dem berühmtesten Scan:

• FileScan: Liest Daten aus Eingabedateien. Es kann für verschiedene Dateiformate wie Parquet, ORC, CSV, JSON und andere optimiert werden.

2. Beitreten (Dieses hier gibt einige B.O.):

• Broadcast Hash Join: Wird verwendet, wenn einer der Datensätze klein genug ist, um an alle Knoten im Cluster übertragen zu werden, wodurch Shuffle vermieden wird (ich werde mehr darüber erklären, aber kurz gesagt handelt es sich um einen Daten-Shuffle-Vorgang). endgültige Verbindung).

• Shuffle Hash Join: Beide Datensätze (Tabellen, wenn Sie es vorziehen) werden gemischt, sodass sich die entsprechenden Schlüssel in derselben Partition befinden. Es wird verwendet, wenn die Datensätze groß sind und nicht an andere Knoten übertragen werden können.

• Zusammenführung sortieren: Erfordert, dass beide Datensätze vor dem Zusammenführen sortiert werden. Es ist effizient für große Datensätze, die bereits partitioniert und geordnet sind, d. h. die Verknüpfung erfolgt über partitionierte und auch geordnete Spalten (z. B. df.write.partitionBy("coluna1").sortBy("coluna2").parquet(" Pfad /zum/Speichern/partitioniert")

3. Aggregation (Summe, Anzahl, Gruppierung nach usw.):

• HashAggregate: verwendet eine Hash-Tabelle, um Daten zu aggregieren. Es ist effizient für große Datensätze, die in den Speicher passen.

• SortAggregate. Aggregiert Daten nach dem Sortieren. Es wird verwendet, wenn die Daten nicht in den Speicher passen.

4. Shuffle (Pass auf diesen Kerl auf):

• Shuffle: Verteilt Daten zwischen Partitionen für Vorgänge, die eine Neuorganisation erfordern, wie z. B. Verknüpfungen und Aggregationen. Es ist ein teurer Vorgang im Hinblick auf E/A und Netzwerk.

5. Umtausch:

• Ändert die Verteilung der Daten zwischen Partitionen. Es kann verwendet werden, um die Arbeitslast auf Clusterknoten zu verteilen. (eine Strategie, um das Gleichgewicht zu halten und dem Durcheinander zu entkommen)

6. Projekt:

• Wählt eine Teilmenge von Spalten aus einem DataFrame oder RDD aus.

7. Filter:

• Wendet Bedingungen an, um Datenzeilen zu filtern.

8. Sortieren:

• Bestellt Daten basierend auf einer oder mehreren Spalten.

Alle oben genannten Algorithmen können, wie ich bereits sagte, mit dem Befehl „explain()“ beobachtet werden.

Reale Shuffle-Problemszenarien:

1. Beitreten und GroupBy-Operationen
Vorgänge wie „join()“ und „groupByKey()“ lösen häufig Shuffle aus, wodurch Daten zwischen Partitionen neu verteilt werden. Dies kann Folgendes zur Folge haben:
Hohe Festplatten-E/A-Nutzung: Shuffle generiert viele Zwischendateien, die die lokale Festplatte der Ausführenden überlasten können.
Hohe Netzwerklast: Die zwischen Executoren übertragene Datenmenge kann beträchtlich sein, abhängig von der Anzahl der erforderlichen Verbindungen (Anzahl der Mapper multipliziert mit der Anzahl der Reduzierer)

• Identifikation: Überprüfen Sie in der Spark-Benutzeroberfläche auf der Registerkarte „Stufe“ die Werte „Shuffle Read Size/Records“ und „Shuffle Spill (Disk)“. Ein hohes Volumen in diesen Messwerten weist auf ein potenzielles Problem hin.

1. Partitionsungleichgewicht (Datenversatz) Wenn Daten ungleichmäßig auf Partitionen verteilt sind, können einige Aufgaben viel länger dauern als andere, was zu einer Beeinträchtigung der Gesamtleistung führt. Die Identifizierung ist die gleiche: Gehen Sie zur Spark-Benutzeroberfläche, gehen Sie zu dem Job, der sich auf den Abschnitt bezieht, der Zeit in Anspruch nimmt (hier kommt ein bewährter Punkt, den ich weiter unten erwähnen werde) und überprüfen Sie die feststeckende Phase (sie läuft, läuft aber nicht). Fortschritt) und sehen Sie sich die Shuffle-Metriken an, im Allgemeinen ein hohes Volumen im Speicher und beginnendes Volumen auf der Festplatte beim Aktualisieren. Dies weist darauf hin, dass dieses Ungleichgewicht den Speicher erreicht hat und mit dem Schreiben auf die Festplatte begonnen hat und die Festplatte offensichtlich langsamer ist, dann sitzen Sie und weine, wenn du dieses Szenario zulässt.

Abhilfe

• Um Probleme im Zusammenhang mit der Zufallswiedergabe zu entschärfen: Reduzieren Sie Vorgänge, die ein Mischen verursachen: Minimieren Sie nach Möglichkeit die Verwenden Sie groupByKey() und bevorzugen Sie ReduceByKey(). Passen Sie die Anzahl an Partitionen: Verwenden Sie spark.sql.shuffle.partitions, um die Anzahl anzupassen Partitionen während Shuffle-Vorgängen. Verwenden Sie Techniken wie Broadcast Joins: Um große Mengen von zu verbinden Daten mit kleineren Sätzen, wodurch unnötiges Mischen vermieden wird.

Mischen Sie die Metriken in der Spark-Benutzeroberfläche:

Wie Shuffle funktioniert und warum es kostspielig ist:

Zum Schluss und vielleicht am wichtigsten: Gute Praktiken:

1. Die überwiegende Mehrheit arbeitet mit Notebooks, da Databricks, Jupyter Notebook und Google Colab sehr beliebt sind. Teilen Sie daher jede Abfrage oder Transformation in separate Zellen auf, damit Sie leichter erkennen können, welcher Teil das Leistungsproblem darstellt. Lässt man alles zusammen, gibt es mehrere Jobs und es ist schwierig zu wissen, um welche Stufe es sich handelt.

2. Verwenden Sie Merge statt Overwrite. Ich weiß, dass es mehr Arbeit macht, aber es ist logischer und performativer, da Merge weniger I/O verbraucht als ein erneutes „Dump“-Überschreiben der gesamten Tabelle im Datalake.

3. Verwenden Sie „cache()“ oder „persist()“, um Zwischendaten im Speicher zu speichern, insbesondere wenn diese über mehrere Vorgänge hinweg wiederverwendet werden. Dies kann die Neuberechnungszeit verkürzen und die Leistung verbessern.

4. Falls Sie es nicht wissen: Spark läuft auf einer JVM, ist also nativ Java. Wenn Sie die berühmte UDF – User Definition Function mit Python erstellen, hinterlassen Sie eine Art „Black Box“ für Spark und verhindern so automatische Optimierungen. Verwenden Sie nach Möglichkeit integrierte Spark SQL-Funktionen, die auf Leistung optimiert sind.

Nun, ich glaube, ich habe alles geschrieben, was ich mir vorgestellt habe. Ich schreibe gerne Artikel, weil es mir hilft, mich an einige Szenarien zu erinnern. Ich habe vor, ein Video aufzunehmen, das das alles zeigt, in der Praxis mit einigen öffentlichen Daten, ich werde es wahrscheinlich auf Kaggle bekommen, also folgen Sie mir auf LinkedIn, um über alles auf dem Laufenden zu bleiben, was mit der Welt der Daten, künstlicher Intelligenz und Softwareentwicklung zu tun hat

--> https://www.linkedin.com/in/airton-lira-junior-6b81a661

Folgen Sie mir auf LinkedIn, geben Sie mir einen Schub, ich freue mich über Feedback und bin auch völlig offen dafür, den Wissensaustausch zu verbessern.

Wenn Sie bis hierhin gelesen haben, herzlichen Glückwunsch!!! Ich hoffe, dass es alle Leistungsprobleme überwindet. Im nächsten Artikel werde ich auf die Vorteile von Databricks eingehen. Folgen Sie mir also auf LinkedIn, um es herauszufinden. Vielen Dank!!

Das obige ist der detaillierte Inhalt vonOptimierungsstrategien für Apache Spark verstehen und anwenden. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!