Optimierung der Schreibleistung

In diesem Abschnitt werden Tabelleneigenschaften beschrieben, die Sie anpassen können, um die Schreibleistung auf Iceberg-Tabellen unabhängig von der Engine zu optimieren.

Stellen Sie den Tabellenverteilungsmodus ein

Iceberg bietet mehrere Schreibverteilungsmodi, die definieren, wie Schreibdaten auf Spark-Aufgaben verteilt werden. Einen Überblick über die verfügbaren Modi finden Sie unter Verteilungsmodi schreiben in der Iceberg-Dokumentation.

Für Anwendungsfälle, in denen der Schreibgeschwindigkeit Priorität eingeräumt wird, insbesondere bei Streaming-Workloads, legen Sie die Einstellung auf fest. write.distribution-mode none Dadurch wird sichergestellt, dass Iceberg kein zusätzliches Spark-Shuffling anfordert und dass Daten geschrieben werden, sobald sie in Spark-Aufgaben verfügbar sind. Dieser Modus eignet sich besonders für Spark Structured Streaming-Anwendungen.

Anmerkung

Wenn Sie den Schreibverteilungsmodus auf einstellen, werden in der none Regel zahlreiche kleine Dateien erzeugt, was die Leseleistung beeinträchtigt. Wir empfehlen eine regelmäßige Komprimierung, um diese kleinen Dateien aus Gründen der Abfrageleistung in Dateien mit der richtigen Größe zu konsolidieren.

Wählen Sie die richtige Aktualisierungsstrategie

Verwenden Sie eine merge-on-read Strategie zur Optimierung der Schreibleistung, wenn langsamere Lesevorgänge mit den neuesten Daten für Ihren Anwendungsfall akzeptabel sind.

Wenn Sie Iceberg verwenden merge-on-read, schreibt Iceberg Aktualisierungen und Löschungen als separate kleine Dateien in den Speicher. Wenn die Tabelle gelesen wird, muss der Leser diese Änderungen mit den Basisdateien zusammenführen, um die neueste Ansicht der Daten zurückzugeben. Dies führt zu Leistungseinbußen bei Lesevorgängen, beschleunigt aber das Schreiben von Aktualisierungen und Löschungen. In der Regel merge-on-read ist es ideal für das Streaming von Workloads mit Updates oder Jobs mit wenigen Updates, die auf viele Tabellenpartitionen verteilt sind.

Sie können merge-on-read Konfigurationen (write.update.modewrite.delete.mode, undwrite.merge.mode) auf Tabellenebene oder unabhängig voneinander auf der Anwendungsseite festlegen.

Die Verwendung merge-on-read erfordert eine regelmäßige Komprimierung, um zu verhindern, dass sich die Leseleistung im Laufe der Zeit verschlechtert. Durch die Komprimierung werden Aktualisierungen und Löschungen mit vorhandenen Datendateien abgeglichen, um einen neuen Satz von Datendateien zu erstellen, wodurch die Leistungseinbußen beim Lesen vermieden werden. Standardmäßig werden bei der Komprimierung von Iceberg gelöschte Dateien nicht zusammengeführt, es sei denn, Sie ändern die Standardeinstellung der delete-file-threshold Eigenschaft auf einen kleineren Wert (siehe die Iceberg-Dokumentation). Weitere Informationen zur Komprimierung finden Sie im Abschnitt Eisberg-Komprimierung weiter unten in diesem Handbuch.

Wählen Sie das richtige Dateiformat

Iceberg unterstützt das Schreiben von Daten in den Formaten Parquet, ORC und Avro. Parquet ist das Standardformat. Parquet und ORC sind spaltenförmige Formate, die eine hervorragende Leseleistung bieten, aber im Allgemeinen langsamer zu schreiben sind. Dies stellt den typischen Kompromiss zwischen Lese- und Schreibleistung dar.

Wenn die Schreibgeschwindigkeit für Ihren Anwendungsfall wichtig ist, z. B. bei Streaming-Workloads, sollten Sie erwägen, im Avro-Format zu schreiben, indem Sie Avro in den write-format Writer-Optionen auf setzen. Da Avro ein zeilenbasiertes Format ist, bietet es schnellere Schreibzeiten auf Kosten einer langsameren Leseleistung.

Um die Leseleistung zu verbessern, sollten Sie regelmäßig komprimieren, um kleine Avro-Dateien zusammenzuführen und in größere Parquet-Dateien umzuwandeln. Das Ergebnis des Komprimierungsvorgangs wird durch die write.format.default Tabelleneinstellung bestimmt. Das Standardformat für Iceberg ist Parquet. Wenn Sie also in Avro schreiben und dann die Komprimierung ausführen, wandelt Iceberg die Avro-Dateien in Parquet-Dateien um. Ein Beispiel:

spark.sql(f"""
    CREATE TABLE IF NOT EXISTS glue_catalog.{DB_NAME}.{TABLE_NAME} (
        Col_1 float, 
        <<<…other columns…>>
        ts timestamp)
    USING iceberg
    PARTITIONED BY (days(ts))
    OPTIONS (
      'format-version'='2',
      write.format.default'=parquet)
""")

query = df \
    .writeStream \
    .format("iceberg") \
    .option("write-format", "avro") \
    .outputMode("append") \
    .trigger(processingTime='60 seconds') \
    .option("path", f"glue_catalog.{DB_NAME}.{TABLE_NAME}") \
    .option("checkpointLocation", f"s3://{BUCKET_NAME}/checkpoints/iceberg/")

    .start()