2021-05-14

Effizientes SQL auf Pandas mit DuckDB

Mark Raasveldt, Hannes Mühleisen

Kürzlich erschien ein Artikel, der für SQL in der Datenanalyse wirbt. Im DuckDB-Team sind wir große Fans von SQL. Es ist eine vielseitige Sprache, mit der sich viele Datenumwandlungen effizient ausführen lassen, ohne dass man sich um die physische Darstellung der Daten oder um den optimalen Ausführungsweg kümmern muss.

Aggregationen und Transformationen lassen sich sehr gut in einem externen Datenbanksystem wie Postgres ausführen, wenn die Daten dort liegen. Irgendwann müssen die Daten aber wieder nach Pandas und NumPy. Diese Bibliotheken sind der Standard für den Datenaustausch im großen Ökosystem der Data-Science-Bibliotheken in Python¹, etwa scikit-learn oder TensorFlow.

¹Apache Arrow gewinnt in diesem Bereich ebenfalls stark an Bedeutung, und DuckDB spricht auch Arrow.

Wenn Sie aus einer Datei lesen (z. B. CSV oder Parquet), landen die Daten oft gar nicht erst in einem externen Datenbanksystem, sondern direkt in einem Pandas-DataFrame.

SQL auf Pandas

Nachdem die Daten in einem Pandas-DataFrame liegen, sind oft noch weitere Aufbereitung und Analyse nötig. SQL ist ein sehr mächtiges Werkzeug für solche Transformationen. Mit DuckDB lassen sich SQL-Abfragen effizient direkt auf Pandas-DataFrames ausführen.

Als kurzer Vorgeschmack: Dieses Snippet führt beliebige SQL-Abfragen direkt auf Pandas-DataFrames mit DuckDB aus.

# to install: pip install duckdb
import pandas as pd
import duckdb

mydf = pd.DataFrame({'a' : [1, 2, 3]})
print(duckdb.query("SELECT sum(a) FROM mydf").to_df())

Im Rest des Artikels gehen wir genauer darauf ein, wie das funktioniert und wie schnell es ist.

Datenintegration und SQL auf Pandas

Eines der Kernziele von DuckDB ist, dass der Zugriff auf gängige Formate einfach sein soll. DuckDB kann Abfragen parallel direkt auf einem Pandas-DataFrame ausführen (oder auf einer Parquet-/CSV-Datei oder einer Arrow-Tabelle …). Ein eigener (zeitaufwendiger) Importschritt ist nicht nötig.

DuckDB kann Abfrageergebnisse auch direkt in jedes dieser Formate schreiben. Sie können DuckDB nutzen, um einen Pandas-DataFrame parallel per SQL zu verarbeiten und das Ergebnis wieder in einen Pandas-DataFrame zu wandeln, um es in anderen Data-Science-Bibliotheken weiterzuverwenden.

Führt man eine SQL-Abfrage aus, sucht DuckDB nach Python-Variablen, deren Name zu den Tabellennamen in der Abfrage passt, und liest automatisch die Pandas-DataFrames. Im vorherigen Beispiel sieht man das in Aktion:

import pandas as pd
import duckdb

mydf = pd.DataFrame({'a' : [1, 2, 3]})
print(duckdb.query("SELECT sum(a) FROM mydf").to_df())

Der SQL-Tabellenname mydf wird als lokale Python-Variable mydf interpretiert, die zufällig ein Pandas-DataFrame ist, den DuckDB direkt lesen und abfragen kann. Spaltennamen und Typen werden automatisch aus dem DataFrame übernommen.

Das ist nicht nur bequem, sondern auch sehr effizient. Für viele Abfragen können Sie DuckDB nutzen, um Daten schneller als Pandas und mit deutlich geringerem Speicherverbrauch zu verarbeiten, ohne das binäre Pandas-DataFrame-Format zu verlassen („Pandas-in, Pandas-out“). Anders als bei einem externen Datenbanksystem wie Postgres ist die Übertragungszeit von Eingabe und Ausgabe vernachlässigbar (siehe Anhang A).

Leistung von SQL auf Pandas

Um die Leistung von DuckDB bei SQL auf Pandas-DataFrames zu zeigen, stellen wir einige Benchmarks vor. Der Quellcode ist interaktiv in Google Colab verfügbar. In diesen Benchmarks arbeiten wir ausschließlich auf Pandas-DataFrames. Sowohl der DuckDB- als auch der Pandas-Code läuft vollständig nach dem Prinzip „Pandas-in, Pandas-out“.

Benchmark-Aufbau und Datensatz

Wir führen den Benchmark vollständig in der Google-Colab-Umgebung aus. Als Datensatz nutzen wir den bekannten TPC-H-Datensatz. Konkret konzentrieren wir uns auf die Tabellen lineitem und orders, die größten Tabellen im Benchmark. Der gesamte Datensatz ist etwa 1 GB im unkomprimierten CSV-Format („Scale Factor“ 1).

Da DuckDB mehrere Prozessoren nutzen kann (Multithreading), nehmen wir eine einthreadige Variante und eine Variante mit zwei Threads auf. DuckDB skaliert weit über zwei Threads hinaus, Google Colab unterstützt jedoch nur zwei.

Setup

Zuerst müssen wir DuckDB installieren. Das ist eine einfache Zeile.

pip install duckdb

Für die Verarbeitung laden wir zwei Parquet-Dateien mit wget. Anschließend laden wir die Daten mit dem eingebauten Parquet-Reader von DuckDB in einen Pandas-DataFrame. Das System erkennt automatisch, dass wir eine Parquet-Datei lesen, anhand der Endung .parquet.

lineitem = duckdb.query(
    "SELECT * FROM 'lineitemsf1.snappy.parquet'"
).to_df()
orders = duckdb.query(
    "SELECT * FROM 'orders.parquet'"
).to_df()

Ungroupierte Aggregationen

Für unsere erste Abfrage führen wir eine Reihe ungroupierter Aggregationen über den Pandas-DataFrame aus. Hier die SQL-Abfrage:

SELECT
    sum(l_extendedprice),
    min(l_extendedprice),
    max(l_extendedprice),
    avg(l_extendedprice)
FROM lineitem;

Der Pandas-Code sieht ähnlich aus:

lineitem.agg(
  Sum=('l_extendedprice', 'sum'),
  Min=('l_extendedprice', 'min'),
  Max=('l_extendedprice', 'max'),
  Avg=('l_extendedprice', 'mean')
)

Dieser Benchmark ist eine sehr einfache Abfrage, und Pandas schneidet hier gut ab. Bei solchen einfachen Abfragen glänzt Pandas, weil es direkt die Numpy-Routinen für diese Aggregationen aufrufen kann, die sehr effizient sind. Dennoch liegt DuckDB im einthreadigen Fall auf Augenhöhe mit Pandas und profitiert bei Bedarf von Multithreading.

Gruppierte Aggregation

Für die zweite Abfrage führen wir dieselben Aggregationen aus, diesmal mit einer Gruppierungsbedingung. In SQL fügen wir dafür eine GROUP-BY-Klausel hinzu.

SELECT
    l_returnflag,
    l_linestatus,
    sum(l_extendedprice),
    min(l_extendedprice),
    max(l_extendedprice),
    avg(l_extendedprice)
FROM lineitem
GROUP BY
    l_returnflag,
    l_linestatus;

In Pandas nutzen wir vor der Aggregation die Funktion groupby.

lineitem.groupby(
  ['l_returnflag', 'l_linestatus']
).agg(
  Sum=('l_extendedprice', 'sum'),
  Min=('l_extendedprice', 'min'),
  Max=('l_extendedprice', 'max'),
  Avg=('l_extendedprice', 'mean')
)

Diese Abfrage ist schon komplexer, und obwohl Pandas anständig arbeitet, ist es etwa doppelt so langsam wie die einthreadige DuckDB-Variante. DuckDB hat eine stark optimierte Aggregate-Hash-Tabelle, die Gruppierung und Berechnung aller Aggregationen in einem Durchlauf über die Daten erledigt.

Gruppierte Aggregation mit Filter

Angenommen, wir wollen nicht über alle Daten aggregieren, sondern nur über eine Teilmenge. Das geht mit einem Filter, der uninteressante Tupel entfernt. In SQL erledigen wir das über die WHERE-Klausel.

SELECT
    l_returnflag,
    l_linestatus,
    sum(l_extendedprice),
    min(l_extendedprice),
    max(l_extendedprice),
    avg(l_extendedprice)
FROM lineitem
WHERE
    l_shipdate <= DATE '1998-09-02'
GROUP BY
    l_returnflag,
    l_linestatus;

In Pandas können wir eine gefilterte Variante des DataFrame mit Auswahlklammern erzeugen.

# filter out the rows
filtered_df = lineitem[
  lineitem['l_shipdate'] < "1998-09-02"]
# perform the aggregate
result = filtered_df.groupby(
  ['l_returnflag', 'l_linestatus']
).agg(
  Sum=('l_extendedprice', 'sum'),
  Min=('l_extendedprice', 'min'),
  Max=('l_extendedprice', 'max'),
  Avg=('l_extendedprice', 'mean')
)

In DuckDB kombiniert der Query-Optimizer Filter und Aggregation zu einem Durchlauf über die Daten und liest nur relevante Spalten. In Pandas haben wir dieses Glück nicht. Der Filter reduziert die gesamte lineitem-Tabelle, einschließlich aller Spalten, die wir nicht nutzen. Dadurch ist die Filteroperation deutlich teurer als nötig.

Wir können diese Optimierung („Projection Pushdown“ in der Datenbankliteratur) manuell nachbauen. Dazu wählen wir zuerst nur die für die Abfrage relevanten Spalten und filtern dann den DataFrame. Das ergibt folgendes Snippet:

# projection pushdown
pushed_down_df = lineitem[
  ['l_shipdate',
   'l_returnflag',
   'l_linestatus',
   'l_extendedprice']
]
# perform the filter
filtered_df = pushed_down_df[
  pushed_down_df['l_shipdate'] < "1998-09-02"]
# perform the aggregate
result = filtered_df.groupby(
  ['l_returnflag', 'l_linestatus']
).agg(
  Sum=('l_extendedprice', 'sum'),
  Min=('l_extendedprice', 'min'),
  Max=('l_extendedprice', 'max'),
  Avg=('l_extendedprice', 'mean')
)

Der manuelle Projection Pushdown beschleunigt die Abfrage in Pandas deutlich, aber die gefilterte Aggregation bleibt teuer. Für einen Filter schreibt Pandas eine Kopie des gesamten DataFrame (minus der ausgefilterten Zeilen) zurück in den Speicher. Das kostet Zeit, wenn der Filter nicht sehr selektiv ist.

Dank holistischem Query-Optimizer und effizienter Query-Engine schneidet DuckDB bei dieser Abfrage deutlich besser ab.

Joins

Für die letzte Abfrage joinen wir (merge in Pandas) die Tabelle lineitem mit orders und filtern auf den interessierenden Auftragsstatus. Das führt zu folgender SQL-Abfrage:

SELECT
    l_returnflag,
    l_linestatus,
    sum(l_extendedprice),
    min(l_extendedprice),
    max(l_extendedprice),
    avg(l_extendedprice)
FROM lineitem
JOIN orders ON (l_orderkey = o_orderkey)
WHERE l_shipdate <= DATE '1998-09-02'
  AND o_orderstatus='O'
GROUP BY
    l_returnflag,
    l_linestatus;

Für Pandas müssen wir einen merge-Schritt ergänzen. Im einfachen Ansatz mergen wir lineitem und orders, wenden dann die Filter an und schließlich Gruppierung und Aggregation. Das ergibt folgendes Snippet:

# perform the join
merged = lineitem.merge(
  orders,
  left_on='l_orderkey',
  right_on='o_orderkey')
# filter out the rows
filtered_a = merged[
  merged['l_shipdate'] < "1998-09-02"]
filtered_b = filtered_a[
  filtered_a['o_orderstatus'] == "O"]
# perform the aggregate
result = filtered_b.groupby(
  ['l_returnflag', 'l_linestatus']
).agg(
  Sum=('l_extendedprice', 'sum'),
  Min=('l_extendedprice', 'min'),
  Max=('l_extendedprice', 'max'),
  Avg=('l_extendedprice', 'mean')
)

Hier verpassen wir zwei Leistungshebel:

• Erstens mergen wir viel zu viele Spalten, weil Spalten dabei sind, die für den Rest der Abfrage nicht nötig sind (Projection Pushdown).
• Zweitens mergen wir viel zu viele Zeilen. Wir können die Filter vor dem Merge anwenden, um die zu mergende Datenmenge zu reduzieren (Filter Pushdown).

Diese beiden Optimierungen manuell angewendet ergeben folgendes Snippet:

# projection & filter on lineitem table
lineitem_projected = lineitem[
  ['l_shipdate',
   'l_orderkey',
   'l_linestatus',
   'l_returnflag',
   'l_extendedprice']
]
lineitem_filtered = lineitem_projected[
  lineitem_projected['l_shipdate'] < "1998-09-02"]
# projection and filter on order table
orders_projected = orders[
  ['o_orderkey',
   'o_orderstatus']
]
orders_filtered = orders_projected[
  orders_projected['o_orderstatus'] == 'O']
# perform the join
merged = lineitem_filtered.merge(
  orders_filtered,
  left_on='l_orderkey',
  right_on='o_orderkey')
# perform the aggregate
result = merged.groupby(
  ['l_returnflag', 'l_linestatus']
).agg(
  Sum=('l_extendedprice', 'sum'),
  Min=('l_extendedprice', 'min'),
  Max=('l_extendedprice', 'max'),
  Avg=('l_extendedprice', 'mean')
)

Beide Optimierungen wendet DuckDBs Query-Optimizer automatisch an.

Der naive Ansatz ist im Vergleich zur optimierten Variante extrem zeitaufwendig. Das zeigt den Nutzen des automatischen Query-Optimizers. Selbst nach der Optimierung ist der Pandas-Code deutlich langsamer als DuckDB, weil Zwischenergebnisse nach den einzelnen Filtern und Joins im Speicher gehalten werden.

Fazit

Mit DuckDB nutzen Sie die ausdrucksstarke SQL-Sprache, ohne Daten in Pandas hinein- und wieder herausschieben zu müssen. DuckDB ist sehr einfach zu installieren und bietet viele Vorteile: Query-Optimizer, automatisches Multithreading und Verarbeitung größerer Datenmengen als der Hauptspeicher. DuckDB nutzt den Postgres-SQL-Parser und bietet viele der gleichen SQL-Funktionen wie Postgres, einschließlich fortgeschrittener Features wie Window-Funktionen, korrelierte Subqueries, (rekursive) Common Table Expressions, verschachtelte Typen und Sampling. Fehlt Ihnen etwas, öffnen Sie bitte ein Issue.

Anhang A: Hin und zurück: Daten von Pandas in eine SQL-Engine und zurück

Klassische SQL-Engines nutzen das Client-Server-Paradigma: Ein Client verbindet sich über einen Socket mit einem Server. Abfragen laufen auf dem Server, Ergebnisse kommen danach zum Client. Dasselbe gilt etwa für Postgres aus Python. Leider ist diese Übertragung ein ernstes Nadelöhr. In-Process-Engines wie SQLite oder DuckDB haben dieses Problem nicht.

Um zu zeigen, wie teuer die Übertragung über einen Socket ist, haben wir einen Benchmark mit Postgres, SQLite und DuckDB ausgeführt. Der Quellcode liegt auf GitHub.

In diesem Benchmark kopieren wir einen (recht kleinen) Pandas-DataFrame mit 10 Millionen 4-Byte-Integern (40 MB) von Python nach PostgreSQL, SQLite und DuckDB. Da das Standard-to_sql von Pandas eher langsam war, haben wir eine eigene Optimierung ergänzt: Pandas schreibt den DataFrame in eine temporäre CSV-Datei, und PostgreSQL kopiert die Daten direkt aus dieser Datei in eine neu erzeugte Tabelle. Das funktioniert natürlich nur, wenn der Datenbankserver auf derselben Maschine wie Python läuft.

SQLite ist hier deutlich besser als Postgres, aber immer noch eher langsam. Das liegt daran, dass to_sql in Pandas eine große Zahl von INSERT INTO-Statements ausführt, wobei alle einzelnen Werte des Pandas-DataFrame in eine zeilenweise Darstellung von Python-Objekten umgewandelt und an das System übergeben werden. DuckDB liest dagegen direkt das zugrunde liegende Array aus Pandas, wodurch die Operation fast instantan ist.

Das Zurückholen von Abfrageergebnissen oder Tabellen aus dem SQL-System nach Pandas ist ein weiteres mögliches Nadelöhr. Das eingebaute read_sql_query ist extrem langsam, und selbst der optimiertere CSV-Weg braucht für diesen winzigen Datensatz mindestens eine Sekunde. DuckDB erledigt auch diese Umwandlung fast instantan.

Anhang B: Vergleich mit PandaSQL

Es gibt ein Paket namens PandaSQL, das ebenfalls SQL direkt auf Pandas ausführt. Es baut jedoch auf der to_sql- und from_sql-Infrastruktur auf, die wir in Anhang A als extrem langsam gesehen haben.

Zur Sicherheit haben wir die erste ungroupierte Aggregationsabfrage in PandaSQL gestoppt. Beim ersten Versuch auf dem Originaldatensatz kam es zu einem Out-of-Memory-Fehler, der unsere Colab-Sitzung beendet hat. Deshalb haben wir den Benchmark für PandaSQL mit einer Stichprobe von 10 % der Originalgröße (600.000 Zeilen) wiederholt. Hier die Ergebnisse:

PandaSQL (angetrieben von SQLite) ist bei diesem einfachen Benchmark etwa 1000× langsamer als Pandas oder DuckDB. Der Leistungsunterschied war so groß, dass wir die übrigen Benchmarks für PandaSQL nicht ausgeführt haben.

Anhang C: Abfrage direkt auf Parquet

In den Benchmarks oben haben wir die Parquet-Dateien vollständig nach Pandas gelesen. DuckDB kann Abfragen aber auch direkt auf Parquet-Dateien ausführen (parallel!). In diesem Anhang zeigen wir die Leistung im Vergleich zum vorherigen Laden nach Python.

Für den Benchmark führen wir zwei Abfragen aus: die einfachste (ungroupierte Aggregation) und die komplexeste (der letzte Join) und vergleichen die Kosten der direkten Ausführung auf der Parquet-Datei mit dem Laden nach Pandas über read_parquet.

Setup

In DuckDB können wir eine View über die Parquet-Datei mit folgender Abfrage anlegen. Damit können wir Abfragen über die Parquet-Datei ausführen, als wäre sie eine normale Tabelle. Um Projection Pushdown müssen wir uns nicht kümmern: Ein SELECT * reicht, DuckDBs Optimizer projiziert zur Laufzeit nur die benötigten Spalten.

CREATE VIEW lineitem_parquet AS
    SELECT * FROM 'lineitemsf1.snappy.parquet';
CREATE VIEW orders_parquet AS
    SELECT * FROM 'orders.parquet';

Ungroupierte Aggregation

Nach dem Anlegen dieser View können wir dieselben Abfragen wie zuvor ausführen, diesmal gegen die Tabelle lineitem_parquet.

SELECT sum(l_extendedprice), min(l_extendedprice), max(l_extendedprice), avg(l_extendedprice) FROM lineitem_parquet;

Für Pandas müssen wir zuerst read_parquet ausführen, um die Daten nach Pandas zu laden. Dazu nutzen wir den von Apache Arrow angetriebenen Parquet-Reader. Danach können wir die Abfrage wie zuvor ausführen.

lineitem_pandas_parquet = pd.read_parquet('lineitemsf1.snappy.parquet')
result = lineitem_pandas_parquet.agg(Sum=('l_extendedprice', 'sum'), Min=('l_extendedprice', 'min'), Max=('l_extendedprice', 'max'), Avg=('l_extendedprice', 'mean'))

Wieder das Problem: Pandas liest die Parquet-Datei vollständig. Um das zu umgehen, müssen wir Projection Pushdown erneut manuell ausführen, indem wir read_parquet die gewünschten Spalten mitgeben.

Der Optimizer in DuckDB erkennt das selbst anhand der ausgeführten Abfrage.

lineitem_pandas_parquet = pd.read_parquet('lineitemsf1.snappy.parquet', columns=['l_extendedprice'])
result = lineitem_pandas_parquet.agg(Sum=('l_extendedprice', 'sum'), Min=('l_extendedprice', 'min'), Max=('l_extendedprice', 'max'), Avg=('l_extendedprice', 'mean'))

Der Leistungsunterschied zwischen Pushdown und keinem Pushdown ist dramatisch. Mit Pushdown liegt Pandas in derselben Größenordnung wie DuckDB. Ohne Pushdown wird die gesamte Datei von der Platte geladen, einschließlich der anderen 15 Spalten, die für die Abfrage nicht nötig sind.

Joins

Nun zur letzten Abfrage aus dem Join-Abschnitt. Zur Erinnerung:

SELECT
    l_returnflag,
    l_linestatus,
    sum(l_extendedprice),
    min(l_extendedprice),
    max(l_extendedprice),
    avg(l_extendedprice)
FROM lineitem
JOIN orders ON (l_orderkey = o_orderkey)
WHERE l_shipdate <= DATE '1998-09-02'
  AND o_orderstatus='O'
GROUP BY
    l_returnflag,
    l_linestatus;

Für Pandas erzeugen wir wieder zwei Versionen: eine naive und eine manuell optimierte. Den genauen Code finden Sie in Google Colab.

Bei dieser komplexeren Abfrage verschwindet der kleine Leistungsunterschied zwischen Pandas-DataFrame und Parquet-Datei, und die DuckDB-Zeiten ähneln stark denen von zuvor. Das zusätzliche Lesen von Parquet macht manuelle Optimierungen am Pandas-Code noch nötiger, die bei SQL in DuckDB gar nicht erforderlich sind.