2026-02-21 · Aktualisiert 2026-07-15

[!NOTE] Automatische Übersetzung Dieser Artikel wurde automatisch aus der englischen Originalversion übersetzt.

Vergleich moderner Datenverarbeitungsengines: Polars, DataFusion, Daft, Ray Data, Pandas und Spark

Jahrelang kümmerte sich Pandas um tabellarische Verarbeitungen im Arbeitsspeicher, während Apache Spark für die verteilt ausgeführten Aufgaben zuständig war. Diese Aufteilung funktionierte einwandfrei, solange die Daten strukturiert waren.

Moderne Pipelines verarbeiten außerdem Bilder, Audiodaten und Videomaterial. In solchen Arbeitslasten kann die CPU-Decodierung die Ressourcenverfügbarkeit für GPU Inference stark einschränken, während die Garbage Collection der JVM sowie das Global Interpreter Lock in Python die Throughput beeinträchtigen. Neue Engine-Architekturen nutzen Rust und Apache Arrow, um diese Kosten zu senken.

Um die verschiedenen Optionen miteinander zu vergleichen, führte ich Benchmarks mit Polars, DataFusion, Daft, Ray Data sowie Spark anhand von zwei realen Datasets durch. Die Taxifahrten in New York stellen tabellarische Daten dar, während die Food-101-Bilder ein multimodales Pipeline darstellen.

Der Repository für die Demonstration des Engine-Vergleichs der Code ist enthalten. Führen Sie ihn auf Ihrer eigenen Hardware aus, da die Leistung des Engines von der Maschine, Dataset, sowie von der Arbeitslast abhängt.

TL;DR: Beginnen Sie mit der Struktur der zu bewältigenden Aufgabe und nicht mit dem Ziel, einen Benchmark zu erzielen. Polars eignet sich hervorragend als Standard für lokale DataFrame Pipelines; DataFusion ist für Teams geeignet, die einen Abfragespeicher benötigen Embedding; Ray Data sowie Daft sind auf gemischte CPU/GPU Pipelines-Szenarien ausgelegt; während Spark weiterhin eine etablierte Wahl für verteilte SQL-Anwendungen sowie ETL-Prozesse darstellt. Die in diesem Leitfaden angegebenen Geschwindigkeitsverbesserungen stellen spezifische Ergebnisse für bestimmte Arbeitslasten dar und sind keine allgemeingültigen Versprechen. Führen Sie vor der Entscheidungsfindung den dazugehörigen Benchmarks erneut mit Ihren eigenen Daten und Hardware aus.

Arten von Datenverarbeitungsarbeitslasten

Da sich die Engine-Architekturen spezialisieren, lautet die erste Frage, die beantwortet werden muss, welche Art von Arbeitslast man tatsächlich vorliegen hat. Die größte Unterscheidung besteht dabei zwischen strukturierten und multimodalen Aufgaben.

Strukturierte oder tabellarische Daten beinhalten Operationen wie Filtern, Aggregationen sowie Join-Vorgänge. Sie sind in der Regel durch CPU begrenzt und passen häufig in den Arbeitsspeicher. Laut Polars beträgt die Größe etwa 90 % der Abfragen verarbeiten weniger als 1 TB., Daher können viele solcher Agenten auf einem modernen Rechner statt auf einem Cluster ausgeführt werden.

Multimodale AI Systeme verarbeiten Bilder, Audiodaten oder Videomaterial. Inference kann auf einem GPU ausgeführt werden, wobei CPU bei der Dekodierung sowie bei Remote-Lesvorgängen oder Batching-Beschränkungen die Leistungsfähigkeit von Pipeline einschränkt. Das Verhältnis variiert je nach Instanz und verwendeten Operatoren; daher sollte die Auslastung entlang des gesamten Datenflusses gemessen werden, anstatt die Kapazität von GPU isoliert zu dimensionieren.

Die neueren Engine-Implementierungen richten sich auf unterschiedliche Bereiche dieses Anwendungsbereichs. Eine native Ausführung kann die Overhead-Kosten im Python-Umfeld verringern, arrow-kompatible Schnittstellen ermöglichen einen kostengünstigeren Datenaustausch, und eine Streaming-Ausführung kann den Speicherverbrauch bei Datasets größeren Mengen als dem verfügbaren RAM begrenzen. Keiner dieser Vorteile tritt automatisch für jeden Operator oder jede Konvertierung ein.

Teil 1: Einzelknotenverarbeitung

Pandas bevorzugt ein sofort ausführbares, im Arbeitsspeicher basierendes Programmieren Model. Viele gängige Operationen erzeugen Zwischenergebnisse, und die parallele Ausführung wird nicht über einen Abfragenoptimierer koordiniert. Dieser Kompromiss sorgt dafür, dass explorativer Code unkompliziert bleibt, kann jedoch bei größeren analytischen Durchsichten kostspielig werden. Bei Polars hingegen PDS-H Benchmark Beim Skalierungsfaktor 10 benötigte Pandas für den gesamten Datensatz etwa 365 Sekunden, während der Streaming-Engine von Polars lediglich 3,89 Sekunden zur Verfügung standen. Das Ergebnis von rund 94-mal unterscheidet sich je nach Benchmark und Konfiguration – es handelt sich dabei nicht um einen allgemeinen Umrechnungsfaktor von Pandas zu Polars.

Polars für lokale Tabellen Pipelines

Polars Es handelt sich dabei um einen praktischen Standardfall, wenn ein lokaler DataFrame Pipeline die Grenzen der sofortigen, einzeprozessigen Ausführung überschreitet. Seine verspätete API-Verarbeitung erstellt vor der Ausführung einen Abfragenplan, wodurch Optimierungen wie Predicate Pushdown sowie das Kürzen von Projektionen möglich werden. Der Engine kann anschließend Operatoren parallel ausführen und, sofern unterstützt, in Form von strömenden Batch-Verarbeitungen arbeiten.

Der Unterschied vergrößert sich mit zunehmenden Datenmengen. Bei Skalierungsfaktor 100 (~100 GB), Der Streaming-Engine von Polars benötigte 23,94 Sekunden, im Vergleich zu 152,27 Sekunden für seine eigene In-Memory-Engine – das entspricht einer Geschwindigkeitssteigerung von etwa 6-fach bei Datenmengen, die größer sind als der RAM-Bereich.

Von engine_comparison_examples.ipynb:

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 in der Ausführung Model. Bei der oben beschriebenen „lazy“ Abfrage kann Polars den Filter sowie die Auswahl der Spalten direkt in den Parquet-Scan einbeziehen. Dadurch können ganze Zeilengruppen übersprungen werden und es entfällt das Lesen ungenutzter Spalten. Ein typischer „eager“ Pandas Pipeline verfügt hingegen über keinen gesamtheitlichen Abfraplans zur Optimierung, obwohl eine sorgfältige Nutzung von Parquet-Filtern, ausgewählten Spalten sowie alternativen Backends-Ansätzen einen Teil dieses Nachteils ausgleichen kann.

DataFusion für eingebettete Abfragesysteme

Wo Polars eine Bibliothek ist, die Sie direkt verwenden. DataFusion ist der Engine, auf dem weitere Engines entwickelt werden. Er liefert die Leistung. InfluxDB 3.0, GreptimeDB, und Apples Comet Spark-Akcelerator.

In November 2024 Die von dem DataFusion-Projekt veröffentlichte ClickBench-Ausführung, DataFusion war führend bei den getesteten Einzelknoten-Parquet-Konfigurationen. Das Embucket-Team veröffentlichte später eine TPC-H-Skalenfaktor-1000-Ausführung auf einem einzigen großen Knoten. Diese Ergebnisse zeigen, welche Leistungssteigerung unter bestimmten Bedingungen erzielt werden kann; sie beweisen jedoch nicht, dass jede 1-TB-Arbeitslast unbedingt einen Cluster vermeiden muss.

Von engine_comparison_examples.ipynb:

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()

DataFusion’s Stärke liegt in seinem Modularität. Die Erweiterung APIs umfasst benutzerdefinierte Kataloge, Tabellenanbieter, Optimierungsregeln sowie Ausführungspläne – weshalb sie immer dann zum Einsatz kommt, wenn jemand eine eigene Datenplattform oder einen Embedding Abfragemotor für sein Produkt entwickelt.

Unbrauchbar für multimodale Daten

unvernünftig Es integriert Bilder, Audio, Video sowie Embeddings in den DataFrame Workflow. Es bietet native Ausdrücke für Operationen wie die Dekodierung von Bildern und das Herunterladen von Inhalten über URLs, wodurch herkömmliche Vorbereitungsverfahren keinen Schleifenaufruf in Python Zeile für Zeile erfordern. Genau diese Integration ist der Hauptgrund, warum man es gegenüber herkömmlichen tabellarischen Engines in Betracht zieht.

Von engine_comparison_examples.ipynb:

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()
)

Der API wird Ihnen vertraut vorkommen, wenn Sie mit Pandas vertraut sind, doch der zugrunde liegende Engine ist derselbe Rust + Arrow-Stack, wie er in Polars und DataFusion verwendet wird. Daft zeigt seine Vorteile insbesondere dann, wenn die „Zeilen“ Ihres DataFrames aus Bildern, PDFs oder Tensorn bestehen.

Einfachknoten Benchmark

Ich habe alle vier Engine-Versionen sowie den nativen Rust-Code über Polars-rs auf etwa 41 M ausgewertet. NYC-Gelbtaxifahrten für das gesamte Jahr 2024. Die vollständigen Ergebnisse finden Sie in der Demo-Repo:

Bei dieser Größe (etwa 660 MB im Parquet-Format) schließen alle vier neueren Engines ihre Verarbeitung schnell ab. Der oben angegebene Wert von 94x für die Konvertierung von Polars in Pandas stammt aus PDS-H; meine Testdaten mit 41 Millionen Zeilen ergaben einen geringeren Unterschied. Polars, DataFusion, Daft sowie Polars-rs lagen im gleichen groben Bereich, wobei DataFusion in diesem Testlauf die kürzeste Gesamtverarbeitungszeit aufwies. Dies ist ein nützliches Ergebnis für diese Abfrage – doch es handelt sich dabei nicht um eine zuverlässige Rangliste: API Entscheidend sollten hier vielmehr eine gute Passgenauigkeit sowie wiederholte Messungen an repräsentativen Datensätzen sein.

Teil 2: multimodale Daten

Tabellarisches ETL reduziert in der Regel die Datenmenge: Es werden Filter angewendet, Werte aggregiert und am Ende weniger Daten geschrieben als ursprünglich eingelesen. Multimodale AI Pipelines-Verarbeitung tut genau das Gegenteil. Ein einziger Dokumentpfad kann sich in Dutzende von Textteilen sowie Embedding-Vektoren aufteilen.

In einem herkömmlichen PySpark Pipeline werden Bilder und Audiodaten in der Regel als binäre Daten übergeben und in Python-Bibliotheken weitergeleitet, um dort decodiert oder verarbeitet zu werden. Der Wechsel zwischen der JVM und Python sowie die Serialisierung der Ergebnisse können einen erheblichen Anteil der Gesamtlaufzeit ausmachen. Spark bietet zwar Arrow-basierte Datenstrukturen, vektorisierte UDFs sowie Integrationen mit Beschleunigungsmechanismen an, doch der Einsatz dieser Optionen erfordert eine gezielte Konzeption und detaillierte Messungen.

Pipelining der Ausführung und Nutzung von GPU

Spark plant die Ausführung in Phasen, die durch Shuffle-Grenzen voneinander getrennt sind. Eine einfache Implementierung könnte das Herunterladen, Decode, sowie Inference in einer bestimmten Reihenfolge ausführen, wodurch entweder Kapazitäten für CPU oder GPU ungenutzt bleiben. Da Spark GPU-basierte Ressourcenplanung sowie Plugins unterstützt, ist eine inaktive Zeit nicht unvermeidlich; entscheidend ist vielmehr, dass Überschneidungen und Rückdrucke keine automatischen Eigenschaften eines gewöhnlichen PySpark-DataFrame-Jobs sind.

Pipelinierte Ausführung ist die Alternative. Anstatt die einzelnen Phasen nacheinander auszuführen, überlappt der Engine die I/O-Aufgaben, CPU sowie die GPU Inference, sodass zu jedem Zeitpunkt nur der langsame Bestandteil wartet.

Bildverarbeitung Benchmark

Ich habe den Bildverarbeitungsprozess anhand von 500 echten Fotografien getestet. Einführung in Lebensmittelwissenschaften Dataset. Ergebnisse aus dem Demo-Repository:

Polars und DataFusion fehlen, da dieser Test auf native Bildoperationen abzielt und nicht auf tabellarische Ausdrücke. Bei diesem Testlauf mit 500 Bildern war der Bildpfad von Daft 3,6-mal schneller als die Baseline aus Pandas und Pillow, während die Rust-Implementierung 4,2-mal schneller war. Das Beispiel eignet sich gut dazu, die Overhead-Kosten von Orchestration aufzuzeigen, ist aber allein zu klein, um die Leistung in der Produktion unter Throughput vorhersagen zu können.

Teil 3: verteilte Verarbeitung

Sobald eine einzelne Maschine nicht ausreicht, muss man entscheiden, wie die Arbeit aufgeteilt werden soll. Genau an diesem Punkt beginnen die architektonischen Unterschiede zwischen den Engines tatsächlich eine Rolle zu spielen.

Apache Spark

Spark stellt eine ausgereifte Alternative für umfangreiche tabellarbasierte ETL-Prozesse, join-Berechnungen mit hohem Shuffle-Aufwand sowie für Organisationen dar, die bereits auf seinem Ökosystem setzen. Seine recovery-Funktionen, die auf der Verfolgung von Änderungsverläufen beruhen, sowie die breite Plattformunterstützung sind besonders wertvoll, wenn Zuverlässigkeit und Vertrautheit mit den Arbeitsabläufen wichtiger sind als lokale Geschwindigkeit. Gemischte CPU/GPU Pipelines-Szenarien erfordern eine sorgfältigere Konfiguration der Ressourcen sowie eine gründlichere Pipeline-Planung im Vergleich zum SQL-Ansatz – in diesem Bereich bieten Ray Data und Daft eine gezieltere Abstraktion.

Von distributed_spark.ipynb:

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 Rechenumgebungen

Ray Data wurde um Workloads von AI herum konzipiert. Anstelle der Stage-Barrier von Spark verwendet es ein Streaming Model dies hält GPUs am Laufen. Die Funktion, die seine Notwendigkeit rechtfertigt, ist die gemischte Ressourcenplanung: Man kann angeben, dass ein Akteur „1 GPU, 4 CPUs” benötigt, während ein anderer ausschließlich CPUs verlangt, wobei Ray den Rest automatisch berechnet.

Amazon hat berichtet mehr als 120 Millionen US-Dollar an jährlichen Einsparungen Nach dem Umzug ausgewählter interner Datenverarbeitungslasten von Spark auf Ray weist der Bericht in dem Proof of Concept eine um 91 % bessere Kosteneffizienz sowie in der Produktion einen Zuwachs von 82 % pro GiB S3-Eingang auf. Das Ausmaß dieser Verbesserungen ist bemerkenswert, doch es handelt sich dabei um ein Fallstudie zu einer spezifischen Workload-Migration und nicht um die erwarteten Einsparungen bei einer typischen Einführung von Ray.

Von distributed_ray.ipynb:

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/")

Ein wichtiger Aspekt, den man kennen sollte: Je mehr CPUs man pro GPU hinzufügt, desto stärker skaliert Ray Data, während andere Engines an einem Plateau ankommen. In Anyscales Benchmarks, Durch den Wechsel von einem Verhältnis von 4:1 auf 32:1 CPU zu GPU wurde eine dreifache Beschleunigung bei der Verarbeitung von Bildern Inference erzielt, da die CPU-Feeder endlich mit der Geschwindigkeit der GPU mithalten konnten.

Verteilter Daft

Daft skaliert sich durch seine Flottillen-Engine: Jeder Knoten verfügt über einen einzigen Swordfish-Worker, wobei die darüber liegende Flotilla für die gesamtklusterweite Scheduling-Aufgabe zuständig ist. Swordfish kümmert sich um die lokale Ausführung in Rust sowie um die Pipelines-I/O-Operationen mithilfe von kleinen Batch-Streaming-Prozessen, sodass jeder Knoten stets beschäftigt bleibt und nicht auf die nächste Verarbeitungsstufe warten muss.

In Benchmarks veröffentlicht von Daft, Flotilla war in vier multimodalen Arbeitslasten 2–18-mal schneller als die getesteten Spark-Implementierungen. Der größte Unterschied zeigte sich im Fall der Video-Objekterkennung. Man sollte dieses Ergebnis als Beleg dafür betrachten, dass das Ausführungsdesign für diese Pipelines von entscheidender Bedeutung ist, und es anschließend mit den Ergebnissen des Wettbewerbers Anyscale sowie mit einem lokalen Test vergleichen.

Anyscale, das Unternehmen hinter Ray, veröffentlichte ein konkurrenzfähiger Benchmark in denen Ray Data nach der Anpassung die Lücke bei vielen hochwertigen CPU-Instanzen verringerte oder sogar schloss. Zusammen bilden diese beiden Herstellerstudien einen Grund, repräsentative Operatoren sowie die Verhältnisse von CPU zu GPU zu testen – und nicht, eine dauerhafte Rangliste abzuleiten.

Von distributed_daft.ipynb:

import daft
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").url.download())

# Define a GPU UDF for embedding generation
@daft.udf(return_dtype=daft.DataType.list(daft.DataType.float32()))
class TextEmbedder:
    def __init__(self):
        from sentence_transformers import SentenceTransformer
        self.model = SentenceTransformer("all-MiniLM-L6-v2", device="cuda")

    def __call__(self, text_col):
        texts = text_col.to_pylist()
        embeddings = self.model.encode(texts, batch_size=32)
        return [emb.tolist() for emb in embeddings]

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

Verteilte Vergleichbarkeit

Teil 4: Die Rolle von Rust und Arrow

Im Hintergrund dieser Konkurrenz ist die Konvergenz die interessantere Entwicklung. Polars, Daft und DataFusion nutzen Rust in großem Umfang, während Ray neben seinen Python-basierten Komponenten auch native Komponenten bereitstellt APIs. Alle können Daten über entsprechende Schnittstellen austauschen. Apache Arrow Ökosystem.

Der Arrow PyCapsule-Schnittstelle (__arrow_c_stream__ Das Protokoll bietet kompatiblen Bibliotheken eine standardisierte Methode zum Austausch von Arrow-Streams. Durch diese Übertragungsmethode wird eine serielle Verarbeitung Zeile für Zeile vermieden, und bei kompatiblen Schemata sowie Speicheraufteilungen können Buffers wiederverwendet werden. Dennoch können Vorgänge wie Materialisierung, Umgruppierung der Datenblöcke, Typumwandlung oder Übertragung auf andere Geräte weiterhin Datenkopien erzeugen; daher sollte die Effizienz der Datenübernahme durch Profilierung überprüft werden, anstatt davon auszugehen, dass sie kostenlos erfolgt.

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 Engine-Rahmenwerke Rust für die native Ausführung?

1. Kein Tracing-Garbage-Collector im Rust-Code. Durch das Konzept der Eigentumsverwaltung haben die nativen Operatoren eine direktere Kontrolle über die Zuweisung und Freigabe von Speicher. Dadurch können GC-bezogene Latency-Ereignisse reduziert werden, doch es wird weder der Speicherdruck noch Ausfälle aufgrund von zu wenig Speicher verhindert.

2. Kompakte native Repräsentationen. Rust-Strukturen enthalten keine Objekt-Header wie in Java. Der praktische Vorteil hängt von der Architektur des Engines ab: Säulensysteme der JVM vermeiden ebenfalls die Darstellung jedes Wertes als eigenständiges Objekt.

3. Stärkere Konkurrenzprüfungen. Rust mit seiner sicheren Architektur verhindert bereits zur Kompilierzeit viele Datenkonkurrenzen. Der Engine-Code kann weiterhin unsichere Abschnitte sowie logische Fehler im Umgang mit Konkurrenzsituationen enthalten, doch die Sprache verringert dadurch das Potenzial für Fehler.

In Kombination mit Arrows koloniellem Speicherlayout können diese Entscheidungen die Overhead-Kosten für Zuweisung und Serialisierung verringern. Sie beseitigen ihn jedoch nicht an jeder Grenze – Python-Objekte, der Netzwerktransport, inkompatible Schemata sowie Bewegungen der Geräte spielen weiterhin eine Rolle.

Teil 5: Einsatz mehrerer Engine‑Architekturen

Es ist nicht notwendig, die gesamte Plattform durch einen einzigen Engine zu verarbeiten. Die Komposition ist sinnvoll, wenn die Gestaltung der Grenzen kostengünstiger ist als die Entwicklung eines Systems, das alle Arbeitslasten selbst bewältigen muss.

Eine mögliche Architektur nutzt Spark für effiziente Lakehouse-Vergleiche, schreibt die Ergebnisse in ein offenes Tabellen- oder Dateiformat und übergeben diese anschließend an Ray Data oder Daft, um GPU Inference durchzuführen oder multimodale Transformationen vorzunehmen. Polars oder DuckDB ermöglichen hingegen lokale Analysen direkt anhand derselben Dateien. Ein kleineres Team benötigt möglicherweise nur einen dieser Engines; ein weiterer wird hinzugefügt, sobald ein messbarer Bottleneck die Notwendigkeit für eine solche Erweiterung rechtfertigt.

Was diese Architektur überhaupt erst möglich macht, sind offene Formate wie Parquet, Delta, Iceberg und Arrow. Jeder Komponenten in der Pipeline kann diese Formate nativ lesen und schreiben, wodurch der Datenaustausch zwischen den verschiedenen Schritten lediglich über einen Dateipfad erfolgt.

Auswahl des Engines nach Bewertungskriterien

Diagonale Skalierung

Lange Zeit bestand die Wahl zwischen Skalierung nach oben – also dem Einsatz eines größeren Rechners – oder Skalierung nach außen – durch den Einsatz mehrerer Rechner. Polars Cloud Es wird dabei eine Art Zwischenschritt durchgeführt, der als diagonal scaling bezeichnet wird: Zunächst erfolgt eine horizontale Skalierung beim Lesen aus dem Cloud-Speicher, um die I/O-Leistung voll auszunutzen. Anschließend wird, sobald Filter und Aggregationen die Datenmenge reduziert haben, auf einen einzigen großen Knoten umgestellt, wodurch das gesamte Verfahren des verteilten Shufflings weggelassen wird.

Die wichtigere Erkenntnis besteht darin, den Kosten pro erfolgreich abgeschlossenem Job zu vergleichen und nicht den Preis pro Stundenzugriff. Ein größeres Knoten kann günstiger sein, wenn die durch Runtime erzielte Kostensenkung die höhere Rechenrate ausgleicht – doch das Endergebnis hängt von I/O-Auslastung, Speicherdruck sowie von der Menge an Arbeit ab, die bereits vor dem Shuffle eliminiert werden kann.

Die Vergleichsanalyse reproduzieren

Der Repositorium für die Begleit-Demo Es enthält die Skripte, Notebooks sowie die Umgebung, die für die Vergleiche verwendet werden. Der untenstehende Befehl nutzt einen Monat an Taxidaten; folgen Sie den Anweisungen im Repository zu Dataset, um den zuvor gezeigten, umfassenderen Jahresverlauf nachzubilden.

# Install dependencies
uv sync

# Run the tabular benchmark (~2.9M NYC taxi trips)
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 ..

In diesem Repository finden sich außerdem Notebooks zur parallelen API Vergleich von Modellen sowie Docker Compose-Konfigurationen für lokale, verteilte Ausführungen von Spark, Ray und Daft.

Wichtige Erkenntnisse

1. Beweisen Sie, dass eine einzige Maschine unzureichend ist. Die Ergebnisse der Skalierungsstudien zeigen, dass einige umfangreiche Analysevorgänge auf einem einzelnen Knoten abgebildet werden können. Messen Sie vor der Annahme der zusätzlichen Overhead-Kosten eines Clusters den Speicherbedarf, die I/O-Leistung sowie die Anforderungen hinsichtlich des Speicherausflusses und der Wiederherstellung.

2. Die Ausführung Models ist wichtiger als vertraute Syntax. Träge Planning, Pushdown-, Streaming- und parallele Operatoren erklären einen großen Teil des Unterschieds zwischen einem eager-Modell basierenden DataFrame-Skript und einem analytischen Engine.

3. Messen Sie den gesamten Beschleuniger Pipeline. Decode, Netzwerk-E/A, Batching, Serialisierung sowie Backpressure bestimmen, ob ein GPU ständig in Betrieb bleibt. Ray Data und Daft machen diese Zusammenhänge zu einer zentralen Abstraktion; Spark kann dies durch zusätzliche Konzepte und Werkzeuge unterstützen.

4. Wählen Sie zunächst auf Basis der Arbeitslast aus und anschließend Benchmark. Nutzen Sie die Passung zum bestehenden Ökosystem, um das Auswahlfeld einzugrenzen, sowie eine repräsentative End-to-End-Aufgabe zur Entscheidungsfindung. Die Ergebnisse der Bewertung von Anbietern mittels Benchmark sind lediglich Hypothesen und keine Garantien.

5. Offene Formate ermöglichen eine rückgängige Auswahl. arrow-kompatible Schnittstellen, Parquet sowie offene Tabellenformate können die Kosten für den Datentransfer verringern. Prüfen Sie, ob bei einer bestimmten Konvertierung Buffers wiederverwendet werden oder kopiert werden.

Falls Sie einen Ausgangspunkt benötigen, sollten Sie für lokale tabellare Pipeline-Strukturen Polars in Betracht ziehen, während Daft oder Ray Data geeignet sind, um gemischte CPU/GPU Pipeline-Szenarien zu bewältigen. Wählen Sie Spark dann, wenn seine verteilte Ausführung, das damit verbundene Ökosystem oder bereits vorhandene Infrastrukturen eine bestimmte Anforderung erfüllen – und nicht nur dann, wenn die Dataset einen bestimmten Größenrahmen überschreitet.

Referenzen

Benchmarks sowie Leistungsdaten

• Polars PDS-H Benchmarks - Polars Streaming im Vergleich zu Pandas bei SF-10 (94-fach) sowie bei SF-100 (6,4-fach schneller als im In-Memory-Modus) DataFusion ClickBench-Ergebnisse (November 2024) - Von dem Projekt veröffentlichte Ergebnisse im Parquet-Format für einzelne Knoten
• Embucket: TPC-H SF-1000 bei DataFusion - Vollständige Ausführung von TPC-H mit etwa 1 TB auf einem einzigen Node Migration von Amazon Spark zu Ray - Einsparungen von 120 Mio. US-Dollar pro Jahr, 82 % Effizienz bei den Produktionskosten Die dumme Flotte Benchmarks - 2–18-mal schneller als Spark bei multimodalen Arbeitslasten
• Anyscale Ray im Vergleich zu Daft Benchmarks - Konkurernder Anbieter Benchmark für multimodale Arbeitslasten

Engine und Frameworks

• Polars - Rust-DataFrame-Bibliothek mit verzögerter Ausführung
• Apache DataFusion - Einbaufähiger Rust-AbfragemotorGitHub)
• unvernünftig - Multimodaler, natives verteilter DataFrame
• Ray - Verteilter Rechenverkehr Framework für AI (Dateninterne Strukturen)
• Apache Spark - Verteilter ETL- und SQL-Engine
• DuckDB - Analytischer SQL-Engine im Prozessinneren Dask - Parallelrechnung, die direkt in Python implementiert ist

Architektur und Ökosystem

• Polars Cloud: Diagonale Skalierung - Dynamische vertikale/horizontale Skalierung Architektur der Daft Flotilla - Schwertfisch + Flottillen-basierte verteilte Engine
• Apache DataFusion Comet - Der Spark-Akcelerator, ursprünglich bei Apple entwickelt.
• Apache Arrow - Säulenbasiertes In-Memory-FormatFlug-RPC, PyCapsule-Schnittstelle)
• InfluxDB 3.0 + DataFusion - Zeitreihendatenbank, die auf DataFusion basiert GreptimeDB - Observability-Datenbank mit DataFusion

Demo-Repositorium

• Engine-Vergleichs-Demo - Begleitcode zu diesem Artikel: Benchmarks, Notebooks sowie Docker Compose für Spark/Ray/Daft

Datasets

• Protokolle von NYC-Taxifahrten - NYC TLC-Daten zu gelben Taxis (Parquet, ~2,9 M Zeilen pro Monat) Einführung in Lebensmittelwissenschaften Dataset - 101.000 Nahrungsbildern von der ETH Zürich (Bossard et al., ECCV 2014)