2024-10-16

CSV-Performance vorantreiben: DuckDB mit dem NYC-Taxi-Datensatz benchmarken

Pedro Holanda

Der NYC-Taxi-Datensatz ist eine Sammlung vieler Jahre Taxi-Fahrten in New York City. Er ist sehr einflussreich und wird für Datenbank-Benchmarks, maschinelles Lernen, Datenvisualisierung und mehr genutzt.

2022 hat der Datenanbieter beschlossen, den Datensatz als Serie von Parquet-Dateien statt CSV-Dateien zu verteilen. Performance-technisch ist das klug: Parquet-Dateien sind viel kleiner als CSV-Dateien, und ihr natives spaltenorientiertes Format erlaubt schnelle Ausführung direkt darauf. Der Wechsel erschwert aber, wie viele Systeme die Dateien nativ laden können.

Im Blogbeitrag „Billion Taxi Rides in Redshift“ wird ein neuer Datenbank-Benchmark vorgeschlagen, der die Performance von Aggregationen über den Taxi-Datensatz bewertet. Der Datensatz wird außerdem mit anderen Datensätzen zu Wetter, Taxi-Typen und Abhol-/Zielorten gejoint und denormalisiert. Er wird dann als mehrere komprimierte, gzippte CSV-Dateien gespeichert, jeweils mit 20 Millionen Zeilen.

Der Taxi-Datensatz als CSV-Dateien

Da DuckDB für die Performance seines CSV-Readers bekannt ist, wollten wir wissen, ob der Ladeprozess dieses Benchmarks uns neue Performance-Engpässe im CSV-Loader zeigen kann. Diese Neugier hat uns auf die Reise geführt, diese Datensätze zu erzeugen und ihre Performance in DuckDB zu analysieren. Laut einer aktuellen Studie zur AWS-Redshift-Flotte sind CSV-Dateien der am häufigsten genutzte externe Quelldatentyp in S3, und 99 % davon sind gzippt. Deshalb hat mich besonders interessiert, dass der vorgeschlagene Benchmark ebenfalls aufgeteilte gzippte Dateien nutzt.

In diesem Blogbeitrag führen wir Sie durch die Ausführung dieses Benchmarks in DuckDB und diskutieren einige Lektionen und Ideen für unseren CSV-Reader. Der Datensatz für diesen Benchmark ist öffentlich verfügbar. Der Datensatz ist partitioniert und als Sammlung von 65 gzippten CSV-Dateien verteilt, jeweils mit 20 Millionen Zeilen und bis zu 1,8 GB pro Datei. Der gesamte Datensatz umfasst 111 GB komprimiert und 518 GB unkomprimiert. Wir beschreiben außerdem, wie wir den Datensatz erzeugt haben, und heben die Unterschiede zum Original aus dem Beitrag „Billion Taxi Rides in Redshift“ hervor.

Den Benchmark reproduzieren

Faires Benchmarking ist ein schwieriges Problem, besonders wenn Daten, Queries und Ergebnisse nicht leicht zugänglich und ausführbar sind. Den in diesem Beitrag besprochenen Benchmark haben wir leicht ausführbar gemacht: Die Skripte liegen im GitHub-Repository taxi-benchmark.

Das Repository enthält drei zentrale Python-Skripte:

generate_prepare_data.py: Lädt alle nötigen Dateien herunter und bereitet sie für den Benchmark vor.
benchmark.py: Führt den Benchmark aus und prüft die Ergebnisse.
analyse.py: Analysiert die Benchmark-Ergebnisse und erzeugt einige der in diesem Beitrag diskutierten Einsichten.

Der Benchmark ist nicht als makellos gedacht – kein Benchmark ist das. Wir glauben aber, dass das Teilen dieser Skripte ein positiver Schritt ist, und freuen uns über Beiträge, die sie sauberer und effizienter machen.

Das Repository enthält außerdem eine README mit detaillierten Anweisungen. Dieses Repository ist die Grundlage der Experimente in diesem Blogbeitrag.

Den Datensatz vorbereiten

Zuerst müssen Sie die Dateien herunterladen und vorbereiten, indem Sie python generate_prepare_data.py ausführen. Das lädt alle 65 Dateien in den Ordner ./data. Zusätzlich werden die Dateien dekomprimiert und zu einer einzigen großen Datei zusammengefügt.

Im Ordner ./data liegen dann 65 gzippte CSV-Dateien (von trips_xaa.csv.gz bis trips_xcm.csv.gz) und eine einzige große unkomprimierte CSV-Datei mit den gesamten Daten (decompressed.csv).

Unser Benchmark läuft dann in zwei Settings:

Über 65 komprimierte Dateien.
Über eine einzelne unkomprimierte Datei.

Sind die Dateien vorbereitet, führen Sie den Benchmark mit python benchmark.py aus.

Laden

Die Ladephase des Benchmarks läuft für jedes Setting sechs Mal. Aus den ersten fünf Läufen nehmen wir die Median-Ladezeit. Im sechsten Lauf erfassen wir Ressourcennutzung (z. B. CPU-Auslastung und Platten-Lese-/Schreibzugriffe).

Das Laden erfolgt mit einer In-Memory-DuckDB-Instanz: Die Daten werden nicht im DuckDB-Speicher persistiert und existieren nur, solange die Verbindung aktiv ist. Das ist wichtig, weil der Datensatz nicht in den Speicher passt und in einen temporären Bereich auf der Platte gespillt wird. Die Entscheidung, die Daten nicht zu persistieren, hat erheblichen Einfluss auf die Performance: Das Laden wird deutlich schneller, das Abfragen etwas langsamer, weil DuckDB eine unkomprimierte Darstellung nutzt. Wir haben das für den Benchmark so gewählt, weil unser Fokus auf dem CSV-Loader liegt, nicht auf den Queries.

Unser Tabellenschema steht in schema.sql.

[`schema.sql`](https://github.com/pdet/taxi-benchmark/blob/0.1/sql/schema.sql).

CREATE TABLE trips (
    trip_id                 BIGINT,
    vendor_id               VARCHAR,
    pickup_datetime         TIMESTAMP,
    dropoff_datetime        TIMESTAMP,
    store_and_fwd_flag      VARCHAR,
    rate_code_id            BIGINT,
    pickup_longitude        DOUBLE,
    pickup_latitude         DOUBLE,
    dropoff_longitude       DOUBLE,
    dropoff_latitude        DOUBLE,
    passenger_count         BIGINT,
    trip_distance           DOUBLE,
    fare_amount             DOUBLE,
    extra                   DOUBLE,
    mta_tax                 DOUBLE,
    tip_amount              DOUBLE,
    tolls_amount            DOUBLE,
    ehail_fee               DOUBLE,
    improvement_surcharge   DOUBLE,
    total_amount            DOUBLE,
    payment_type            VARCHAR,
    trip_type               VARCHAR,
    pickup                  VARCHAR,
    dropoff                 VARCHAR,
    cab_type                VARCHAR,
    precipitation           BIGINT,
    snow_depth              BIGINT,
    snowfall                BIGINT,
    max_temperature         BIGINT,
    min_temperature         BIGINT,
    average_wind_speed      BIGINT,
    pickup_nyct2010_gid     BIGINT,
    pickup_ctlabel          VARCHAR,
    pickup_borocode         BIGINT,
    pickup_boroname         VARCHAR,
    pickup_ct2010           VARCHAR,
    pickup_boroct2010       BIGINT,
    pickup_cdeligibil       VARCHAR,
    pickup_ntacode          VARCHAR,
    pickup_ntaname          VARCHAR,
    pickup_puma             VARCHAR,
    dropoff_nyct2010_gid    BIGINT,
    dropoff_ctlabel         VARCHAR,
    dropoff_borocode        BIGINT,
    dropoff_boroname        VARCHAR,
    dropoff_ct2010          VARCHAR,
    dropoff_boroct2010      BIGINT,
    dropoff_cdeligibil      VARCHAR,
    dropoff_ntacode         VARCHAR,
    dropoff_ntaname         VARCHAR,
    dropoff_puma            VARCHAR);

Der Loader für die 65 Dateien nutzt folgende Query:

COPY trips FROM 'data/trips_*.csv.gz' (HEADER false);

Der Loader für die einzelne unkomprimierte Datei nutzt diese Query:

COPY trips FROM 'data/decompressed.csv' (HEADER false);

Abfragen

Nach dem Laden führt das Benchmark-Skript jede der Benchmark-Queries fünf Mal aus, um die Ausführungszeit zu messen. Wichtig: Die Query-Ergebnisse werden gegen die zugehörigen Antworten geprüft. So können wir die Korrektheit des Benchmarks verifizieren. Die Queries sind identisch mit denen im Original-Benchmark „Billion Taxi Rides“.

Ergebnisse

Ladezeit

Obwohl es um viele Zeilen einer CSV-Datei mit 51 Spalten geht, kann DuckDB sie recht schnell aufnehmen.

Standardmäßig erhält DuckDB die Einfügereihenfolge der Daten, was die Performance negativ beeinflusst. In den folgenden Ergebnissen wurden alle Datensätze mit dieser Option auf false geladen.

SET preserve_insertion_order = false;

Alle Experimente liefen auf meinem Apple M1 Max mit 64 GB RAM. Wir vergleichen die Ladezeiten für eine einzelne unkomprimierte CSV-Datei und die 65 komprimierten CSV-Dateien.

Name	Zeit (min)	Mittlere Abweichung der CPU-Nutzung von 100%
Einzeldatei – unkomprimiert	11:52	31.57
Mehrere Dateien – komprimiert	13:52	27.13

Wenig überraschend ist das Laden aus mehreren komprimierten Dateien CPU-effizienter als aus einer einzelnen unkomprimierten Datei. Das zeigt die geringere mittlere Abweichung der CPU-Nutzung bei mehreren komprimierten Dateien: weniger verschwendete CPU-Zyklen. Zwei Hauptgründe: (1) Die komprimierten Dateien sind etwa achtmal kleiner als die unkomprimierte Datei, wodurch deutlich weniger Daten von der Platte geladen werden müssen und die CPU weniger auf Daten wartet. (2) Das parallele Laden mehrerer Dateien ist viel einfacher als das einer einzelnen Datei, weil jeder Thread eine Datei übernehmen kann.

Der Unterschied in der CPU-Effizienz zeigt sich auch in den Ausführungszeiten: Das Lesen einer einzelnen unkomprimierten Datei ist 2 Minuten schneller als das Lesen mehrerer komprimierter Dateien. Der Grund liegt in unserem Dekompressionsalgorithmus, der zugegebenermaßen nicht optimal entworfen ist. Das Lesen einer komprimierten Datei umfasst drei Aufgaben: (1) Daten von der Platte in einen komprimierten Buffer laden, (2) diese Daten in einen dekomprimierten Buffer dekomprimieren und (3) den dekomprimierten Buffer verarbeiten. In unserer aktuellen Implementierung sind Aufgaben 1 und 2 zu einer Operation zusammengefasst: Wir können nicht weiterlesen, bis der aktuelle Buffer vollständig dekomprimiert ist – das führt zu Leerlaufzyklen.

Unter der Haube

Wir können uns auch ansehen, was unter der Haube passiert, um unsere Schlussfolgerung zur Ladezeit zu prüfen.

In der Abbildung unten sehen Sie einen Snapshot der CPU- und Plattenauslastung für den Lauf „Einzeldatei – unkomprimiert“. 100 % CPU-Auslastung zu erreichen ist schwierig, und wir erleben häufig Stalls durch Schreibzugriffe auf die Platte, weil wir eine Tabelle aus einem Datensatz erzeugen, der nicht in den Speicher passt. Ein weiterer Punkt: Die CPU-Auslastung hängt eng mit den Plattenlesevorgängen zusammen – unsere Threads warten oft auf Daten, bevor sie sie verarbeiten. Asynchrones I/O für den CSV-Reader/Writer könnte die Performance bei paralleler Verarbeitung deutlich verbessern, weil ein einzelner Thread den Großteil unseres Platten-I/O übernehmen könnte, ohne die CPU-Auslastung negativ zu beeinflussen.

Unten ein ähnlicher Snapshot für das Laden der 65 komprimierten Dateien. Wir erleben häufig Stalls bei Schreibzugriffen; die CPU-Auslastung ist aber deutlich besser, weil wir weniger auf das Laden der Daten warten (die Daten sind etwa 8-mal kleiner als im unkomprimierten Fall). Parallelisierung ist hier auch viel einfacher. Wie im unkomprimierten Fall könnten diese Lücken in der CPU-Auslastung durch asynchrones I/O gemildert werden, plus einem zerlegten Dekompressionsalgorithmus.

Query-Zeiten

Der Vollständigkeit halber liefern wir auch die Ergebnisse der vier Queries auf einem MacBook Pro mit M1-Pro-CPU. Der Vergleich zeigt die Zeitunterschiede zwischen dem Abfragen einer Datenbank, die nicht in den Speicher passt, über eine reine In-Memory-Verbindung (ohne Storage) und einer, bei der die Daten zuerst geladen und in der Datenbank persistiert wurden.

Name	Zeit – ohne Storage (s)	Zeit – mit Storage (s)
Q 01	2.45	1.45
Q 02	3.89	0.80
Q 03	5.21	2.20
Q 04	11.2	3.12

Der Hauptunterschied: Nutzt DuckDB eine Storage-Datei, sind die Daten stark komprimiert, was zu viel schnellerem Zugriff beim Abfragen führt. Ohne persistenten Storage speichert unsere In-Memory-Datenbank Daten vorübergehend in einer unkomprimierten .tmp-Datei, um Memory Overflow zu erlauben – das erhöht das Platten-I/O und führt zu langsameren Query-Ergebnissen. Das wirft eine mögliche Untersuchungsfrage auf: Ob Kompression temporärer Daten lohnenswert wäre.

Wie dieser Datensatz erzeugt wurde

Der Originalbeitrag hat den Datensatz aus CSV-Dateien der NYC Taxi and Limousine Commission erzeugt. Ursprünglich enthielten diese Dateien genaue Längen- und Breitengrade für Abholung und Ziel. Ab Mitte 2016 wurden diese genauen Koordinaten aus Datenschutzgründen durch Geometrieobjekte für Abholung und Ziel anonymisiert. (Es gibt sogar Geschichten von zerbrochenen Ehen, weil jemand die tatsächlichen Taxiziele nachgesehen hat.) In den letzten Jahren hat die TLC die Daten als Parquet-Dateien neu verteilt und diese Datenpunkte vollständig anonymisiert, einschließlich Daten vor Mitte 2016.

Das ist ein Problem, weil der Datensatz aus „Billion Taxi Rides in Redshift“ auf diesen detaillierten Informationen beruht. Schauen wir uns folgendes Datenschnipsel an:

649084905,VTS,2012-08-31 22:00:00,2012-08-31 22:07:00,0,1,-73.993908,40.741383000000006,-73.989915,40.75273800000001,1,1.32,6.1,0.5,0.5,0,0,0,0,7.1,CSH,0,0101000020E6100000E6CE4C309C7F52C0BA675DA3E55E4440,0101000020E610000078B471C45A7F52C06D3A02B859604440,yellow,0.00,0.0,0.0,91,69,4.70,142,54,1,Manhattan,005400,1005400,I,MN13,Hudson Yards-Chelsea-Flatiron-Union Square,3807,132,109,1,Manhattan,010900,1010900,I,MN17,Midtown-Midtown South,3807

Wir sehen genaue Längen- und Breitengrade: -73.993908, 40.741383000000006, -73.989915, 40.75273800000001, dazu einen PostGIS-Geometry-Hex-Blob aus diesen Informationen: 0101000020E6100000E6CE4C309C7F52C0BA675DA3E55E4440, 0101000020E610000078B471C45A7F52C06D3A02B859604440 (erzeugt als ST_SetSRID(ST_Point(longitude, latitude), 4326)).

Da diese Information für den Datensatz wesentlich ist, ist es nicht mehr möglich, Dateien wie im Beitrag „Billion Taxi Rides in Redshift“ zu erzeugen – die detaillierten Ortsdaten fehlen. Das Internet vergisst aber nie. Deshalb haben wir Instanzen des Originaldatensatzes aus verschiedenen Quellen gefunden, etwa [1], [2] und [3]. Aus diesen Quellen haben wir die Original-CSV-Dateien mit Wetterinformationen aus den Skripten kombiniert, auf die der Beitrag „Billion Taxi Rides in Redshift“ verweist.

Wie unterscheidet sich dieser Datensatz vom Original?

Es gibt zwei wesentliche Unterschiede zwischen dem von uns verteilten Datensatz und dem aus „Billion Taxi Rides in Redshift“:

Unser Datensatz enthält Daten bis zum letzten Datum, an dem Längen- und Breitengrade verfügbar waren (30. Juni 2016), während der Originalbeitrag nur Daten bis Ende 2015 enthielt (verständlich, der Beitrag erschien im Februar 2016).
Wir haben auch Uber-Fahrten aufgenommen, die im Originalbeitrag ausgeschlossen waren.

Wenn Sie den Benchmark mit einem Datensatz möglichst nah am Original ausführen möchten, können Sie eine neue Tabelle erzeugen, indem Sie die zusätzlichen Daten herausfiltern. Zum Beispiel:

CREATE TABLE trips_og AS
    FROM trips
    WHERE pickup_datetime < '2016-01-01'
      AND cab_type != 'uber';

Fazit

In diesem Blogbeitrag haben wir gezeigt, wie man den Taxi-Benchmark auf DuckDB ausführt, und alle Skripte bereitgestellt, damit Sie auch Ihr bevorzugtes System benchmarken können. Wir haben außerdem gezeigt, wie dieser sehr relevante Benchmark unsere Operatoren bewerten und Einsichten für weitere Verbesserungen liefern kann.