2025-05-23

Arrow-IPC-Unterstützung in DuckDB

Pedro Holanda (DuckLabs), Ian Cook (Columnar), Dewey Dunnington (Wherobots), Bryce Mecum (Voltron Data)

DuckDB und Arrow

In den letzten Jahren hat das Apache-Arrow-Projekt in der Datenwelt viel Schwung aufgenommen, dank seines spaltenorientierten Formats, das den einfachen Austausch von Daten zwischen unterschiedlichen Systemen erlaubt – meist mit einem Zero-Copy-Ansatz. Apache Arrow treibt unter anderem die Integration zwischen DuckDB und Polars. In der Praxis nutzt DuckDB beim Erzeugen oder Konsumieren eines Polars-DataFrames darunter das Arrow-Spaltenformat.

Die Bedeutung eines solchen Formats ist auch einer der Hauptgründe, warum DuckDB zu den Pionieren bei der Integration mit Arrow gehörte und eine Arrow-Database-Connectivity-(ADBC)-Schnittstelle implementiert hat – besonders weil Arrow das ohne zusätzliche Abhängigkeiten möglich macht, dank seiner C-Data-Interface.

Eine Einschränkung von Arrows C-Data-Interface ist aber, dass Daten über Pointer (Speicheradressen) ausgetauscht werden. Das begrenzt die Möglichkeiten, wenn Sie Arrow-Daten zwischen unterschiedlichen Prozessen oder Systemen austauschen wollen. Um das zu überwinden, spezifiziert das Arrow-Projekt auch das Arrow-IPC-Format, mit dem sich Arrow-Spaltendaten effizient serialisieren und zwischen Prozessen oder über ein Netzwerk weitergeben lassen. Diese Daten können als Stream konsumiert werden, direkt aus einem Speicherpuffer oder aus einer Datei.

Wir freuen uns, bekannt zu geben, dass DuckDB diese Arrow-Streams jetzt über die neue arrow-Community-Erweiterung konsumieren und erzeugen kann. In diesem Beitrag beschreiben wir das Arrow-IPC-Serialisierungsformat genauer, zeigen, wie man die neue arrow-Community-Erweiterung für DuckDB installiert, und geben eine Demo zur Nutzung.

Arrow Interprocess Communication (Arrow IPC)

Das Arrow-IPC-Format bietet eine Möglichkeit, Daten im Arrow-Format zu serialisieren (und optional zu komprimieren), sodass Sie Daten über ein Netzwerk übertragen oder auf der Platte speichern können, während sie im Arrow-Format bleiben – ohne den Overhead, sie in ein anderes Format zu konvertieren. Arrow IPC unterstützt LZ4- und Zstd-Kompression, und als Datei unterstützt es außerdem einen Datei-Footer, der das Holen und Verarbeiten beschleunigen kann, indem Teile der Daten übersprungen werden (ähnlich dem Ansatz von Parquet). Im Vergleich zu Parquet hat das Arrow-IPC-Format zwei Hauptvorteile:

Einfache Implementierung: Einen Low-Level-Arrow-IPC-Consumer/Producer zu schreiben ist weniger komplex als einen Parquet-Consumer/Producer, besonders wenn das System bereits mit dem Arrow-Format integriert.
Schnelleres Encoding und Decoding: Das Serialisieren und Deserialisieren von Arrow-Daten ist deutlich einfacher und schneller als bei Parquet. Das kann schnellere Verarbeitungszeiten bringen – besonders wenn Sie Daten streamen, die danach nicht auf die Platte müssen.

Arrow und Parquet sind komplementäre Formate. Parquets ausgeklügeltes Encoding- und Kompressionssystem liefert typischerweise deutlich kleinere Dateien, was Parquet zur besseren Wahl für Archivspeicher macht. Arrows Fähigkeit, Encoding- und Decoding-Overheads zu eliminieren, liefert typischerweise schnelleren und effizienteren Datenaustausch, was Arrow zur besseren Wahl für Query-Result-Transfer und ephemeres Caching macht.

Zur Veranschaulichung, wie einfach das Arrow-IPC-Format ist, betrachten Sie die folgende Abbildung. Im Arrow-IPC-Format wird eine Tabelle als Folge von Record Batches serialisiert (eine Sammlung von Datensätzen in spaltenorientiertem Layout), vorangestellt von ihrem gemeinsamen Schema:

An illustration of an Arrow IPC stream transmitting data from a table with three columns. The first record batch contains the values for the first three rows, the second record batch contains the values for the next three rows, and so on. Actual Arrow record batches might contain thousands to millions of rows.

Abbildung aus Apache Arrow Blog: How the Apache Arrow Format Accelerates Query Result Transfer.

Hinweis: In realistischen Szenarien sind Record Batches deutlich größer, und die Abbildung oben ist zur Veranschaulichung vereinfacht.

Die Arrow-Community-Erweiterung

DuckDB hat die Integration mit dem Arrow-IPC-Format schon seit vielen Jahren über die (inzwischen deprecated) Arrow-Kern-Erweiterung. Der Hauptzweck dieser Unterstützung war aber die Interoperabilität von DuckDB mit JavaScript, daher war sie nur darauf ausgelegt, serialisierte In-Memory-Buffer zu lesen, nicht Arrow-IPC-Dateien. Die Codekomplexität und Wartbarkeit der Erweiterung waren sehr hoch, weil die Arbeit mit Arrow IPC die gesamte Arrow-C++-Bibliothek als Abhängigkeit erforderte – wir wollten keinen eigenen Serialisierungs- und Deserialisierungscode für das Arrow-IPC-Format schreiben.

Inzwischen hat eine deutlich kleinere Arrow-C++-Implementierung als Weg zur Interaktion mit Arrow-IPC-Daten Popularität gewonnen: die nanoarrow-Bibliothek. Mit nanoarrow haben wir die alte DuckDB-Arrow-Erweiterung komplett neu gestaltet: deutlich kleinere Abhängigkeit, sauberere Codebase und die Fähigkeit, Arrow-IPC-Dateien zu scannen. Wir haben die Gelegenheit auch genutzt, die Arrow-DuckDB-Erweiterung von einer Kern-Erweiterung zu einer Community-Erweiterung zu machen. Das aus zwei Hauptgründen. Erstens, damit die Arrow-Entwickler- und Nutzer-Community stärker am Bau und Support der Erweiterung beteiligt sein kann. Zweitens, um einen Release-Kalender zu haben, der nicht an den DuckDB-Release-Kalender gebunden ist. In der Praxis heißt das: Mitglieder der Kern-Arrow-Entwickler-Community können entscheiden, wann eine neue Version der Erweiterung erscheint.

Installation und Laden der neuen Arrow-Erweiterung sind sehr einfach:

INSTALL arrow FROM community;
LOAD arrow;

Demo

In dieser Demo nutzen wir die neue Arrow-DuckDB-Erweiterung, um die TPC-H-Tabelle lineitem mit Scale Factor 10 als Arrow-IPC-Datei zu erzeugen. Unsere Demo konzentriert sich auf Arrow-IPC-Daten in einer Datei, die Erweiterung selbst erlaubt aber auch, das Arrow-IPC-Format direkt als Buffer zu konsumieren und zu erzeugen. Detaillierte Beispiele zur Nutzung und akzeptierte Parameter stehen im README der Erweiterung.

Wir beginnen mit dem Laden der arrow-Erweiterung und dem Erzeugen unserer TPC-H-Tabellen.

LOAD arrow;

CALL dbgen(sf = 10);

Um die Arrow-IPC-Dateien zu erzeugen, können wir einfach die Klausel COPY ... TO ... nutzen. Wir verwenden die empfohlene Dateiendung .arrows, weil diese Datei im Arrow-IPC-Stream-Format vorliegt.

COPY lineitem TO 'lineitem.arrows';

In dieser Demo haben wir der Einfachheit halber unsere Tabelle in eine einzelne Datei geschrieben. Unsere Arrow-COPY ... TO ...-Klausel erlaubt uns aber, chunk_size und die Zahl der row_groups pro Datei zu setzen. Diese Optionen erlauben Daten, die für die beste Performance in Ihrem Use Case optimiert sind. Eine kleinere chunk_size kann zum Beispiel die Gesamtperformance senken, aber Streaming-Szenarien nützen.

Wir können jetzt TPC-H-Query 6 direkt auf unserer Datei mit der Funktion read_arrow ausführen:

SELECT
    sum(l_extendedprice * l_discount) AS revenue
FROM
    read_arrow('lineitem.arrows')
WHERE
    l_shipdate >= CAST('1994-01-01' AS date)
    AND l_shipdate < CAST('1995-01-01' AS date)
    AND l_discount BETWEEN 0.05
    AND 0.07
    AND l_quantity < 24;

was ausgibt:

┌─────────────────┐
│     revenue     │
│  decimal(38,4)  │
├─────────────────┤
│ 1230113636.0101 │
│ (1.23 billion)  │
└─────────────────┘

Dank Replacement Scans können Sie die Funktion read_arrow weglassen, wenn der Dateiname auf .arrow oder .arrows endet. Zum Beispiel:

SELECT count(*) FROM 'lineitem.arrows';

was ausgibt:

┌─────────────────┐
│  count_star()   │
│      int64      │
├─────────────────┤
│    59986052     │
│ (59.99 million) │
└─────────────────┘

Der Einfachheit halber konzentrieren wir uns auf das Lesen einer einzelnen Datei, aber unser Reader unterstützt Multi-File-Reading, funktional auf Augenhöhe mit dem DuckDB-Parquet-Reader.

Was, wenn Sie einen Arrow-IPC-Stream direkt von einem Server in DuckDB holen wollen? Um das zu zeigen, können wir einen HTTP-Dateiserver in dem Verzeichnis starten, in dem wir lineitem.arrows gespeichert haben. Wir nutzen das Node.js-Paket serve dafür (statt Pythons eingebautem http.server), weil es HTTP-Range-Requests unterstützt:

npx serve -l 8008

Dann können Sie DuckDBs httpfs-Erweiterung nutzen, um die Arrow-Daten über das HTTP(S)-Protokoll abzufragen:

INSTALL httpfs;
LOAD httpfs;
LOAD arrow;

SELECT count(*) FROM read_arrow('http://localhost:8008/lineitem.arrows');

was dasselbe Ergebnis ausgibt:

┌─────────────────┐
│  count_star()   │
│      int64      │
├─────────────────┤
│    59986052     │
│ (59.99 million) │
└─────────────────┘

Alternativ können Sie ein Tool wie curl nutzen, um Arrow-IPC-Daten von einem Server zu holen und im Terminal an DuckDB zu pipen:

URL="http://localhost:8008/lineitem.arrows"
SQL="LOAD arrow; FROM read_arrow('/dev/stdin') SELECT count(*);"

curl -s "$URL" | duckdb -c "$SQL"

was dasselbe Ergebnis ausgibt. Weitere Demos der arrow-Erweiterung finden Sie in unserem arrow-ipc-demo-Repository.

Bonus: Coole Use Cases für Arrow IPC

Eine DuckDB-Abfrage gegen Daten im Arrow-IPC-Format wie in der Demo oben ist ein ziemlich netter Trick, und er funktioniert so gut, weil DuckDB und Arrow ein natürliches Paar sind – beide nutzen ein spaltenorientiertes Datenlayout. Sie fragen sich aber vielleicht, was Sie sonst noch mit Arrow-IPC-Daten tun können. Eines der Hauptziele von Arrow ist Interoperabilität, und indem wir unsere Daten im Arrow-IPC-Format speichern, haben wir viele Optionen für die Verbindung mit anderen Tools eröffnet.

Zum Beispiel können wir jetzt mit PyArrow mit unseren Daten arbeiten:

import pyarrow as pa

with open('lineitem.arrows', 'rb') as source:
   stream = pa.ipc.open_stream(source)
   tbl = stream.read_all()

oder Polars:

import polars as pl

tbl = pl.read_ipc_stream("lineitem.arrows")

oder ClickHouse:

CREATE TABLE
    lineitem
ENGINE MergeTree()
ORDER BY tuple()
AS
    SELECT * FROM file('lineitem.arrows', 'ArrowStream');

oder jede der zahlreichen anderen Arrow-Bibliotheken (in einem Dutzend Sprachen) oder Arrow-kompatiblen Systeme.

Die Vorteile von Arrow IPC hören dort nicht auf: Arrow IPC eignet sich auch ideal für Larger-than-Memory-Use-Cases. Mit PyArrow können wir unsere Datei lineitem.arrows memory-mappen und damit arbeiten, ohne sie vollständig in den Speicher zu lesen:

import pyarrow as pa

with pa.memory_map('lineitem.arrows', 'rb') as source:
    stream = pa.ipc.open_stream(source)
    tbl = stream.read_all()

tbl.num_rows
# => 59986052

Dann können wir prüfen, dass PyArrow keine Buffer allokieren musste, um die Daten zu halten, weil sie alle auf der Platte leben:

pa.total_allocated_bytes()
# => 0

Jetzt können wir dieselbe Abfrage wie in der Demo oben ausführen und zeigen, dass wir dasselbe Ergebnis bekommen:

import datetime
import pyarrow.compute as pc

subset = tbl.filter(
    (pc.field("l_shipdate") >= datetime.datetime(1994, 1, 1)) &
    (pc.field("l_shipdate") < datetime.datetime(1995, 1, 1)) &
    (pc.field("l_discount") >= 0.05) &
    (pc.field("l_discount") <= 0.07) &
    (pc.field("l_quantity") < 24.)
)
pc.sum(pc.multiply(subset.column("l_extendedprice"), subset.column("l_discount")))
# => <pyarrow.Decimal128Scalar: Decimal('1230113636.0101')>

Und obwohl lineitem.arrows über 10 GB groß ist, musste PyArrow nur einen Bruchteil des Speichers allokieren:

pa.total_allocated_bytes()
# => 201594240 (192MB)

Fazit und Ausblick

In diesem Blogbeitrag haben wir die neue Arrow-Community-Erweiterung vorgestellt, mit der DuckDB-Nutzerinnen und -Nutzer mit Arrow-IPC-Streaming-Buffern und -Dateien interagieren können. Besonderer Dank an Voltron Data für die Ermöglichung dieser Erweiterung durch die Zusammenarbeit mit DuckLabs. Unten unsere Pläne für diese Erweiterung:

Unterstützung für ZSTD- und LZ4-Kompression beim Schreiben von Arrow IPC. DuckDB unterstützt derzeit nur das Schreiben unkomprimierter Buffer.
Unterstützung für LZ4-Kompression beim Lesen von Arrow-Buffern. Der Reader unterstützt derzeit nur ZSTD oder unkomprimierte Buffer.
Unterstützung für das Schreiben des Arrow-IPC-Dateiformats mit Datei-Footer und Nutzung des Footers, um Reads zu beschleunigen.
Implementierung von C-API-DuckDB-Funktionen zum Erzeugen und Konsumieren von Arrow-IPC-Daten.

Wenn Sie an einem dieser geplanten Features arbeiten oder andere Features vorschlagen möchten oder Bugs finden, loggen Sie sie gern in unserem Issue-Tracker. Happy hacking!