Data-Processing-Engines: Polars, DataFusion, Ray und Spark

Viacheslav Dubrov

2026-02-21 · Aktualisiert 2026-07-15 · 14 Min. · KI-Infrastruktur

Automatische Übersetzung Dieser Artikel wurde automatisch aus der englischen Originalversion übersetzt.

Jahrelang war Pandas für tabellarische In-Memory-Verarbeitung und Apache Spark für die verteilte Variante zuständig. Diese Aufteilung funktionierte gut, solange die Daten strukturiert waren.

Moderne Pipelines verarbeiten zusätzlich Bilder, Audio und Video. Bei diesen Workloads kann CPU-Decoding die GPU-Inference ausbremsen, während JVM Garbage Collection und Pythons Global Interpreter Lock den Throughput begrenzen. Neue Engines nutzen Rust und Apache Arrow, um diese Kosten zu reduzieren.

Um die Optionen zu vergleichen, habe ich Pandas, Polars, DataFusion, Daft und natives Rust auf zwei realen Datasets gebenchmarkt. Spark und Ray werden in separaten verteilten Notebooks behandelt. NYC-Taxifahrten stehen für tabellarische Verarbeitung; Food-101-Bilder repräsentieren eine multimodale Pipeline.

Das engine-comparison-demo Repository enthält den Code. Führe ihn auf deiner eigenen Hardware aus, denn die Performance einer Engine hängt von Maschine, Dataset und Workload ab.

Typen von Data-Processing-Workloads

Engines sind spezialisiert. Die erste Frage lautet daher: Welche Art von Workload liegt tatsächlich vor? Die wichtigste Unterscheidung ist strukturiert versus multimodal.

Strukturierte oder tabellarische Daten umfassen Filtering, Aggregation und Joins. Sie sind normalerweise CPU-bound und passen häufig in den Arbeitsspeicher. Viele solcher Queries können auf einer modernen Maschine ausgeführt werden. Dennoch bestimmen Memory, I/O, Concurrency, Recovery und die Form des Workloads, ob ein Cluster sinnvoll ist.

Multimodale AI verarbeitet Bilder, Audio oder Video. Die Inference kann auf einer GPU laufen, während CPU-Decoding, Remote Reads oder Batching die Pipeline begrenzen. Welche Stage den Throughput begrenzt, ändert sich mit dem Instance-Typ und den Operatoren. Miss daher die Auslastung über den gesamten Pfad, statt die GPU isoliert zu dimensionieren.

Die neueren Engines zielen auf unterschiedliche Bereiche dieses Spektrums. Native Execution kann Python-Overhead reduzieren, Arrow-kompatible Interfaces können den Datenaustausch günstiger machen, und Streaming Execution kann den Memory-Bedarf bei Datasets begrenzen, die größer als der RAM sind. Keiner dieser Vorteile ist für jeden Operator oder jede Konvertierung automatisch gegeben.

Teil 1: Single-Node-Processing

Pandas setzt auf ein eager, In-Memory-Programmiermodell. Viele gängige Operationen erzeugen Intermediates, und die parallele Ausführung wird nicht über einen Query Optimizer koordiniert. Dieser Trade-off hält explorativen Code direkt, kann bei größeren analytischen Scans aber teuer werden. Im PDS-H-Benchmark von Polars benötigte Pandas bei Scale Factor 10 etwa 365 Sekunden für die Suite, während Polars’ Streaming Engine 3,89 Sekunden benötigte. Das Ergebnis von ungefähr 94x beschreibt diesen Benchmark und diese Konfiguration; es ist kein allgemeiner Umrechnungsfaktor von Pandas zu Polars.

Polars für lokale tabellarische Pipelines

Polars ist ein praktischer Default, wenn eine lokale DataFrame-Pipeline über eager, Single-Process-Execution hinausgewachsen ist. Die Lazy API erstellt vor der Ausführung einen Query Plan und ermöglicht Optimierungen wie Predicate Pushdown und Projection Pruning. Die Engine kann Operatoren anschließend parallel und, sofern unterstützt, in Streaming-Batches ausführen.

Mit wachsender Datenmenge wird der Abstand größer. Bei Scale Factor 100 (~100 GB) war Polars’ Streaming Engine nach 23,94 Sekunden fertig, gegenüber 152,27 Sekunden für die eigene In-Memory Engine – also etwa 6x schneller, bei Daten, die größer als der RAM sind.

Das Folgende ist ein illustrativer API-Entwurf und kein wörtlicher Auszug aus der aktuellen Version von engine_comparison_examples.ipynb oder tabular.py:

import polars as pl

# scan_parquet reads only the schema — no data loaded yet
q = (
    pl.scan_parquet("yellow_tripdata_2024-01.parquet")
    .filter(
        (pl.col("trip_distance") > 5.0)
        & (pl.col("total_amount") > 30.0)
    )
    .group_by("payment_type")
    .agg(
        pl.col("total_amount").mean().alias("avg_fare"),
        pl.col("trip_distance").mean().alias("avg_distance"),
        pl.len().alias("trip_count"),
    )
)

# The entire plan is optimized and executed here, in parallel
result = q.collect()

Der entscheidende Unterschied liegt im Execution Model. In der obigen Lazy Query kann Polars den Filter und die Spaltenauswahl in den Parquet-Scan verschieben. Dadurch lassen sich möglicherweise Row Groups überspringen und nicht benötigte Spalten vermeiden. Eine typische eager Pandas-Pipeline verfügt über keinen Whole-Query-Plan zur Optimierung. Durch die sorgfältige Nutzung von Parquet-Filtern, ausgewählten Spalten und alternativen Backends lässt sich ein Teil des Abstands jedoch aufholen.

DataFusion für eingebettete Query Engines

Während Polars eine Library ist, die du direkt verwendest, ist DataFusion die Engine, auf der du andere Engines aufbaust. Sie bildet die Grundlage für InfluxDB 3.0, GreptimeDB und Apples Comet Spark Accelerator.

In einem im November 2024 vom DataFusion-Projekt veröffentlichten ClickBench-Lauf lag DataFusion bei den getesteten Single-Node-Parquet-Konfigurationen vorn. Das Embucket-Team veröffentlichte später eine Vendor Case Study für TPC-H Scale Factor 1000 mit datafusion-cli auf r6gd.metal, 64 vCPUs, 512 GB RAM, lokalem NVMe und Timings bei aktivem Cache. Q18 und Q21 schlugen im ursprünglichen Lauf fehl oder liefen übermäßig lange und wurden anschließend für die abgeschlossenen Ergebnisse strukturell umgeschrieben. Das zeigt Scale-up auf einem Node unter diesem Setup, aber kein allgemeines TPC-H-Ergebnis.

Das Folgende ist ein illustrativer API-Entwurf und kein wörtlicher Auszug aus der aktuellen Version von engine_comparison_examples.ipynb oder tabular.py:

from datafusion import SessionContext

ctx = SessionContext()
ctx.register_parquet("taxi", "yellow_tripdata_2024-01.parquet")

# SQL executed directly against Parquet — no intermediate copies
df = ctx.sql("""
    SELECT payment_type,
           COUNT(*)           AS trip_count,
           AVG(trip_distance) AS avg_distance,
           AVG(total_amount)  AS avg_fare
    FROM taxi
    WHERE trip_distance > 5.0 AND total_amount > 30.0
    GROUP BY payment_type
    ORDER BY trip_count DESC
""")
result = df.to_pandas()

Die Stärke von DataFusion ist seine Modularität. Die Extension APIs decken Custom Catalogs, Table Providers, Optimizer Rules und Execution Plans ab. Deshalb taucht DataFusion überall dort auf, wo jemand eine eigene Data Platform entwickelt oder eine Query Engine in sein Produkt einbettet.

Daft für multimodale Daten

Daft macht Bilder, Audio, Video und Embeddings zu Bestandteilen des DataFrame-Workflows. Die Engine stellt native Expressions für Operationen wie Image Decoding und URL-Downloads bereit, sodass gängige Preprocessing-Pfade keine Python-Schleife über einzelne Rows benötigen. Diese Integration ist der wichtigste Grund, Daft einer allgemeinen tabellarischen Engine vorzuziehen.

Das Folgende ist ein illustrativer API-Entwurf und kein wörtlicher Auszug aus der aktuellen Version von engine_comparison_examples.ipynb oder multimodal.py:

import daft

df = daft.read_parquet("yellow_tripdata_2024-01.parquet")

result = (
    df.where(
        (daft.col("trip_distance") > 5.0)
        & (daft.col("total_amount") > 30.0)
    )
    .groupby("payment_type")
    .agg(
        daft.col("total_amount").mean().alias("avg_fare"),
        daft.col("trip_distance").mean().alias("avg_distance"),
        daft.col("trip_distance").count().alias("trip_count"),
    )
    .collect()
)

Die API wirkt vertraut, wenn du Pandas kennst. Die Engine nutzt jedoch denselben Rust- und Arrow-Stack wie Polars und DataFusion. Daft spielt seine Stärken aus, wenn die „Rows“ in deinem DataFrame aus Bildern, PDFs oder Tensors bestehen.

Single-Node-Benchmark

Ich habe Pandas, Polars, DataFusion und Daft sowie natives Rust über Polars-rs auf etwa 41 Millionen NYC Yellow Taxi Trips des vollständigen Jahres 2024 ausgeführt. Spark und Ray erscheinen in den verteilten Beispielen, nicht in diesem Single-Node-Benchmark. Die vollständigen Ergebnisse stehen im Demo-Repository:

Das Diagramm dient als Kontext für den begleitenden Benchmark, ist aber kein fairer Eins-zu-eins-Ranking der Engines: Pandas lädt die Trip- und Zonendaten vor seiner gemessenen Funktion, während Polars, DataFusion, Daft und Rust unterschiedliche Datei-Reads, Registrierungen oder Query-Setup-Arbeiten innerhalb des gemessenen Abschnitts ausführen. Das Script wiederholt jede Operation dreimal und gibt nach dem Laden der Imports den Median aus. Der Zustand des Filesystem-Caches, Warm-up, CPU und andere Laufbedingungen können die Werte verändern. Richte die Timing-Grenzen aneinander aus und führe den Test unter kontrollierten Bedingungen erneut aus, bevor du die Reihenfolge als Performance-Ergebnis interpretierst.

Bei dieser Größe (~660 MB Parquet) werden die drei neueren Python-Engines schnell fertig. Die oben genannte Zahl von 94x zwischen Polars und Pandas stammt aus PDS-H; mein Taxi-Workload mit 41 Millionen Rows erzeugte einen kleineren Abstand. In diesem Lauf mit gemischten Timing-Grenzen lagen Polars, DataFusion, Daft und Polars-rs im selben groben Bereich, wobei DataFusion die niedrigste gemeldete Zeit erzielte. Diese Reihenfolge gilt für diese Query und diese Messgrenze, nicht allgemein für die Engines. Wenn Engines so dicht beieinanderliegen, sollten API-Fit und wiederholte Messungen mit repräsentativen Daten die Entscheidung bestimmen.

Teil 2: multimodale Daten

Tabellarisches ETL verkleinert Daten normalerweise: Filtern, aggregieren und weniger schreiben, als gelesen wurde. Multimodale AI-Pipelines machen das Gegenteil. Ein einzelner Dokumentpfad kann sich in Dutzende Text-Chunks und Embedding-Vektoren auffächern.

In einer konventionellen PySpark-Pipeline gelangen Bilder und Audio häufig als Binärdaten hinein und werden zur Decodierung oder Transformation in Python-Libraries überführt. Das Überschreiten der JVM/Python-Grenze und das Serialisieren der Ergebnisse kann einen großen Teil der Wall-Clock-Zeit ausmachen. Spark unterstützt Arrow-basierte Pfade, vectorized UDFs und Accelerator-Integrationen. Diese Optionen erfordern jedoch explizites Design und Messungen.

Pipelined Execution und GPU-Auslastung

Spark plant die Arbeit in Stages, die durch Shuffle-Grenzen getrennt sind. Eine einfache Implementierung kann Downloads, Decoding und Inference in eine Sequenz bringen, in der entweder CPU- oder GPU-Kapazität wartet. Spark unterstützt GPU Resource Scheduling und Plugins, Leerlauf ist also nicht unvermeidbar. Overlap und Backpressure sind jedoch keine automatischen Eigenschaften eines gewöhnlichen PySpark-DataFrame-Jobs.

Pipelined Execution ist die Alternative. Statt Stages nacheinander auszuführen, überlappt die Engine I/O, CPU-Arbeit und GPU-Inference. Wenn Stages und Buffer ausbalanciert sind, kann dieser Overlap Leerlauf reduzieren und den Throughput erhöhen. Startup und Drain, Backpressure oder eine langsamere Stage können jedoch weiterhin dazu führen, dass Ressourcen warten.

Image-Processing-Benchmark

Ich habe Image Processing mit 500 echten Fotografien aus dem Food-101-Dataset gebenchmarkt. Die Ergebnisse aus dem Demo-Repository:

Polars und DataFusion fehlen, weil dieser Test native Image Operations und keine tabellarischen Expressions untersucht. Bei diesem Lauf mit 500 Bildern war Dafts Image Path etwa 3,7x schneller als die Pandas-+—Pillow-Baseline. Der exakte Wert von multimodal_results.json 1.0148517909983639s geteilt durch den Wert von rust_multimodal_results.json 0.235865209s ergibt 4.30267692, sodass die Rust-Implementierung 4,3x schneller war. Das begleitende README zeigt gerundete Werte für 1.01s und 0.24s, deren Verhältnis 4,2x beträgt. Das ist lediglich Display-Rounding; die JSON-Werte sind maßgeblich. Das Sample eignet sich, um Orchestration-Overhead sichtbar zu machen, ist aber allein zu klein, um den Production-Throughput vorherzusagen.

Teil 3: Distributed Processing

Sobald eine einzelne Maschine nicht mehr ausreicht, musst du entscheiden, wie du die Arbeit verteilst.

Apache Spark

Spark ist eine ausgereifte Option für großes tabellarisches ETL, Shuffle-intensive Joins und Organisationen, die bereits sein Ökosystem betreiben. Seine lineage-basierte Recovery und die breite Plattformunterstützung sind relevant, wenn Reliability und operative Vertrautheit wichtiger sind als lokale Geschwindigkeit. Gemischte CPU/GPU-Pipelines erfordern eine bewusstere Resource Configuration und ein sorgfältigeres Pipeline-Design als der SQL-Pfad. Genau hier bieten Ray Data und Daft eine fokussiertere Abstraktion.

Das Folgende ist ein illustrativer Spark-API-Entwurf und kein wörtlicher Auszug aus der aktuellen Version von distributed_spark.ipynb oder spark_etl.py:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder \
    .appName("TaxiETL") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

orders = spark.read.parquet("s3a://lake/taxi/*.parquet")
zones = spark.read.csv("s3a://lake/taxi_zones.csv", header=True)

# Spark excels at this: joining massive tables with a shuffle
result = (
    orders
    .filter(F.col("trip_distance") > 5.0)
    .join(zones, orders.PULocationID == zones.LocationID, "inner")
    .groupBy("Borough")
    .agg(
        F.sum("total_amount").alias("total_revenue"),
        F.avg("total_amount").alias("avg_fare"),
    )
    .orderBy(F.desc("total_revenue"))
)

Ray Data für heterogene Compute-Workloads

Ray Data wurde für AI-Workloads entwickelt. Anders als die Stage Barriers von Spark nutzt Ray Data ein Streaming Model, das GPUs ausgelastet hält. Das wichtigste Feature für AI-Pipelines ist Mixed Resource Scheduling: Du kannst deklarieren, dass ein Actor „1 GPU, 4 CPUs“ und ein anderer ausschließlich CPUs benötigt; Ray übernimmt den Rest.

Amazon berichtete nach der Migration ausgewählter interner Data-Processing-Workloads von Spark zu Ray von mehr als $120 Millionen jährlichen Einsparungen. Der Bericht nennt im Proof of Concept eine um 91 % bessere Cost Efficiency und in Production 82 % pro GiB S3-Input. Die Größenordnung ist bemerkenswert, aber es handelt sich um eine Migration Case Study für spezifische Workloads, nicht um eine zu erwartende Einsparung bei einer typischen Ray-Einführung.

Das Folgende ist ein illustrativer Ray-Data-API-Entwurf und kein wörtlicher Auszug aus der aktuellen Version von distributed_ray.ipynb oder ray_inference.py:

import ray

ray.init()

ds = ray.data.read_images("s3://my-bucket/food101/")

class ImageClassifier:
    def __init__(self):
        import torch
        from torchvision.models import resnet18, ResNet18_Weights
        self.model = resnet18(weights=ResNet18_Weights.DEFAULT).cuda()
        self.model.eval()
        self.preprocess = ResNet18_Weights.DEFAULT.transforms()

    def __call__(self, batch):
        import torch
        tensors = torch.stack([
            self.preprocess(img) for img in batch["image"]
        ]).cuda()
        with torch.no_grad():
            preds = self.model(tensors)
        return {
            "prediction": preds.argmax(dim=1).cpu().numpy(),
            "confidence": preds.max(dim=1).values.cpu().numpy(),
        }

# ActorPoolStrategy creates persistent GPU workers
predictions = ds.map_batches(
    ImageClassifier,
    compute=ray.data.ActorPoolStrategy(size=4),
    num_gpus=1,
    batch_size=64,
)
predictions.write_parquet("s3://output/predictions/")

Das Verhältnis von CPU zu GPU verdient hier besondere Aufmerksamkeit. In Anyscales Benchmarks skalierte Ray Data weiter, als die Zahl der CPUs pro GPU stieg, während die anderen getesteten Engines ein Plateau erreichten. Die Quelle berichtet bei Image Inference von bis zu 3x Speedup, wenn die CPU-Starvation abnimmt, weil die CPU-Feeder mit der GPU Schritt halten können.

Verteiltes Daft

Daft skaliert über seine Flotilla Engine: einen Swordfish Worker pro Node, mit Flotilla darüber für das clusterweite Scheduling. Swordfish übernimmt die lokale Rust Execution und pipelined I/O mit Compute über Streaming in Small Batches. So bleibt jeder Node ausgelastet, ohne auf die nächste Stage zu warten.

In von Daft veröffentlichten Benchmarks war Flotilla bei vier multimodalen Workloads 4–18x schneller als die getesteten Spark-Implementierungen. Die größte Differenz trat bei einem Video-Object-Detection-Fall auf. Interpretiere dieses Ergebnis als Beleg dafür, dass das Execution Design bei solchen Pipelines wichtig ist. Vergleiche es anschließend mit den konkurrierenden Anyscale-Ergebnissen weiter unten und mit einem lokalen Test.

Anyscale, das Unternehmen hinter Ray, veröffentlichte einen konkurrierenden Benchmark, in dem Ray Data den Abstand auf CPU-starken Instances nach Tuning schloss oder umkehrte. Zusammen liefern die beiden Vendor Studies einen Grund, repräsentative Operatoren und CPU/GPU-Verhältnisse zu testen, statt eine dauerhafte Rangliste abzuleiten.

Das Folgende ist ein illustrativer Daft-API-Entwurf und kein wörtlicher Auszug aus der aktuellen Version von distributed_daft.ipynb oder daft_pipeline.py:

import daft
import numpy as np
from daft import col

# Daft's Flotilla engine distributes work across the cluster
df = daft.read_parquet("s3://data-lake/pdf_metadata/*.parquet")

# Download PDFs — Daft parallelizes downloads in Rust
df = df.with_column("pdf_bytes", col("pdf_url").download())

# Define a GPU class UDF for batched embedding generation
@daft.cls(gpus=1)
class TextEmbedder:
    def __init__(self):
        from sentence_transformers import SentenceTransformer
        self.model = SentenceTransformer("all-MiniLM-L6-v2", device="cuda")

    @daft.method.batch(
        return_dtype=daft.DataType.fixed_size_list(daft.DataType.float32(), 384)
    )
    def __call__(self, text_col):
        texts = text_col.to_pylist()
        embeddings = self.model.encode(texts, batch_size=32)
        return np.asarray(embeddings, dtype=np.float32)

# Daft schedules CPU downloads and GPU embeddings simultaneously
embedder = TextEmbedder()
df = df.with_column("embedding", embedder(col("text")))
df.write_parquet("s3://output/embeddings/")

Vergleich verteilter Engines

Teil 4: Die Rolle von Rust und Arrow

Unterhalb des Wettbewerbs ist die Konvergenz die interessantere Geschichte. Polars, Daft und DataFusion nutzen Rust intensiv, während Ray neben seinen Python-APIs native Komponenten enthält. Alle können Daten über Teile des Apache-Arrow-Ökosystems austauschen.

Das Arrow PyCapsule Interface (__arrow_c_stream__-Protokoll) bietet kompatiblen Libraries einen Standardweg zum Austausch von Arrow-Streams. Transfers können eine zeilenweise Serialisierung vermeiden und Buffer wiederverwenden, wenn Schemas und Memory Layouts kompatibel sind. Materialization, Rechunking, Type Conversion oder Device Transfer können weiterhin Daten kopieren. Überprüfe die Übergabe daher mit Profiling, statt anzunehmen, dass sie kostenlos ist.

Das Folgende ist ein illustrativer API-Entwurf und kein eigenständiges Script: Es setzt voraus, dass events.parquet bereits existiert. Das begleitende engine_comparison_examples.ipynb erzeugt .data/events.parquet in seinen Setup-Zellen vor dem Arrow-Beispiel.

from datafusion import SessionContext
import polars as pl

# Compute in DataFusion (Rust-native execution)
ctx = SessionContext()
ctx.register_parquet("events", "events.parquet")
df = ctx.sql("""
    SELECT user_id, COUNT(*) as event_count
    FROM events WHERE event_type = 'purchase'
    GROUP BY user_id HAVING COUNT(*) > 5
""")

# Convert through Arrow; compatible buffers may be reused
arrow_table = df.to_arrow_table()
df_polars = pl.from_arrow(arrow_table)

# Continue analysis in Polars
result = df_polars.with_columns(
    pl.col("event_count").rank().alias("rank")
).sort("rank")

Warum verwenden diese neueren Engines Rust für native Execution?

1. Kein Tracing Garbage Collector im Rust-Code. Ownership gibt nativen Operatoren eine direktere Kontrolle über Allocation und Deallocation. Das kann GC-bedingte Latency reduzieren, verhindert aber weder Memory Pressure noch Out-of-Memory-Fehler.

2. Kompakte native Repräsentationen. Rust-Strukturen enthalten keine Java-Object-Headers. Der praktische Vorteil hängt vom Layout der Engine ab: Columnar-JVM-Systeme vermeiden ebenfalls, jeden Wert als individuelles Objekt zu repräsentieren.

3. Stärkere Concurrency Checks. Safe Rust schließt viele Data Races bereits zur Compile-Zeit aus. Engine-Code kann weiterhin unsafe-Blöcke und logische Concurrency-Bugs enthalten, aber die Sprache reduziert die Fehlerfläche.

Zusammen mit Arrows Columnar Memory Layout können diese Entscheidungen Allocation- und Serialisierungs-Overhead reduzieren. Sie beseitigen ihn jedoch nicht an jeder Grenze; Python-Objekte, Network Transport, inkompatible Schemas und Device Movement bleiben relevant.

Teil 5: Mehrere Engines einsetzen

Du musst nicht die gesamte Platform durch eine einzige Engine führen. Composition ist sinnvoll, wenn die Übergabe zwischen Engines weniger kostet, als ein System für jeden Workload passend zu machen.

Eine mögliche Relay-Architektur nutzt Spark für etablierte Lakehouse-Joins, schreibt in ein offenes Table- oder File-Format und übergibt das Ergebnis für GPU-Inference oder multimodale Transformationen an Ray Data oder Daft. Polars oder DuckDB können lokale Analysen über denselben Dateien ausführen. Ein kleineres Team benötigt möglicherweise nur eine dieser Engines. Füge eine weitere hinzu, wenn ein gemessener Bottleneck die operative Grenze rechtfertigt.

Möglich wird diese Composition durch Interoperabilität. Parquet ist in diesen Beispielen das gemeinsame File-Handoff-Format, während Arrow den In-Memory-Austausch ermöglichen kann. Die Unterstützung für Delta und Iceberg hängt von der Engine und dem Connector ab; überprüfe daher jede Grenze. Auch ein File Handoff verursacht I/O- und Conversion-Kosten und ist nicht automatisch kostenlos.

Engine-Auswahl nach Evaluationsfall

Diagonal Scaling

Lange Zeit bestand die Wahl zwischen Scale-up (eine größere Maschine) und Scale-out (mehr Maschinen). Polars Cloud verfolgt einen Mittelweg, den das Unternehmen Diagonal Scaling nennt. Skaliere horizontal, während du aus Cloud Storage liest, um die I/O maximal auszunutzen. Sobald Filter und Aggregationen die Datenmenge reduziert haben, wechselst du auf einen einzelnen großen Node. Der verteilte Shuffle entfällt.

Die übergeordnete Lehre lautet, die Kosten pro erfolgreichem Job zu vergleichen, nicht den stündlichen Instance-Preis. Ein größerer Node kann günstiger sein, wenn die Runtime-Reduktion seinen höheren Preis überwiegt. Das Ergebnis hängt jedoch von I/O, Memory Pressure und der Menge an Arbeit ab, die vor einem Shuffle entfällt.

Vergleich reproduzieren

Das begleitende Demo-Repository enthält die Scripts, Notebooks und die für die Vergleiche verwendete Umgebung. Der folgende Quick Command nutzt das standardmäßige Taxi-Dataset für das vollständige Jahr 2024 und entspricht damit dem weiter oben gezeigten größeren Lauf.

# Install dependencies
uv sync

# Run the tabular benchmark (~41M NYC taxi trips from full-year 2024)
uv run python -m engine_comparison.benchmarks.tabular

# Run the multimodal benchmark (500 real food photos)
uv run python -m engine_comparison.benchmarks.multimodal

# Run native Rust benchmarks (Polars-rs + image crate)
cd rust_benchmark && cargo run --release && cd ..

Das Repository enthält außerdem Notebooks für API-Vergleiche nebeneinander sowie Docker-Compose-Setups für lokale verteilte Läufe mit Spark, Ray und Daft.

Zentrale Erkenntnisse

1. Belege, dass eine Maschine nicht ausreicht. Scale-up-Ergebnisse zeigen, dass manche großen analytischen Scans auf einen einzelnen Node passen. Miss Memory, I/O, Spill und Recovery-Anforderungen, bevor du den Overhead eines Clusters akzeptierst.

2. Execution Models sind wichtiger als vertraute Syntax. Lazy Planning, Pushdown, Streaming und parallele Operatoren erklären einen großen Teil des Abstands zwischen einem eager DataFrame-Script und einer analytischen Engine.

3. Miss die gesamte Accelerator-Pipeline. Decoding, Network I/O, Batching, Serialisierung und Backpressure bestimmen, ob eine GPU ausgelastet bleibt. Ray Data und Daft machen diesen Overlap zu einer zentralen Abstraktion; Spark kann ihn mit zusätzlichem Design und Tooling unterstützen.

4. Erstelle die Shortlist anhand des Workloads und benchmarke anschließend. Nutze den Ecosystem-Fit, um das Feld einzugrenzen, und einen repräsentativen End-to-End-Job für die Auswahl. Vendor-Benchmark-Ergebnisse sind Hypothesen, keine Garantien.

5. Offene Formate halten die Entscheidung reversibel. Arrow-kompatible Interfaces, Parquet und offene Table Formats können Handoff-Kosten reduzieren. Prüfe, ob eine bestimmte Konvertierung Buffer wiederverwendet oder kopiert.

Wenn du einen Ausgangspunkt brauchst, probiere Polars für eine lokale tabellarische Pipeline und Daft oder Ray Data für eine gemischte CPU/GPU-Pipeline. Wähle Spark, wenn dessen verteilte Execution, Ökosystem oder bestehende operative Basis eine Anforderung erfüllt. Die Dataset-Größe allein ist ein schwaches Auswahlkriterium.

Referenzen

Benchmarks und Performance-Daten

• Polars-PDS-H-Benchmarks – Polars Streaming vs. Pandas bei SF-10 (94x) und SF-100 (6,4x gegenüber In-Memory)
• DataFusion-ClickBench-Ergebnisse (November 2024) – Vom Projekt veröffentlichte Single-Node-Parquet-Ergebnisse
• Embucket: TPC-H SF-1000 auf DataFusion – Vendor Case Study zu r6gd.metal mit Timings bei aktivem Cache; Q18 und Q21 wurden umgeschrieben, nachdem der ursprüngliche Lauf fehlgeschlagen war oder übermäßig lange dauerte
• Amazon: Migration von Spark zu Ray – $120M Einsparungen pro Jahr, 82 % Cost Efficiency in Production
• Daft-Flotilla-Benchmarks – 4–18x schneller als Spark bei multimodalen Workloads
• Anyscale-Ray-vs.-Daft-Benchmarks – Konkurrierender Vendor-Benchmark für multimodale Workloads

Engines und Frameworks

• Polars – Rust-DataFrame-Library mit Lazy Execution
• Apache DataFusion – Einbettbare Rust Query Engine (GitHub)
• Daft – Multimodal-native verteilte DataFrame-Engine
• Ray – Verteiltes Compute-Framework für AI (Data Internals)
• Apache Spark – Verteilte ETL- und SQL-Engine
• DuckDB – In-Process-analytische SQL-Engine
• Dask – Python-natives Parallel Computing

Architektur und Ökosystem

• Polars Cloud: Diagonal Scaling – Dynamisches vertikales und horizontales Scaling
• Daft-Flotilla-Architektur – Verteilte Engine aus Swordfish und Flotilla
• Apache DataFusion Comet – Ursprünglich bei Apple entwickelter Spark Accelerator
• Apache Arrow – Columnar In-Memory-Format (Flight RPC, PyCapsule Interface)
• InfluxDB 3.0 + DataFusion – Auf DataFusion basierende Time-Series-Datenbank
• GreptimeDB – DataFusion nutzende Observability-Datenbank

Demo-Repository

• engine-comparison-demo – Begleitender Code zu diesem Artikel: Benchmarks, Notebooks und Docker Compose für Spark/Ray/Daft

Datasets

• NYC Taxi Trip Records – Gelbe Taxi-Daten der NYC TLC (Parquet, ~2,9 Mio. Rows/Monat)
• Food-101-Dataset – 101.000 Food-Bilder von ETH Zürich (Bossard et al., ECCV 2014)