2021-12-03

DuckDB quakt Arrow: Eine Zero-Copy-Datenintegration zwischen Apache Arrow und DuckDB

Pedro Holanda, Jonathan Keane

Dieser Beitrag ist eine Zusammenarbeit mit und Cross-Post auf dem Arrow-Blog.

Ein Teil von Apache Arrow ist ein In-Memory-Datenformat, das für analytische Bibliotheken optimiert ist. Wie Pandas und R-Dataframes nutzt es ein spaltenorientiertes Datenmodell. Das Arrow-Projekt enthält aber mehr als nur das Format: Die Arrow-C++-Bibliothek, die in Python, R und Ruby über Bindings zugänglich ist, hat zusätzliche Features, mit denen Sie effizient auf Datensätzen rechnen können. Diese zusätzlichen Features liegen über der oben beschriebenen Implementierung des In-Memory-Formats. Die Datensätze können sich über mehrere Dateien in Parquet, CSV oder anderen Formaten erstrecken, und Dateien können sogar auf remote oder Cloud-Storage wie HDFS oder Amazon S3 liegen. Die Arrow-C++-Query-Engine unterstützt das Streamen von Query-Ergebnissen, hat eine effiziente Implementierung komplexer Datentypen (z. B. Lists, Structs, Maps) und kann wichtige Scan-Optimierungen wie Projection- und Filter-Pushdown durchführen.

DuckDB ist ein neues analytisches Datenmanagementsystem, das dafür gebaut ist, komplexe SQL-Abfragen innerhalb anderer Prozesse auszuführen. DuckDB hat Bindings für R und Python, unter anderem. DuckDB kann Arrow-Datensätze direkt abfragen und Query-Ergebnisse zurück nach Arrow streamen. Diese Integration erlaubt es Nutzern, Arrow-Daten mit DuckDBs SQL-Interface und API abzufragen und dabei DuckDBs parallele vektorisierte Ausführungsengine zu nutzen – ohne extra Datenkopien. Zusätzlich nutzt diese Integration Arrows Predicate- und Filter-Pushdown beim Scannen von Datensätzen voll aus.

Diese Integration ist einzigartig, weil sie Zero-Copy-Streaming von Daten zwischen DuckDB und Arrow und umgekehrt nutzt, sodass Sie eine Abfrage aus beiden zusammen komponieren können. Das ergibt drei Hauptvorteile:

Größer-als-Speicher-Analyse: Da beide Bibliotheken das Streamen von Query-Ergebnissen unterstützen, können wir auf Daten ausführen, ohne sie vollständig von der Platte zu laden. Stattdessen können wir einen Batch nach dem anderen ausführen. So können wir Abfragen auf Daten ausführen, die größer als der Speicher sind.
Komplexe Datentypen: DuckDB kann komplexe Datentypen, die in Arrow-Vektoren gespeichert werden können, effizient verarbeiten, einschließlich beliebig verschachtelter Structs, Lists und Maps.
Fortgeschrittener Optimizer: DuckDBs State-of-the-Art-Optimizer kann Filter und Projektionen direkt in Arrow-Scans pushen. Dadurch werden nur relevante Spalten und Partitionen gelesen, und das System kann z. B. Partition Elimination in Parquet-Dateien nutzen. Das beschleunigt die Query-Ausführung deutlich.

Wer nur an Benchmarks interessiert ist, kann direkt zum [Benchmark-Abschnitt unten](#Benchmark Comparison) springen.

Kurzer Rundgang

Bevor wir in die Details der Integration eintauchen, geben wir in diesem Abschnitt ein kurzes motivierendes Beispiel, wie mächtig und einfach die DuckDB-Arrow-Integration zu nutzen ist. Mit ein paar Zeilen Code können Sie schon Arrow-Datensätze abfragen. Sagen wir, Sie wollen den berüchtigten NYC-Taxi-Datensatz analysieren und herausfinden, ob Gruppen mehr oder weniger Trinkgeld geben als Einzelfahrer.

R

Sowohl Arrow als auch DuckDB unterstützen dplyr-Pipelines für Leute, die sich mit dplyr für ihre Datenanalyse wohler fühlen. Das Arrow-Paket enthält zwei Hilfsfunktionen, mit denen wir Daten zwischen Arrow und DuckDB hin- und herschieben können (to_duckdb() und to_arrow()). Das ist besonders nützlich in Fällen, in denen etwas in Arrow oder DuckDB unterstützt wird, im anderen aber nicht. Finden Sie zum Beispiel eine komplexe dplyr-Pipeline, deren SQL-Übersetzung mit DuckDB nicht funktioniert, nutzen Sie to_arrow() vor der Pipeline, um die Arrow-Engine zu verwenden. Oder, wenn Sie eine Funktion haben (z. B. windowed Aggregates), die in Arrow noch nicht implementiert ist, nutzen Sie to_duckdb(), um die DuckDB-Engine zu verwenden. Und das, ohne Kosten für das (Re-)Serialisieren der Daten zu zahlen, wenn Sie sie hin- und herschieben!

library(duckdb)
library(arrow)
library(dplyr)

# Open dataset using year, month folder partition
ds <- arrow::open_dataset("nyc-taxi", partitioning = c("year", "month"))

ds %>%
  # Look only at 2015 on, where the number of passenger is positive, the trip distance is
  # greater than a quarter mile, and where the fare amount is positive
  filter(year > 2014 & passenger_count > 0 & trip_distance > 0.25 & fare_amount > 0) %>%
  # Pass off to DuckDB
  to_duckdb() %>%
  group_by(passenger_count) %>%
  mutate(tip_pct = tip_amount / fare_amount) %>%
  summarise(
    fare_amount = mean(fare_amount, na.rm = TRUE),
    tip_amount = mean(tip_amount, na.rm = TRUE),
    tip_pct = mean(tip_pct, na.rm = TRUE)
  ) %>%
  arrange(passenger_count) %>%
  collect()

Python

Der Workflow in Python ist so einfach wie in R. In diesem Beispiel nutzen wir DuckDBs Relational API.

import duckdb
import pyarrow as pa
import pyarrow.dataset as ds

# Open dataset using year, month folder partition
nyc = ds.dataset('nyc-taxi/', partitioning=["year", "month"])

# We transform the nyc dataset into a DuckDB relation
nyc = duckdb.arrow(nyc)

# Run same query again
nyc.filter("year > 2014 & passenger_count > 0 & trip_distance > 0.25 & fare_amount > 0")
    .aggregate("SELECT avg(fare_amount), avg(tip_amount), avg(tip_amount / fare_amount) AS tip_pct", "passenger_count").arrow()

DuckDB und Arrow: Die Grundlagen

In diesem Abschnitt schauen wir uns ein paar grundlegende Beispiele für den Code an, der nötig ist, um Arrow-Tabellen in Python und R zu lesen und auszugeben.

Setup

Zuerst müssen wir DuckDB und Arrow installieren. Der Installationsprozess für beide Bibliotheken ist unten gezeigt.

Python:

pip install duckdb
pip install pyarrow

install.packages("duckdb")
install.packages("arrow")

Um die Beispielbeispiele in diesem Abschnitt auszuführen, müssen wir die folgenden eigenen Parquet-Dateien herunterladen:

Python

In Python gibt es zwei Wege, Daten aus Arrow abzufragen.

Über die Relational API:

# Reads Parquet File to an Arrow Table
arrow_table = pq.read_table('integers.parquet')

# Transforms Arrow Table -> DuckDB Relation
rel_from_arrow = duckdb.arrow(arrow_table)

# we can run a SQL query on this and print the result
print(rel_from_arrow.query('arrow_table', 'SELECT sum(data) FROM arrow_table WHERE data > 50').fetchone())

# Transforms DuckDB Relation -> Arrow Table
arrow_table_from_duckdb = rel_from_arrow.arrow()

Indem man Replacement Scans nutzt und das Objekt direkt mit SQL abfragt:

# Reads Parquet File to an Arrow Table
arrow_table = pq.read_table('integers.parquet')

# Gets Database Connection
con = duckdb.connect()

# we can run a SQL query on this and print the result
print(con.execute('SELECT sum(data) FROM arrow_table WHERE data > 50').fetchone())

# Transforms Query Result from DuckDB to Arrow Table
# We can directly read the arrow object through DuckDB's replacement scans.
con.execute("SELECT * FROM arrow_table").fetch_arrow_table()

Es ist möglich, sowohl DuckDB-Relations als auch Query-Ergebnisse zurück nach Arrow zu transformieren.

R

In R können Sie mit Arrow-Daten in DuckDB interagieren, indem Sie die Tabelle als View registrieren (eine Alternative ist, dplyr zu nutzen, wie oben gezeigt).

library(duckdb)
library(arrow)
library(dplyr)

# Reads Parquet File to an Arrow Table
arrow_table <- arrow::read_parquet("integers.parquet", as_data_frame = FALSE)

# Gets Database Connection
con <- dbConnect(duckdb::duckdb())

# Registers arrow table as a DuckDB view
arrow::to_duckdb(arrow_table, table_name = "arrow_table", con = con)

# we can run a SQL query on this and print the result
print(dbGetQuery(con, "SELECT sum(data) FROM arrow_table WHERE data > 50"))

# Transforms Query Result from DuckDB to Arrow Table
result <- dbSendQuery(con, "SELECT * FROM arrow_table")

Daten von/nach Arrow streamen

Im vorherigen Abschnitt haben wir gezeigt, wie man mit Arrow-Tabellen interagiert. Arrow erlaubt Nutzern aber auch, mit den Daten streaming zu interagieren. Entweder sie konsumieren (z. B. aus einem Arrow Dataset) oder produzieren (z. B. einen RecordBatchReader zurückgeben). Und natürlich kann DuckDB Datasets konsumieren und RecordBatchReader produzieren. Dieses Beispiel nutzt den NYC-Taxi-Datensatz, gespeichert in nach Jahr und Monat partitionierten Parquet-Dateien, den wir über das Arrow-R-Paket herunterladen können:

arrow::copy_files("s3://ursa-labs-taxi-data", "nyc-taxi")

Python

# Reads dataset partitioning it in year/month folder
nyc_dataset = ds.dataset('nyc-taxi/', partitioning=["year", "month"])

# Gets Database Connection
con = duckdb.connect()

query = con.execute("SELECT * FROM nyc_dataset")
# DuckDB's queries can now produce a Record Batch Reader
record_batch_reader = query.fetch_record_batch()
# Which means we can stream the whole query per batch.
# This retrieves the first batch
chunk = record_batch_reader.read_next_batch()

R

# Reads dataset partitioning it in year/month folder
nyc_dataset = open_dataset("nyc-taxi/", partitioning = c("year", "month"))

# Gets Database Connection
con <- dbConnect(duckdb::duckdb())

# We can use the same function as before to register our arrow dataset
duckdb::duckdb_register_arrow(con, "nyc", nyc_dataset)

res <- dbSendQuery(con, "SELECT * FROM nyc", arrow = TRUE)
# DuckDB's queries can now produce a Record Batch Reader
record_batch_reader <- duckdb::duckdb_fetch_record_batch(res)

# Which means we can stream the whole query per batch.
# This retrieves the first batch
cur_batch <- record_batch_reader$read_next_batch()

Der vorangehende R-Code zeigt auf niedriger Ebene, wie die Daten gestreamt werden. Wir stellen die Hilfsfunktion to_arrow() im Arrow-Paket bereit, die ein Wrapper darum ist und es einfach macht, dieses Streaming in eine dplyr-Pipeline einzubauen.

In Arrow 6.0.0 gibt to_arrow() derzeit die volle Tabelle zurück, erlaubt aber volles Streaming in unserem kommenden 7.0.0-Release.

Benchmark-Vergleich

Hier zeigen wir in einem einfachen Benchmark den Performance-Unterschied zwischen dem Abfragen von Arrow-Datensätzen mit DuckDB und dem Abfragen von Arrow-Datensätzen mit Pandas. Für den Projection- und Filter-Pushdown-Vergleich nutzen wir Arrow-Tabellen. Das liegt daran, dass Pandas nicht in der Lage ist, Arrow-Stream-Objekte zu konsumieren.

Für die NYC-Taxi-Benchmarks haben wir einen Server im SciLens-Cluster genutzt und für die TPC-H-Benchmarks ein MacBook Pro mit einer M1-CPU. In beiden Fällen wurde Parallelität in DuckDB genutzt (die jetzt standardmäßig an ist).

Für den Vergleich mit Pandas gilt: DuckDB läuft parallel, während Pandas nur Single-Thread-Ausführung unterstützt. Außerdem sollte man beachten, dass wir automatische Optimierungen vergleichen. DuckDBs Query-Optimizer kann Filter und Projektionen automatisch pushen. Diese automatische Optimierung wird in Pandas nicht unterstützt, aber Nutzer können einige dieser Predicate- und Filter-Pushdowns manuell durchführen, indem sie sie im read_parquet()-Aufruf angeben.

Projection Pushdown

In diesem Beispiel führen wir eine einfache Aggregation über zwei Spalten unserer lineitem-Tabelle aus.

# DuckDB
lineitem = pq.read_table('lineitemsf1.snappy.parquet')
con = duckdb.connect()

# Transforms Query Result from DuckDB to Arrow Table
con.execute("""SELECT sum(l_extendedprice * l_discount) AS revenue
                FROM
                lineitem;""").fetch_arrow_table()

# Pandas
arrow_table = pq.read_table('lineitemsf1.snappy.parquet')

# Converts an Arrow table to a Dataframe
df = arrow_table.to_pandas()

# Runs aggregation
res =  pd.DataFrame({'sum': [(df.l_extendedprice * df.l_discount).sum()]})

# Creates an Arrow Table from a Dataframe
new_table = pa.Table.from_pandas(res)

Name	Zeit (s)
DuckDB	0.19
Pandas	2.13

Die lineitem-Tabelle besteht aus 16 Spalten, für die Ausführung dieser Abfrage sind aber nur zwei Spalten nötig: l_extendedprice und l_discount. Weil DuckDB die Projektion dieser Spalten pushen kann, kann es diese Abfrage etwa eine Größenordnung schneller ausführen als Pandas.

Filter Pushdown

Für unseren Filter-Pushdown wiederholen wir dieselbe Aggregation wie im vorherigen Abschnitt, fügen aber Filter auf 4 weiteren Spalten hinzu.

# DuckDB
lineitem = pq.read_table('lineitemsf1.snappy.parquet')

# Get database connection
con = duckdb.connect()

# Transforms Query Result from DuckDB to Arrow Table
con.execute("""SELECT sum(l_extendedprice * l_discount) AS revenue
        FROM
            lineitem
        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; """).fetch_arrow_table()

# Pandas
arrow_table = pq.read_table('lineitemsf1.snappy.parquet')

df = arrow_table.to_pandas()
filtered_df = lineitem[
        (lineitem.l_shipdate >= "1994-01-01") &
        (lineitem.l_shipdate < "1995-01-01") &
        (lineitem.l_discount >= 0.05) &
        (lineitem.l_discount <= 0.07) &
        (lineitem.l_quantity < 24)]

res =  pd.DataFrame({'sum': [(filtered_df.l_extendedprice * filtered_df.l_discount).sum()]})
new_table = pa.Table.from_pandas(res)

Name	Zeit (s)
DuckDB	0.04
Pandas	2.29

Der Unterschied zwischen DuckDB und Pandas ist jetzt drastischer: zwei Größenordnungen schneller als Pandas. Wieder, weil sowohl Filter als auch Projektion nach Arrow gepusht werden, liest DuckDB weniger Daten als Pandas, das diese Optimierung nicht automatisch durchführen kann.

Streaming

Wie zuvor gezeigt, kann DuckDB Arrow-Daten streaming konsumieren und produzieren. In diesem Abschnitt führen wir einen einfachen Benchmark aus, um die Vorteile bei Geschwindigkeit und Speicherverbrauch im Vergleich zu voller Materialisierung und Pandas zu zeigen. Dieses Beispiel nutzt den vollen NYC-Taxi-Datensatz, den Sie herunterladen können.

# DuckDB
# Open dataset using year, month folder partition
nyc = ds.dataset('nyc-taxi/', partitioning=["year", "month"])

# Get database connection
con = duckdb.connect()

# Run query that selects part of the data
query = con.execute("SELECT total_amount, passenger_count, year FROM nyc where total_amount > 100 and year > 2014")

# Create Record Batch Reader from Query Result.
# "fetch_record_batch()" also accepts an extra parameter related to the desired produced chunk size.
record_batch_reader = query.fetch_record_batch()

# Retrieve all batch chunks
chunk = record_batch_reader.read_next_batch()
while len(chunk) > 0:
    chunk = record_batch_reader.read_next_batch()

# Pandas
# We must exclude one of the columns of the NYC dataset due to an unimplemented cast in Arrow.
working_columns = ["vendor_id","pickup_at","dropoff_at","passenger_count","trip_distance","pickup_longitude",
    "pickup_latitude","store_and_fwd_flag","dropoff_longitude","dropoff_latitude","payment_type",
    "fare_amount","extra","mta_tax","tip_amount","tolls_amount","total_amount","year", "month"]

# Open dataset using year, month folder partition
nyc_dataset = ds.dataset(dir, partitioning=["year", "month"])
# Generate a scanner to skip problematic column
dataset_scanner = nyc_dataset.scanner(columns=working_columns)

# Materialize dataset to an Arrow Table
nyc_table = dataset_scanner.to_table()

# Generate Dataframe from Arow Table
nyc_df = nyc_table.to_pandas()

# Apply Filter
filtered_df = nyc_df[
    (nyc_df.total_amount > 100) &
    (nyc_df.year >2014)]

# Apply Projection
res = filtered_df[["total_amount", "passenger_count","year"]]

# Transform Result back to an Arrow Table
new_table = pa.Table.from_pandas(res)

Name	Zeit (s)	Spitzen-Speicherverbrauch (GB)
DuckDB	0.05	0.3
Pandas	146.91	248

Der Zeitunterschied zwischen DuckDB und Pandas ist eine Kombination aller Integrationsvorteile, die wir in diesem Artikel erkundet haben. In DuckDB wird der Filter-Pushdown angewendet, um Partition Elimination durchzuführen (d. h. wir überspringen das Lesen der Parquet-Dateien, bei denen das Jahr <= 2014 ist). Der Filter-Pushdown wird auch genutzt, um unrelevante row_groups zu eliminieren (d. h. Row Groups, in denen der total amount immer <= 100 ist). Durch unseren Projection Pushdown muss Arrow nur die interessierenden Spalten aus den Parquet-Dateien lesen, sodass es nur 4 von 20 Spalten liest. Pandas dagegen kann keine dieser Optimierungen automatisch pushen, sodass der volle Datensatz gelesen werden muss. Das ergibt den Unterschied von 4 Größenordnungen in der Query-Ausführungszeit.

In der Tabelle oben zeigen wir auch den Vergleich des Spitzen-Speicherverbrauchs zwischen DuckDB (Streaming) und Pandas (voll materialisierend). In DuckDB müssen wir nur die interessierende Row Group in den Speicher laden. Deshalb ist unser Speicherverbrauch niedrig. Wir haben auch konstanten Speicherverbrauch, weil wir immer nur eine dieser Row Groups gleichzeitig im Speicher halten müssen. Pandas muss dagegen alle Parquet-Dateien vollständig materialisieren, wenn die Abfrage ausgeführt wird. Deshalb sehen wir einen konstanten steilen Anstieg des Speicherverbrauchs. Der gesamte Unterschied im Speicherverbrauch der beiden Lösungen liegt bei etwa 3 Größenordnungen.

Fazit und Feedback

In diesem Blogpost haben wir vor allem gezeigt, wie man Abfragen auf Arrow-Datensätzen mit DuckDB ausführt. Es gibt weitere Bibliotheken, die ebenfalls das Arrow-Format konsumieren können, sie haben aber andere Zwecke und Fähigkeiten.

Wenn Sie Probleme bei der Nutzung unserer Integration haben, öffnen Sie bitte ein Issue im Issue-Tracker von DuckDB oder im Issue-Tracker von Arrow, je nachdem, welche Bibliothek das Problem hat.