Apache Spark mit Apache Iceberg - ein Weg zur Steigerung der Leistung und Sicherheit Ihrer Datenpipeline

Die SQL-Sprache wurde 1970 erfunden und treibt Datenbanken seit Jahrzehnten an. Sie ermöglicht Ihnen nicht nur die Abfrage von Daten, sondern auch deren einfache Änderung auf Zeilenebene. Die Big Data-Evolution im Jahr 2006 änderte diese Sichtweise, indem sie die Unveränderlichkeit als Heilmittel für die Reaktionsfähigkeit von analytischen Abfragen propagierte. Auch wenn unveränderliche Daten in vielen Fällen sinnvoll sind, besteht immer noch der Bedarf an skalierbaren Datensätzen mit der Möglichkeit, die Zeilen zu ändern und die Transaktionen auszuführen. Aber können wir dies in einer Welt tun, die von Hadoop-basierten Ausführungsmaschinen dominiert wird? Nun, das ist Apache Iceberg.

Apache Iceberg ist ein offenes Tabellenformat für große Analysedatensätze. Apache Iceberg kann mit gängigen Big Data-Verarbeitungsmaschinen wie Apache Spark, Trino, PrestoDB, Flink und Hive verwendet werden .

Apache Iceberg ist quelloffen und seine vollständige Spezifikation ist für jedermann zugänglich, keine Überraschungen.

Einführung

Was können Sie mit Apache Iceberg erreichen und wie können Sie von dieser Technologie profitieren? Stellen Sie sich eine Situation vor, in der der Produzent dabei ist, die Daten zu speichern und der Konsument die Daten mitten in diesem Prozess liest. Apache Iceberg bietet Ihnen serialisierbare Isolierung (Atomicity), Änderungen an der Tabelle sind atomar und außerdem können die Konsumenten keine partiellen oder unbestätigten Ergebnisse sehen. Während des Speichervorgangs sind die Daten nie gesperrt. Die Verbraucher können die Daten zuverlässig lesen, ohne die Sperre zu halten.

Wann immer Ihre Daten beschädigt zu sein scheinen oder für eine neue Version fehlen, können Sie sie mit Hilfe des Versionsverlaufs einfach auf die vorherige Version zurücksetzen .

Apache Iceberg bietet Ihnen die Möglichkeit, gleichzeitig in eine bestimmte Tabelle zu schreiben. Dabei geht es von einem optimistischen Gleichzeitigkeitsmechanismus aus, was bedeutet, dass jeder Writer, der einen Schreibvorgang durchführt, davon ausgeht, dass es zu diesem Zeitpunkt keinen anderen Writer gibt. Nachdem der Prozess abgeschlossen ist, versucht er, die Metadaten-Dateien auszutauschen. Wenn der Swap-Vorgang fehlschlägt, weil der andere Writer das Ergebnis bereits gespeichert hat, wird der Prozess auf der Grundlage des neuen aktuellen Tabellenstatus erneut versucht.

Die Partitionierung trägt dazu bei, die Menge der in den Speicher geladenen Daten zu reduzieren, während Apache Spark einen Partitionsbeschneidungsmechanismus verwendet, der nur ausgewählte Partitionen laden kann. Apache Iceberg ergänzt dieses Verhalten durch eine versteckte Partitionierung. Iceberg nimmt dann den Spaltenwert und kann ihn optional transformieren, behält aber dennoch die Beziehung im Auge. Nehmen wir zum Beispiel an, wir haben eine Tabelle mit Zeitstempelwerten wie unten dargestellt.

‌

Wir können Tabellen erstellen, die nach Datum partitioniert sind, und trotzdem die Beziehungen im Auge behalten und mit dem Partition Pruning-Mechanismus schnelle Abfragen durchführen.

spark.table(“table”).where(“time = ‘2018-07-18’”)

Um eine versteckte Partition zu definieren, schreiben Sie einfach

CREATE TABLE table (

    id bigint,

    time timestamp

)

USING iceberg

PARTITIONED BY (days(time))

Derzeit können Sie Funktionen wie z.B.:

Apache Iceberg verfolgt die Partitionen anhand von Metadaten-Dateien, auf deren Grundlage die Partitionierung während der Existenz der Tabelle weiterentwickelt werden kann.

Kunden müssen sich nicht mehr um die Schema-Evolution kümmern, denn Apache Iceberg übernimmt auch das, indem es Funktionen zur Schema-Evolution hinzufügt:

Verschiedene Tisch-Spezifikationen

Zurzeit unterstützt Apache Iceberg zwei Versionen der Tabellenspezifikation.

Version 1 der Iceberg-Spezifikation definiert, wie man große Tabellen mit unveränderlichen Datenformaten wie Parquet, Avro oder ORC verwalten kann.

Version 2 fügt Aktualisierungen und Löschungen auf Zeilenebene für Version 1 hinzu. Der Hauptunterschied zwischen den Versionen besteht darin, dass Version 2 Löschdateien hinzufügt, um Zeilen zu kodieren, die in bestehenden Datendateien gelöscht werden, um die Menge der neu geschriebenen Daten zu reduzieren.

Wie Apache Iceberg die Daten verwaltet (Tabelle v1)

Der Tisch ist in zwei Bereiche unterteilt:

Daten:

Metadaten

Daten bestehen aus Dateien mit tatsächlichen Daten, auch aus früheren Snapshots. Metadaten enthalten die Informationen in Snapshots und Dateien, die sich auf bestimmte Snapshots beziehen (Avro-Dateien).

Snap-Dateien enthalten Informationen über Avro-Dateien, in denen bestimmte Parkettdateien gefunden werden können. Avro-Dateien, die mit einer bestimmten uuid beginnen, enthalten Verweise auf bestimmte Datendateien für einen bestimmten Snapshot.

Die Dateien vnr.metadata.json enthalten Informationen über das Schema, den Zeitpunkt der letzten Aktualisierung (Anhängen, Überschreiben), die Snapshot-Version, die Partitionierung und einige einfache Statistiken. Der Versionshinweis ist wie ein Tipp in Git, er verweist auf die aktuelle Version.

Wie Sie Apache Iceberg mit Spark verwenden

Apache Iceberg bietet zwei Methoden für Spark-Benutzer, um mit ACID-Tabellen zu interagieren: über DataFrames oder mit einer SQL-Syntax. Alle Iceberg-Funktionen sind für Spark Version 3.x verfügbar, für Version 2.4 gibt es nur Unterstützung für DataFrame overwrite und append.

Apache Iceberg bietet eine einfache Möglichkeit zur Erweiterung von Spark mit Tabellenspezifikationen, dem Hinzufügen eines Jars zu einer Spark-Sitzung, dem Hinzufügen von Erweiterungen durch die Verwendung geeigneter SQL-Erweiterungen und der Angabe des Katalogsortes. .

Die Interaktion mit Tabellen ist ebenfalls vereinfacht. Um eine partitionierte Tabelle auf der Basis des Datenrahmens zu erstellen, schreiben Sie einfaches SQL

CREATE TABLE prod.db.table

USING iceberg

PARTITIONED BY (months(date))

AS SELECT * FROM temporary

ORDER BY date

Die explizite Sortierung ist aufgrund der Beschränkungen von Spark unerlässlich (Spark erlaubt es Iceberg nicht, eine Sortierung anzufordern, bevor eine Schreiboperation durchgeführt wird). Die Funktion Monate erstellt eine versteckte Partition.

Das gleiche Ziel kann mit der v2 API erreicht werden

df.sortWithinPartitions(“date”).writeTo(“prod.db.table”).create()

Darüber hinaus bietet Ihnen Apache Iceberg die Möglichkeit, Daten in die Tabelle einzufügen, auf der Grundlage eines Prädikats zusammenzuführen, zu überschreiben, zu löschen (Löschen auf Zeilenebene erfordert Spark-Erweiterungen), zu aktualisieren, Datenrahmen anzuhängen und zu überschreiben.

Um eine Tabelle zu verwalten, verwenden Sie die Syntax ALTER TABLE, die Ihnen die Möglichkeit bietet:

Apache Iceberg bietet Ihnen die Flexibilität, beliebige Snapshots oder Daten zu einem bestimmten Zeitpunkt zu laden. Um den Versionsverlauf zu überprüfen, führen Sie
spark.read.table("local.db.table.history").show(10, false)

‌

Um einen Snapshot nach ID zu lesen, führen Sie

spark.read.option("snapshot-id", 2626515310899177788L).table("local.db.table")

Oder zu einem beliebigen Zeitpunkt

spark.read.option("as-of-timestamp", 1637879675001L).table("local.db.table")

Warum lohnt es sich, integrierte Wartungsprogramme zu verwenden?

Apache Iceberg schreibt viele neue Dateien auf einen Datenträger. Bei vielen Versionen kann der belegte Speicherplatz auf der Festplatte drastisch ansteigen. Um dies zu vermeiden, können bestimmte Versionen des Snapshots als abgelaufen markiert und vom Datenträger entfernt werden.

In manchen Situationen kann es vorkommen, dass Produzenten während des Schreibens der Daten ausfallen und die Datendateien nicht mit den Metadaten verbunden sind. Solche Dateien verursachen zwar keine Probleme beim Laden von Daten, belegen aber definitiv Speicherplatz. Um dies zu vermeiden,entfernen Sieverwaiste Dateien einfach mit der Java- oder Scala-API.

Ohne die Tabelle für Verbraucher zu sperren, bietet Apache Iceberg die Möglichkeit, Dateien mit Hilfe eines Datenverdichtungsmechanismus zu größeren Dateien zu verdichten.

Wenn die Daten in der Tabelle eine lange Versionsgeschichte haben, ist es wichtig, alte Metadaten-Dateien zu entfernen, insbesondere bei Streaming-Aufträgen, die viele neue Metadaten-Dateien erzeugen können.

Zusammenfassung

Apache Iceberg ist eine vielversprechende Technologie, die von der Open Source Community entwickelt wurde. Es handelt sich um ein offenes Tabellenformat, das sowohl in Clustern vor Ort als auch in der Cloud verwendet werden kann. Um die Iceberg-Funktionalität zu Apache Spark hinzuzufügen, müssen Sie lediglich zusätzliche Pakete bereitstellen und ein paar Spark-Konfigurationsoptionen angeben (spark.sql.extensions, spark.sql.catalog.spark_catalog etc.). Rollback von Daten, wann immer es aufgrund der Datenhistorie erforderlich ist, Laden bestimmter Versionen, wenn Sie keinen Alptraum mehr haben, wenn die Daten nach einem Ausfall wiederhergestellt werden müssen.

Die Partitionierung in Apache Iceberg ist dynamisch, Metadaten-Dateien enthalten diese Informationen, sie kann sich während der Lebensdauer der Tabelle weiterentwickeln und wenn eine neue Partitionierungsebene benötigt wird, ist das kein Problem. Die Partitionierung ist ganz einfach - die versteckte Partitionierung ermöglicht das Bereinigen von Partitionen auf der Grundlage von Spaltenbeziehungen und nicht von strengen Werten. Erstellen Sie Datums-Partitionen, aber laden Sie sie auf der Basis von Datum, Monat oder Zeitstempel, da dies für den Benutzer transparent ist.