2024-06-10

Native Delta-Lake-Unterstützung in DuckDB

Sam Ansmink

In den letzten Monaten hat DuckLabs mit Databricks zusammengearbeitet, um First-Party-Unterstützung für Delta Lake in DuckDB über das neue Projekt delta-kernel-rs hinzuzufügen. In diesem Blogpost geben wir einen kurzen Überblick über Delta Lake, Delta Kernel und natürlich die neue DuckDB-Delta-Extension.

Wenn Sie mit Delta Lake und Delta Kernel schon innig vertraut sind oder nur wissen wollen, wie man tanzt, können Sie zu den saftigen Teilen springen, wie man DuckDB mit Delta nutzt.

Intro

Delta Lake ist ein Open-Source-Speicherframework zum Aufbau einer Lakehouse-Architektur. Um Delta Lake zu verstehen, müssen wir also verstehen, was die Lakehouse-Architektur ist. Das Lakehouse ist eine Datenmanagementarchitektur, die die Kosteneffizienz günstigen Object Storage mit einer smarten Managementschicht verbinden will. Einfach gesagt sind Lakehouse-Architekturen eine Sammlung von Dateien in verschiedenen Formaten mit einigen zusätzlichen Metadatenschichten darüber. Diese Metadatenschichten sollen Extrafunktionalität über der rohen Dateisammlung bieten, etwa ACID-Transaktionen, Time Travel, Partition- und Schema-Evolution, Statistiken und vieles mehr. Was eine Lakehouse-Architektur ermöglicht, ist, verschiedene datenintensive Anwendungen wie Datenanalytik und Machine-Learning-Anwendungen direkt auf einer großen Sammlung strukturierter, semistrukturierter und unstrukturierter Daten laufen zu lassen, ohne einen zwischengeschalteten Data-Warehousing-Schritt. Für den Deep Dive empfehlen wir das CIDR-2021-Paper „Lakehouse: A New Generation of Open Platforms that Unify Data Warehousing and Advanced Analytics“ von Michael Armbrust et al. Wenn Sie (verständlicherweise) zögern, in dichte wissenschaftliche Literatur einzutauchen, fasst dieses Bild es ziemlich gut zusammen:

Delta Lake

Zooming jetzt etwas näher auf unseren Star des Abends, Delta Lake. Delta Lake (oder einfach „Delta“) ist derzeit eines der führenden Open-Source- Lakehouse-Formate, neben Apache Iceberg und Apache Hudi. Der einfachste Weg, ein Gefühl dafür zu bekommen, was eine Delta-Tabelle ist, ist, sich eine Delta-Tabelle als „Sammlung von Parquet-Dateien mit etwas Metadaten“ vorzustellen. Mit dieser leichten Vereinfachung im Hinterkopf erzeugen wir jetzt eine Delta-Tabelle und schauen uns die erzeugten Dateien an, um unser Verständnis zu verbessern. Dazu richten wir Python mit den Paketen duckdb, pandas und deltalake ein:

pip install duckdb pandas deltalake

Dann nutzen wir DuckDB, um einige Data Frames mit Testdaten zu erzeugen, und schreiben das mit dem Paket deltalake in eine Delta-Tabelle:

import duckdb
from deltalake import DeltaTable, write_deltalake
con = duckdb.connect()
df1 = con.query("SELECT i AS id, i % 2 AS part, 'value-' || i AS value FROM range(0, 5) tbl(i)").df()
df2 = con.query("SELECT i AS id, i % 2 AS part, 'value-' || i AS value FROM range(5, 10) tbl(i)").df()
write_deltalake(f"./my_delta_table", df1,  partition_by=["part"])
write_deltalake(f"./my_delta_table", df2,  partition_by=["part"], mode='append')

Nach diesem Skript haben wir eine einfache Delta-Tabelle mit 10 Zeilen erzeugt, aufgeteilt auf zwei Partitionen, die wir in zwei getrennten Schritten hinzugefügt haben. Um zu prüfen, dass alles nach Plan läuft, fragen wir die Tabelle mit DuckDB ab:

SELECT *
FROM delta_scan('./my_delta_table')
ORDER BY id;

Das sieht großartig aus! Alle erwarteten Daten sind da. Schauen wir uns jetzt mit tree an, welche Dateien tatsächlich erzeugt wurden:

tree ./my_delta_table`

my_delta_table
├── _delta_log
│   ├── 00000000000000000000.json
│   └── 00000000000000000001.json
├── part=0
│   ├── 0-f45132f6-2231-4dbd-aabb-1af29bf8724a-0.parquet
│   └── 1-76c82535-d1e7-4c2f-b700-669019d94a0a-0.parquet
└── part=1
    ├── 0-f45132f6-2231-4dbd-aabb-1af29bf8724a-0.parquet
    └── 1-76c82535-d1e7-4c2f-b700-669019d94a0a-0.parquet

Die tree-Ausgabe zeigt 2 verschiedene Dateitypen. Eine Delta-Tabelle kann verschiedene andere Dateitypen enthalten, aber diese bilden die Basis jeder Delta-Tabelle.

Erstens gibt es Datendateien im Parquet-Format. Die Datendateien enthalten alle in der Tabelle gespeicherten Daten. Das ist sehr ähnlich dazu, wie Daten gespeichert werden, wenn DuckDB partitionierte Parquet-Dateien schreibt.

Zweitens gibt es Delta-Dateien im JSON-Format. Die Delta-Dateien enthalten ein Log der Änderungen an der Tabelle. Durch Replay dieses Logs kann ein Reader eine gültige Sicht der Tabelle konstruieren. Um das zu veranschaulichen, werfen wir einen kleinen Blick in eine der ersten Delta-Log-Dateien:

cat my_delta_table/_delta_log/00000000000000000000.json

...
{ "add": {
    "path": "part=1/0-f45132f6-2231-4dbd-aabb-1af29bf8724a-0.parquet",
    "partitionValues": { "part": "1" }
  },
  ...
}
{ "add": {
    "path": "part=0/0-f45132f6-2231-4dbd-aabb-1af29bf8724a-0.parquet",
    "partitionValues": { "part": "0" },
  },
  ...
}
...

Wie wir sehen, enthält diese Log-Datei zwei add-Objekte, die das Hinzufügen von Daten zu den Partitionen 1 bzw. 0 beschreiben. Beachten Sie auch, dass die Partitionswerte selbst explizit in diesen Delta-Dateien gespeichert sind: Obwohl die Dateistruktur einem Hive-Style-Partitionierungsschema sehr ähnlich sieht, werden die Ordnernamen intern von Delta nicht genutzt. Stattdessen werden die Partitionswerte aus den Metadaten gelesen.

Mit diesem einfachen Beispiel haben wir die Grundlagen von Delta gezeigt. Für ein gründlicheres Verständnis der Interna verweisen wir auf die offizielle Delta-Spezifikation, die nach Maßstäben von Protokollspezifikationen ziemlich leicht zu lesen ist. Die offizielle Spezifikation beschreibt im Detail, wie Delta jedes Detail handhabt, von den hier beschriebenen Grundlagen bis zu komplexeren Dingen wie Checkpointing, Deletes, Schema-Evolution und vieles mehr.

Implementierung

Der Delta Kernel

Ein relativ komplexes Protokoll wie Delta zu unterstützen, erfordert erheblichen Entwicklungs- und Wartungsaufwand. Deshalb wäre beim Blick auf die Unterstützung eines solchen Protokolls in einer Engine die logische Wahl, nach einer fertigen Bibliothek zu suchen, die das übernimmt. Im Fall von Delta Lake könnten wir zum Beispiel die Bibliothek delta-rs wählen. Für die Implementierung einer nativen DuckDB-Delta-Extension ist das aber problematisch: würden wir die Bibliothek delta-rs für die DuckDB-Extension nutzen, liefe die gesamte Interaktion mit den Delta-Tabellen über die Bibliothek delta-rs. Aber erinnern Sie sich: Eine Delta-Tabelle ist effektiv „nur ein Haufen Parquet-Dateien mit etwas Metadaten“. Das würde bedeuten, dass beim Lesen einer Delta-Tabelle durch DuckDB die Datendateien vom delta-rs-Parquet-Reader mit dem delta-rs-Dateisystem gelesen würden. Das ist ärgerlich: DuckDB kommt bereits mit einem ausgezeichneten Parquet-Reader. Außerdem hat DuckDB bereits Unterstützung für eine Vielzahl von Dateisystemen mit eigenem Credential-Management-System. Eine Bibliothek wie delta-rs für DuckDBs Delta-Extension zu nutzen, würde tatsächlich eine Reihe von Problemen mit sich bringen:

• größere Extension-Binary
• inkonsistentes Nutzererlebnis zwischen delta_scan und read_parquet
• erhöhte Wartungslast

Um diese Probleme zu lösen, würden wir eine Bibliothek bevorzugen, die nur das Delta-Protokoll implementiert und DuckDB alles machen lässt, was es bereits kann.

Zum Glück existiert diese Bibliothek, und sie heißt Delta Kernel Project. Der Delta Kernel ist ein „Satz von Bibliotheken zum Bau von Delta-Connectors, die aus Delta-Tabellen lesen und in sie schreiben können, ohne die Details des Delta-Protokolls verstehen zu müssen“. Das geschieht über zwei relativ einfache API-Sätze, die eine Engine implementieren würde, wie im Bild unten gezeigt:

Mehr Details zum Projekt delta-kernel-rs finden Sie in diesem ausgezeichneten Blogpost, der tief in Interna und Designrationale eintaucht.

Obwohl die Bibliothek delta-kernel-rs noch experimentell ist, hat sie kürzlich ihre Version v0.1.0 gestartet und bietet bereits viel Funktionalität. Weil delta-kernel-rs eine C/C++-Foreign-Function-Interface exponiert, war die Integration in eine DuckDB-Extension sehr unkompliziert.

DuckDB-Delta-Extension delta_scan

Jetzt sind wir bereit, in die Details der Interna der DuckDB-Delta-Extension einzutauchen. Zuerst implementiert die Delta-Extension derzeit eine einzelne Tabellenfunktion: delta_scan. Es ist eine einfache, aber leistungsfähige Funktion, die eine Delta-Tabelle scannt.

Um zu verstehen, wie diese Funktion implementiert ist, müssen wir zuerst die vier beteiligten Hauptkomponenten festlegen:

Außerdem müssen wir verstehen, dass vier Haupt-APIs beteiligt sind:

Jetzt haben wir alle Verbindungen, knüpfen wir sie zusammen. Wenn ein Nutzer eine Abfrage mit der Tabellenfunktion delta_scan ausführt, ruft DuckDB über die API TableFunction in die Funktion delta_scan der Delta-Extension. Die Tabellenfunktion delta_scan ist allerdings tatsächlich nur eine exakte Kopie der regulären Funktion read_parquet. Um read_parquet in delta_scan zu verwandeln, ersetzt sie den regulären MultiFileReader des parquet_scan (der einfach eine Liste oder ein Glob von Dateien scannt) durch einen eigenen DeltaMultiFileReader, der eine Dateiliste auf Basis der Delta-Tabellen-Metadaten erzeugt. Immer wenn die Parquet-Extension irgendein IO braucht, ruft sie über die API FileSystem in DuckDB, um das I/O zu handhaben. Diese gesamte Interaktion ist im Diagramm unten festgehalten.

In diesem Diagramm sehen wir alle vier Komponenten, die an der Verarbeitung einer Abfrage mit der Tabellenfunktion delta_scan beteiligt sind. Die Pfeile stellen die Kommunikation dar, die zwischen den Komponenten über die vier APIs stattfindet. Beim Lesen einer Delta-Tabelle sehen wir, dass die Metadaten auf der rechten Seite über den Delta Kernel laufen. Auf der linken Seite sehen wir, wie die Parquet-Daten durch die Parquet-Extension fließen.

Es fehlen hier offensichtlich einige wichtige Details, etwa der Umgang mit Deletion Vectors und Column Mappings, aber wir haben jetzt das Grundkonzept der DuckDB- Delta-Extension behandelt. Außerdem haben wir gezeigt, wie die aktuelle Implementierung eine sehr natürliche logische Trennung erreicht, wobei Komponenteninterna durch die Verbindung über klar definierte APIs abstrahiert werden. Damit erreicht die Implementierung folgende Schlüsseleigenschaften:

1. Die Details des Delta-Protokolls bleiben für jede DuckDB-Komponente weitgehend undurchsichtig. Der einzige Kontaktpunkt mit den Interna des Delta-Protokolls ist die schmale FFI, die der Delta Kernel exponiert. Das wird vollständig von der Delta-Extension gehandhabt, deren einzige Aufgabe ist, das in native DuckDB-APIs zu übersetzen.

2. Vollständige Wiederverwendung der bestehenden Parquet-Scan-Logik von DuckDB, ohne Code-Wiederverwendung oder Compile-Time-Abhängigkeiten zwischen Extensions. Weil die gesamte Interaktion zwischen Delta- und Parquet-Extension über DuckDB-APIs durch die laufende DuckDB-Instanz läuft, interagieren die Extensions nur über die APIs TableFunction und MultiFileReader. Das bedeutet auch, dass zukünftige Optimierungen der Parquet-Extension automatisch in der Delta-Extension verfügbar sind.

3. Alles I/O läuft über DuckDBs FileSystem-API. Das bedeutet, dass alle Dateisysteme (Azure, S3 usw.), die DuckDB zur Verfügung stehen, zum Scannen verfügbar sind. Jedes DuckDB-Dateisystem, das Dateien lesen und listen kann, kann für Delta genutzt werden. Das ist auch in DuckDB-Wasm nützlich, wo eigene Dateisystemimplementierungen genutzt werden. Warnung, zwei kleine Anmerkungen müssen hier gemacht werden. Erstens lässt die DuckDB-Delta-Extension derzeit noch einen kleinen Teil des IO vom Delta Kernel über interne Dateisystembibliotheken handhaben; das liegt daran, dass die FFI die FileSystem-APIs noch nicht exponiert, das wird sich aber sehr bald ändern. Zweitens: Obwohl das architektonische Design der Delta-Extension mit DuckDB-Wasm im Hinterkopf gemacht wurde, ist die Wasm-Version der Extension noch nicht verfügbar.

Wie man Delta in DuckDB nutzt

Die Delta-Extension in DuckDB zu nutzen ist sehr einfach, denn sie wird als eine der Core-DuckDB-Extensions verteilt und ist für Autoloading verfügbar. Das bedeutet, Sie können DuckDB einfach starten (ab v0.10.3) und ausführen:

SELECT * FROM delta_scan('./my_delta_table');

DuckDB installiert und lädt die Delta-Extension automatisch. Dann fragt sie die lokale Delta-Tabelle ./my_delta_table ab.

Liegt Ihre Delta-Tabelle auf S3, wollen Sie wahrscheinlich einige S3-Credentials setzen. Liegen diese Credentials bereits an einer der Standardstellen, etwa einer Umgebungsvariable oder in der Datei ~/.aws/credentials? Führen Sie einfach aus:

CREATE SECRET delta_s1 (
    TYPE s3,
    PROVIDER credential_chain
)
SELECT * FROM delta_scan('s3://⟨some-bucket⟩/⟨path/to/a/delta/table⟩');

Merken Sie sich Ihre AWS-Tokens lieber auswendig und möchten sie eintippen? Dann so:

CREATE SECRET delta_s2 (
    TYPE s3,
    KEY_ID '⟨AKIAIOSFODNN7EXAMPLE⟩',
    SECRET '⟨wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY⟩',
    REGION '⟨eu-west-1⟩'
)
SELECT * FROM delta_scan('s3://⟨some-bucket⟩/⟨path/to/a/delta/table⟩');

Haben Sie mehrere Delta-Tabellen mit verschiedenen Credentials? Kein Problem, Sie können scoped Secrets nutzen:

CREATE SECRET delta_s3 (
    TYPE s3,
    KEY_ID '⟨AKIAIOSFODNN7EXAMPLE1⟩',
    SECRET '⟨wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY1⟩',
    REGION '⟨eu-west-1⟩',
    SCOPE 's3://⟨some-bucket-1⟩'
)
CREATE SECRET delta_s4 (
    TYPE s3,
    KEY_ID '⟨AKIAIOSFODNN7EXAMPLE2⟩',
    SECRET '⟨wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY2⟩',
    REGION '⟨us-west-1⟩',
    SCOPE 's3://⟨some-bucket-2⟩'
)
SELECT * FROM delta_scan('s3://⟨some-bucket-1⟩/⟨table1⟩');
SELECT * FROM delta_scan('s3://⟨some-bucket-2⟩/⟨table2⟩');

Ist Ihre Tabelle öffentlich, aber außerhalb der Standard-AWS-Region? Setzen Sie die Region mit einem leeren S3-Secret:

CREATE SECRET delta_s5 (
    TYPE s3,
    REGION '⟨eu-west-2⟩'
)
SELECT * FROM delta_scan('s3://⟨some-public-bucket⟩/⟨table1⟩');

Aktueller Stand der Delta-Extension

Derzeit gilt die Delta-Extension noch als experimentell. Das liegt zum Teil daran, dass die Delta-Extension selbst noch sehr neu ist, aber auch daran, dass das Projekt delta-kernel-rs, auf das sie sich stützt, noch experimentell ist. Dennoch werden zentrale Delta-Scan-Features bereits von der aktuellen Version der Delta-Extension unterstützt, etwa:

• Alle Datentypen
• Filter- und Projektions-Pushdown
• File Skipping auf Basis von Filter-Pushdown
• Deletion Vectors
• Partitionierte Tabellen
• Vollständig paralleles Scannen

Die Delta-Extension ist auf den Plattformen linux_amd64, linux_arm64, osx_amd64 und osx_arm64 verfügbar. Unterstützung für die restlichen Plattformen kommt bald. Außerdem arbeiten wir weiter mit Databricks zusammen, um die Delta-Extension weiter zu verbessern und mehr Features hinzuzufügen, etwa:

• Schreibunterstützung
• Column Mapping
• Time Travel
• Variant, RowIds
• Wasm-Unterstützung

Details und Infos zu neu hinzugefügten Features finden Sie in den Delta-Extension-Docs und im Repository.

Fazit

In diesem Blogpost haben wir DuckDBs neue Delta-Extension vorgestellt, die die einfache Interaktion mit Delta Lake direkt aus der Bequemlichkeit Ihrer eigenen DuckDB-Umgebung ermöglicht. Dazu haben wir gezeigt, wie das Delta- Lake-Format aussieht, indem wir eine Delta-Tabelle erzeugt und sie mit DuckDB analysiert haben.

Wir wollen betonen, dass durch die Implementierung der Delta-Extension mit der Bibliothek delta-kernel-rs sowohl DuckDB als auch die Delta-Extension relativ einfach und weitgehend agnostisch gegenüber den Interna des Delta-Protokolls geblieben sind.

Wir hoffen, Sie geben der Delta-Extension eine Chance und freuen uns auf Feedback aus der Community! Wenn Sie am 2024 Databricks Data + AI Summit teilnehmen, schauen Sie sich unbedingt den Vortrag von DuckDB-Mitgründer Hannes Mühleisen am Donnerstag während der Keynote und die ausführliche Breakout-Session, ebenfalls am Donnerstag, für mehr Details zur DuckDB–Delta-Integration an.