Wie verwalte und optimiere ich Iceberg-Tabellen für eine effiziente Datenspeicherung und -abfrage?

Lösung

Typ der Iceberg-Tabelle für den Anwendungsfall auswählen

Die von dir gewählte Lösung hängt davon ab, welchen Typ von Iceberg-Tabelle du eingerichtet hast.

Copy-On-Write (CoW)-Tabelle für Lesevorgänge verwenden

Bei diesem Typ von Iceberg-Tabelle schreibt jeder Aktualisierungs- oder Löschvorgang, den du an den Datensätzen vornimmst, die entsprechenden Datendateien im Backend neu. Ein Neuschreiben verlangsamt die Leistung, insbesondere wenn es mehrere Aktualisierungs- und Löschvorgänge gibt. Verwende eine CoW-Tabelle, wenn die Anwendungsfälle mehr Lesevorgänge als Schreibvorgänge umfassen.

Führe den folgenden Befehl von Apache Spark SQL aus, um eine vorhandene Iceberg-Tabelle in eine CoW-Tabelle zu konvertieren:

spark.sql("ALTER TABLE <table-name>
SET TBLPROPERTIES ('write.delete.mode'='copy-on-write','write.update.mode'='copy-on-write')")

Um eine neue CoW-Tabelle zu erstellen, verwende die Tabelleneigenschaften 'write.delete.mode'='copy-on-write','write.update.mode'='copy-on-write':

dataFrame.createOrReplaceTempView("tmp_<your_table_name>")

query = f"""
CREATE TABLE glue_catalog.<your_database_name>.<your_table_name>
USING iceberg
TBLPROPERTIES
('format-version'='2','write.delete.mode'='copy-on-write','write.update.mode'='copy-on-write')
AS SELECT * FROM tmp_<your_table_name>
"""
spark.sql(query)

Merge-On-Read (MoR)-Tabelle für Schreibvorgänge verwenden

Wenn du Datensätze in einer MoR-Tabelle aktualisierst oder löschst, fügt die Aktion neue Datendateien hinzu. Die neu hinzugefügten Löschungs-Datendateien werden während eines Lesevorgangs zusammengeführt. Schreibvorgänge erfordern jedoch lediglich, dass du vorhandenen Dateien neue Dateien hinzufügst. Wenn du häufiger in die Tabelle schreibst als ihr Tabelle liest, wähle eine MoR-Tabelle aus.

Führe den folgenden Spark-SQL-Befehl aus, um eine MoR-Iceberg-Tabelle zu verwenden:

spark.sql("ALTER TABLE <table-name>
    SET TBLPROPERTIES ('write.delete.mode'='merge-on-read','write.update.mode'='merge-on-read')")

Hinweis: Um die Eigenschaften besser kontrollieren zu können, ist es eine bewährte Methode, die Iceberg-Tabellen aus einer Spark-Engine zu erstellen. Du kannst auch AWS Glue, EMR Spark oder Amazon Athena verwenden, um die Tabellen zu erstellen. Athena bietet jedoch nur eingeschränkte Unterstützung für Tabelleneigenschaften und verwendet nur den MoR-Tabellentyp.

Iceberg-Tabellen optimieren

Iceberg-Tabellen können aufgrund einer Zunahme der Anzahl von Metadatendateien, Löschungsdateien usw. einen Rückgang bei der Abfrageleistung aufweisen. Hier sind einige Methoden, die du verwenden kannst, um die Abfrageeffizienz und die Datenspeicherung zu optimieren.

Snapshots ablaufen lassen

Iceberg-Tabellen behalten Snapshots bei, damit du Daten aus einem älteren Zustand der Tabelle abrufen kannst. Diese Snapshots werden für jeden Schreibvorgang erstellt, der für die Tabelle ausgeführt wird, und die relevante Snapshot-ID wird der neuen Metadatendatei hinzugefügt. Im Laufe der Zeit erhöht die Anzahl der Snapshots die Größe der Metadatendatei. Diese zusätzlichen Snapshots verursachen eine verzögerte Abfrageleistung.

Verwende die folgenden Optionen, um Snapshots ablaufen zu lassen:

• Verwende den Vorgang expireSnapshots in Spark, um Snapshots, die älter als der angegebene Zeitstempel sind, für große Tabellen parallel ablaufen zu lassen:

      SparkActions.get()
      .expireSnapshots(table)
      .expireOlderThan(tsToExpire)
      .execute()

• Oder verwende eine Prozedur namens „expire_snapshots“. Weitere Informationen findest du auf der Iceberg-Website unter expire_snapshots.

   spark.sql("CALL glue_catalog.system.expire_snapshots('databasename.tablename',<timestamp value>)")

  Führe den vorhergehenden Code in regelmäßigen Abständen in einem AWS-Glue-Auftrag aus. Wenn du den Snapshot-Ablauf automatisierst, kannst du die Anzahl der Datendateien begrenzen, die Größe der Metadatendatei klein halten und eine effiziente Abfrageleistung gewährleisten.

Alte Metadatendateien entfernen

Lege die Tabelleneigenschaft write.metadata.delete-after-commit.enabled auf „True“ (Wahr) fest, um alte Metadatendateien nach jedem Tabellen-Commit automatisch zu löschen. Du kannst auch write.metadata.previous-versions-max festlegen, um die Anzahl der beizubehaltenden Metadatendateien zu verwalten.

Manifest-Dateien neu schreiben

Eine Iceberg-Tabelle verwendet Manifeste und Manifest-Dateien zur Nachverfolgung aller Datendateien. Im Laufe der Zeit verweist jeder Snapshot auf viele Manifest-Dateien. Diese Vorgänge verlangsamen die Abfragen. Weitere Informationen findest du auf der Iceberg-Website unter Manifest Lists und Manifests.

Verwende die Prozedur „rewrite manifests“, um Manifest-Dateien effizient zu verwalten. Weitere Informationen findest du auf der Iceberg-Website unter rewrite_manifests.

Führe die folgende Spark-SQL-Abfrage aus:

spark.sql("CALL glue_catalog.system.rewrite_manifests('databasename.tablename')")

Datendateien neu schreiben

Iceberg verwaltet und verfolgt alle Datendateien der Tabelle in einer Metadatendatei. Im Laufe der Zeit erhöhen viele angesammelte Datendateien die Größe der Metadatendatei. Unnötige oder offene Dateien in der Metadatendatei verringern die Leseeffizienz. Die Prozedur „rewrite_data_files“ in Spark hilft dabei, Daten parallel zu komprimieren und die Leseeffizienz zu erhöhen. Weitere Informationen findest du auf der Iceberg-Website unter rewrite_data_files.

Führe den folgenden Spark-SQL-Befehl aus:

spark.sql("CALL glue_catalog.system.rewrite_data_files(table=>'databasename.tablename')")

Verwende die Strategien BINPACK oder SORT, um die Datendateien für deinen Anwendungsfall neu zu schreiben. Weitere Informationen findest du auf der Iceberg-Website unter BINPACK und SORT.

BINPACK: Dies ist der kostengünstigste und schnellste Ansatz. Er fasst die kleineren Dateien zu größeren Dateien zusammen und verringert die Gesamtzahl der Ausgabedateien. Die Reihenfolge der Datensätze bleibt unangetastet und Daten werden nicht umsortiert. Dies ist die Standardoption.

CALL catalog.system.rewrite_data_files(
  table => 'test_table',
  strategy => 'binpack',
  options => map(
    'rewrite-job-order','bytes-asc',
    'target-file-size-bytes','<set-afile-size>',
    'max-file-group-size-bytes','<max-group-size>' -- 10GB
  )
)

SORT: Die SORT-Strategie sortiert die Daten und komprimiert die Dateien gleichzeitig. Diese Strategie ist nützlich, wenn du viele Aggregatfunktionen ausführst, die die benachbarten Datensätze vergleichen (zum Beispiel die Funktionen „min“ oder „max“).

CALL catalog_name.system.rewrite_data_files(
    table => 'databasename.tablename',
    strategy => 'sort',
    sort_order => 'id', --can be any column
    options => map('rewrite-all','true')
)

Verwaiste Dateien entfernen

Verwaiste Dateien werden in keiner der Metadatendateien referenziert. Weitere Informationen findest du auf der Iceberg-Website unter remove_orphan_files.

Um verwaiste Dateien zu entfernen, führe den Befehl „remove_orphan_files“ wie unten gezeigt aus:

spark.sql("CALL glue_catalog.system.remove_orphan_files(table=>'databasename.tablename')")

Hinweis: Es ist eine bewährte Methode, einen geplanten Auftrag zur Verwaltung von Wartungsaktivitäten auszuführen. Verwende einen einzigen AWS-Glue-Auftrag, um alle oben genannten Spark-SQL-Abfragen auszuführen.

Weitere Informationen

Using the Iceberg framework in AWS Glue (Verwenden des Iceberg-Frameworks in AWS Glue)

Was ist Apache Iceberg?