Die Notwendigkeit, das Schreiben auf Apache Spark zu optimieren

Artikel
07/16/2024

Analytische Workloads auf Big-Datenverarbeitungs-Engines wie Apache Spark werden am effizientesten ausgeführt, wenn standardisierte größere Dateigrößen verwendet werden. Das Verhältnis zwischen der Dateigröße, der Anzahl der Dateien, der Anzahl der Spark-Worker und ihrer Konfigurationen spielt eine entscheidende Rolle für die Leistung. Die Erfassung von Arbeitslasten in Datenseetabellen könnte das geerbte Merkmal haben, dass viele kleine Dateien ständig geschrieben werden; Dieses Szenario wird häufig als "kleines Dateiproblem" bezeichnet.

Optimize Write ist eine Delta Lake on Synapse-Funktion, die die Anzahl der geschriebenen Dateien reduziert und darauf abzielt, die individuelle Dateigröße der geschriebenen Daten zu erhöhen. Es optimiert Partitionen dynamisch, während Dateien mit einer Standardgröße von 128 MB generiert werden. Die Zieldateigröße kann pro Workloadanforderungen mithilfe von Konfigurationen geändert werden.

Diese Funktion erreicht die Dateigröße, indem eine zusätzliche Daten-Shuffle-Phase über Partitionen verwendet wird, was zusätzliche Verarbeitungskosten beim Schreiben der Daten verursacht. Die kleine Schreibstrafe sollte durch die Leseeffizienz auf den Tabellen aufgewogen werden.

Hinweis

Es ist in Synapse Pools für die höheren Apache Spark-Versionen als 3.1 verfügbar.

Vorteile von Optimize Writes

Es ist in Delta Lake-Tabellen sowohl für Batch- als auch für Streaming-Schreibmuster verfügbar.
Das Befehlsmuster muss nicht spark.write geändert werden. Die Funktion wird durch eine Konfigurationseinstellung oder eine Tabelleneigenschaft aktiviert.
Es reduziert die Anzahl der Schreibtransaktionen im Vergleich zum OPTIMIZE-Befehl.
OPTIMIZE-Vorgänge werden schneller, da weniger Dateien verarbeitet werden.
Der VACUUM-Befehl zum Löschen alter, nicht referenzierter Dateien wird ebenfalls schneller ausgeführt.
Abfragen scannen weniger Dateien mit optimaleren Dateigrößen, was entweder die Leseleistung oder die Ressourcennutzung verbessert.

Optimieren Sie Schreibnutzungsszenarien

Einsatzgebiete

Partitionierte Delta-Lake-Tabellen unterliegen dem Schreiben von Mustern, die suboptimal (weniger als 128 MB) oder nicht standardisierte Dateiengrößen (Dateien mit unterschiedlichen Größen zwischen sich selbst) generieren.
Neu partitionierte Datenrahmen, die mit suboptimaler Dateigröße auf die Festplatte geschrieben werden.
Delta-Lake-partitionierte Tabellen, auf die kleine Batch-SQL-Befehle wie UPDATE, DELETE, MERGE, CREATE TABLE AS SELECT, INSERT INTO usw. abzielen.
Streamingaufnahmeszenarien mit Anhängedatenmustern an partitionierte Delta-Lake-Tabellen, bei denen die zusätzliche Schreiblatenz tolerierbar ist.

Wann sollte man es vermeiden

Nicht partitionierte Tabellen.
Anwendungsfälle, in denen eine zusätzliche Schreiblatenz nicht akzeptabel ist.
Große Tabellen mit gut definierten Optimierungszeitplänen und Lesemustern.

So aktivieren und deaktivieren Sie die Funktion zum Optimieren des Schreibens

Die Funktion zum Optimieren des Schreibens ist standardmäßig deaktiviert. In Spark 3.3-Pool ist sie standardmäßig für partitionierte Tabellen aktiviert.

Sobald die Konfiguration für den Pool oder die Sitzung festgelegt ist, verwenden alle Spark-Schreibmuster die Funktionalität.

Um die Funktion zum Optimieren des Schreibens zu verwenden, aktivieren Sie sie mit der folgenden Konfiguration:

Scala und PySpark

spark.conf.set("spark.microsoft.delta.optimizeWrite.enabled", "true")

Spark SQL

SET `spark.microsoft.delta.optimizeWrite.enabled` = true

Um den aktuellen Konfigurationswert zu überprüfen, verwenden Sie den Befehl wie gezeigt:

Scala und PySpark

spark.conf.get("spark.microsoft.delta.optimizeWrite.enabled")

Spark SQL

SET `spark.microsoft.delta.optimizeWrite.enabled`

Um die Funktion zum Optimieren des Schreibens zu deaktivieren, ändern Sie die folgende Konfiguration wie gezeigt:

Scala und PySpark

spark.conf.set("spark.microsoft.delta.optimizeWrite.enabled", "false")

Spark SQL

SET `spark.microsoft.delta.optimizeWrite.enabled` = false

Steuern des optimierten Schreibens mithilfe von Tabelleneigenschaften

Auf neuen Tischen

CREATE TABLE <table_name> TBLPROPERTIES (delta.autoOptimize.optimizeWrite = true)

Scala

Verwenden der DeltaTableBuilder-API:

val table = DeltaTable.create()
  .tableName("<table_name>")
  .addColumnn("<colName>", <dataType>)
  .location("<table_location>")
  .property("delta.autoOptimize.optimizeWrite", "true") 
  .execute()

Auf bestehenden Tischen

ALTER TABLE <table_name> SET TBLPROPERTIES (delta.autoOptimize.optimizeWrite = true)

Scala

Verwenden der DeltaTableBuilder-API

val table = DeltaTable.replace()
  .tableName("<table_name>")
  .location("<table_location>")
  .property("delta.autoOptimize.optimizeWrite", "true") 
  .execute()

Abrufen und Ändern der aktuellen Konfiguration der maximalen Dateigröße für "Write optimieren"

Verwenden Sie die folgenden Befehle, um den aktuellen Konfigurationswert abzurufen. Der Standardwert ist 128 MB.

Scala und PySpark

spark.conf.get("spark.microsoft.delta.optimizeWrite.binSize")

SET `spark.microsoft.delta.optimizeWrite.binSize`

Um den Konfigurationswert zu ändern

Scala und PySpark

spark.conf.set("spark.microsoft.delta.optimizeWrite.binSize", "134217728")

SET `spark.microsoft.delta.optimizeWrite.binSize` = 134217728

Freigeben über