Apache Spark optimieren
Apache Spark ist ein leistungsstarkes Tool zur Datenverarbeitung, das im Vergleich zu Hadoops MapReduce-Algorithmen oder Single-Node-Verarbeitung eine um Größenordnungen bessere Ausführungsgeschwindigkeit ermöglicht. Dennoch kann Spark – sei es durch alte Gewohnheiten von Programmierern aus prozeduralen Systemen oder einfach aus Unwissenheit – selbst bei moderaten Datenmengen und einfachen Joins ins Straucheln geraten. Bevor Sie Ihrem Cluster weitere Nodes hinzufügen oder bestehende mit mehr RAM ausstatten (was nicht immer möglich ist), sollten Sie diese 5 Aspekte der Spark-API berücksichtigen, um das Framework optimal zu nutzen:
Data at Rest
Die Verwaltung der Speicherstrategie für Ihre Eingabe-/Ausgabe-/Temp-Daten ist entscheidend, nicht nur für die Optimierung Ihrer eigenen Anwendung, sondern auch für alle anderen Anwendungen, die dieselben Daten verwenden. Diese Strategie umfasst u. a. die Wahl des richtigen Speichersystems, der Dateiformate und der Partitionierung der Daten.
Als Entwickler haben Sie wahrscheinlich wenig Einfluss auf das Speichersystem, das Ihre Anwendung bedienen muss. Dennoch ist es wichtig zu wissen, wie ein bestimmter Speicher in Ihrem Szenario abschneidet – z. B. in Bezug auf Durchsatz oder Kosten für Datenänderungen.
Von Apache Spark unterstützte Dateiformate
Spark unterstützt viele Datenformate, sodass man leicht ein suboptimales Format wählt, nur weil man damit vertraut ist, und dabei die potenziellen Performance-Vorteile anderer, oft für verteiltes Computing entwickelter Formate übersieht.
Bevorzugen Sie immer binäre, spaltenbasierte Formate (z. B. Parquet) gegenüber Alternativen (z. B. CSV). Sie sind nicht nur schneller zu laden, sondern ersparen Ihnen auch die Behandlung von Sonderfällen (Escape-Zeichen, Zeilenumbrüche usw.).
Splittable file types
Es wird leicht übersehen, dass einige Dateiformate nicht einfach in Chunks aufgeteilt werden können, was die Nutzung der verteilten Natur von Spark verhindert. Dateien wie ZIP oder JSON müssen vollständig gelesen werden, bevor sie interpretiert werden können.
Stellen Sie sicher, dass die von Ihnen in Spark geladenen Dateien in diesem Sinne splittable sind.
Table Partitioning
Wenn Sie wissen, wie Ihre Tabellen am häufigsten von Endanwendern (und von Ihnen selbst) gelesen werden, können Sie die Rohdaten intelligent nach dem Schlüssel partitionieren, nach dem Sie abfragen. Partitionierung kann in bestimmten Szenarien die zu ladende Datenmenge für Ihren Spark-Cluster begrenzen und so die Ausführungsgeschwindigkeit drastisch erhöhen.
Data Skewness in Apache Spark
„Skewness“ ist ein statistischer Begriff, der die Asymmetrie einer Verteilung beschreibt. Im Big-Data-Kontext spricht man von „skewed data“, wenn Daten nicht gleichmäßig verteilt sind und daher schwer parallelisierbar sind. Besonders betroffen sind Join-Operationen.
Ein Beispiel ist eine Tabelle, in der ein bestimmter Schlüssel einer unverhältnismäßig großen Anzahl von Zeilen zugeordnet ist. Beim Join auf diesen Schlüssel sendet Spark alle Datensätze mit demselben Join-Key in dieselbe Partition. Der „beliebte“ Schlüssel wird also in einem Task verarbeitet, was oft zu erheblichem Shuffle-Overhead führt.
Sie können solche Situationen im Spark UI beobachten. Wenn ein Task deutlich länger dauert als andere und hohe Werte für Shuffle Read Size / Records aufweist, können Sie davon ausgehen, dass ein Node die Hauptarbeit leistet – meist wegen Data Skew. Wenn Sie wissen möchten, wie das Ihrem Unternehmen helfen kann, besuchen Sie unsere Data Engineering Consultancy Seite.
Möglichkeiten zur Abhilfe:
- Broadcast Join statt Sort Merge Join
Wenn Sie zwei Tabellen joinen, von denen eine deutlich kleiner ist (passt in den Executor-Speicher), können Sie Spark „hint“ geben, einen Broadcast Join statt eines Sort Merge Join zu verwenden. Das kann die Join-Zeiten bei skewed data drastisch verbessern. Mehr zu Join-Hints: https://spark.apache.org/docs/latest/sql-ref-syntax-qry-select-hints.html#join-hints
Beachten Sie, dass Spark Broadcast Joins automatisch plant, wenn eine Join-Seite kleiner als der Broadcast-Join-Threshold (spark.sql.autoBroadcastJoinThreshold) ist. - Key Salting
Beim Key Salting wird dem Schlüssel künstliche Variabilität hinzugefügt. Dies kann durch die Einführung von „Buckets“ in Ihrer Haupttabelle erreicht werden, indem Sie die rechte Join-Tabelle mit dem gleichen Bucket-Bereich „explodieren“ (Werte außer der Bucket-Nummer werden kopiert) und den Join über Schlüssel UND Bucket ausführen. So kann der größte Schlüsselwert auf mehrere Partitionen verteilt werden, was die Daten gleichmäßiger auf die Nodes verteilt.
Caching Data in Spark
Der Hauptvorteil von Spark gegenüber MapReduce in Hadoop ist die Nutzung von RAM für Berechnungen.
Spark verwendet jedoch Lazy Evaluation, d. h. Zwischenergebnisse werden erst berechnet, wenn sie wirklich benötigt werden. Das bedeutet, dass diese Zwischenergebnisse nicht im Speicher gehalten werden, da Spark nur das Endergebnis interessiert. Das funktioniert gut bei linearen Transformationen.
Wird ein Objekt jedoch mehrmals gelesen, wird es jedes Mal neu berechnet. Um das zu vermeiden, können Sie mit der cache()-Methode angeben, welche DataFrames im Speicher gehalten werden sollen.
Caching ist selbst eine „lazy operation“ und wird erst nach direktem Zugriff auf die Daten ausgeführt.
Repartitioning mit Apache Spark
Es ist wichtig, wie Ihre Manipulationstechniken die Daten und deren Verteilung auf die Nodes beeinflussen.
Wenn Sie z. B. einen sehr großen Datensatz filtern und einen viel kleineren zurückbehalten, können einige Partitionen auf Worker-Nodes leer bleiben. Das führt zu einer ungleichmäßigen Verteilung und verringert die Parallelität.
Repartition() und coalesce() sind Methoden, die ein Shuffle und eine gleichmäßige Verteilung der Daten erzwingen.
Repartition() erstellt die Partitionen komplett neu und führt ein vollständiges Shuffle durch. Coalesce() fasst Partitionen zusammen, was weniger Datenbewegung bedeutet, aber zu ungleichmäßigen Partitionen und Data Skew führen kann.
UDFs und Broadcast Variables
User-Defined Functions (UDFs) sollten aus Performance-Gründen vermieden werden – sie zwingen die Datenrepräsentation als Objekte in der JVM, was teuer ist. Die strukturierte API von Spark ist davon befreit und wird für Datenmanipulationen empfohlen.
Wenn Sie dennoch UDFs verwenden, nutzen Sie Broadcast Variables. Wenn Sie Daten haben, die in mehreren UDF-Aufrufen verwendet werden, sorgt Broadcasting dafür, dass auf jedem Node eine einzige, schreibgeschützte Kopie der Daten liegt, ohne dass sie bei jedem Aufruf neu übertragen werden müssen. Broadcasting ist auch außerhalb von UDFs nützlich, z. B. bei Lookup-Tabellen.
