Silniki przetwarzania danych: Polars, DataFusion, Ray i Spark

Viacheslav Dubrov

2026-02-21 · Zaktualizowano 2026-07-15 · 14 min · Infrastruktura AI

Tłumaczenie automatyczne Ten artykuł został automatycznie przetłumaczony z angielskiego oryginału.

Przez lata Pandas obsługiwał pracę z tabelarycznymi danymi w pamięci, a Apache Spark — jej rozproszony wariant. Ten podział sprawdzał się, gdy dane były ustrukturyzowane.

Nowoczesne pipeline’y przetwarzają również obrazy, dźwięk i wideo. W takich obciążeniach dekodowanie na CPU może zagłodzić inferencję na GPU, a garbage collection w JVM i Global Interpreter Lock Pythona ograniczają przepustowość. Nowe silniki wykorzystują Rust i Apache Arrow, aby zmniejszyć te koszty.

Aby porównać dostępne opcje, wykonałem benchmarki Pandas, Polars, DataFusion, Daft oraz natywnego Rust na dwóch rzeczywistych zbiorach danych. Spark i Ray omawiam w osobnych notebookach dotyczących przetwarzania rozproszonego. Przejazdy taksówek w Nowym Jorku reprezentują pracę tabelaryczną, a obrazy Food-101 — pipeline multimodalny.

Repozytorium engine-comparison-demo zawiera kod. Uruchom go na własnym sprzęcie, ponieważ wydajność silnika zależy od maszyny, zbioru danych i obciążenia.

Typy obciążeń związanych z przetwarzaniem danych

Silniki specjalizują się w różnych zadaniach, dlatego pierwszym pytaniem powinno być: jaki rodzaj obciążenia faktycznie masz? Najważniejszy podział przebiega między danymi ustrukturyzowanymi a multimodalnymi.

Dane ustrukturyzowane lub tabelaryczne obejmują filtrowanie, agregacje i złączenia. Zwykle obciążają CPU i często mieszczą się w pamięci. Wiele takich zapytań można uruchomić na jednej nowoczesnej maszynie, ale o tym, czy potrzebny jest klaster, nadal decydują pamięć, I/O, współbieżność, odzyskiwanie po awariach i charakter obciążenia.

Multimodalne AI przetwarza obrazy, dźwięk lub wideo. Inferencja może działać na GPU, podczas gdy dekodowanie na CPU, odczyty zdalne albo batching ograniczają pipeline. To, który etap ogranicza przepustowość, zmienia się zależnie od typu instancji i operatorów, dlatego należy mierzyć wykorzystanie zasobów w całej ścieżce, zamiast dobierać GPU w izolacji.

Nowsze silniki celują w różne części tej przestrzeni. Natywne wykonanie może ograniczyć narzut Pythona, interfejsy zgodne z Arrow mogą obniżyć koszt wymiany danych, a wykonanie strumieniowe może ograniczyć zużycie pamięci przy zbiorach większych niż RAM. Żadna z tych korzyści nie jest automatyczna dla każdego operatora ani każdej konwersji.

Część 1: przetwarzanie na pojedynczym węźle

Pandas preferuje zachłanny model programowania działający w pamięci. Wiele typowych operacji tworzy dane pośrednie, a wykonanie równoległe nie jest koordynowane przez optymalizator zapytań. Taki kompromis upraszcza kod eksploracyjny, ale przy większych skanach analitycznych może stać się kosztowny. W benchmarku PDS-H Polars przy współczynniku skali 10 Pandas potrzebował około 365 sekund na cały zestaw, podczas gdy silnik strumieniowy Polars — 3,89 sekundy. Wynik rzędu 94x opisuje ten konkretny benchmark i konfigurację; nie jest ogólnym współczynnikiem konwersji z Pandas do Polars.

Polars dla lokalnych pipeline’ów tabelarycznych

Polars jest praktycznym wyborem domyślnym, gdy lokalny pipeline DataFrame wyrósł z zachłannego wykonania w jednym procesie. Jego leniwe API buduje plan zapytania przed wykonaniem, umożliwiając optymalizacje takie jak predicate pushdown i projection pruning. Silnik może następnie uruchamiać operatory równolegle oraz — jeśli jest to obsługiwane — w strumieniowych batchach.

Różnica rośnie wraz z rozmiarem danych. Przy współczynniku skali 100 (~100 GB) silnik strumieniowy Polars zakończył pracę w 23,94 sekundy, a jego własny silnik działający w pamięci potrzebował 152,27 sekundy — około 6x szybciej, mimo że dane były większe niż RAM.

Poniżej znajduje się ilustracyjny szkic API, a nie dosłowny fragment bieżących wersji engine_comparison_examples.ipynb ani 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()

Najważniejsza różnica dotyczy modelu wykonania. W powyższym leniwym zapytaniu Polars może przenieść filtr i wybór kolumn bezpośrednio do skanu Parquet. Może to pozwolić pominąć grupy wierszy i uniknąć odczytu nieużywanych kolumn. Typowy pipeline Pandas działający zachłannie nie ma globalnego planu zapytania do optymalizacji, choć ostrożne użycie filtrów Parquet, wybranych kolumn i alternatywnych backendów może odtworzyć część tej różnicy.

DataFusion jako osadzany silnik zapytań

Podczas gdy Polars jest biblioteką używaną bezpośrednio, DataFusion to silnik, na którym buduje się inne silniki. Zasila InfluxDB 3.0, GreptimeDB oraz opracowany przez Apple akcelerator Comet Spark.

W opublikowanym przez projekt DataFusion w listopadzie 2024 roku teście ClickBench DataFusion osiągnął najlepsze wyniki w testowanych konfiguracjach Parquet na pojedynczym węźle. Zespół Embucket opublikował później studium przypadku dostawcy dla TPC-H o współczynniku skali 1000, używając datafusion-cli na r6gd.metal z 64 vCPU, 512 GB RAM, lokalnym NVMe i pomiarami na rozgrzanym cache. Q18 i Q21 zakończyły się niepowodzeniem lub działały nadmiernie długo w pierwotnym uruchomieniu, a następnie zostały strukturalnie przepisane na potrzeby ukończonych wyników. Pokazuje to skalowanie pionowe na jednym węźle w tej konkretnej konfiguracji, a nie ogólny wynik TPC-H.

Poniżej znajduje się ilustracyjny szkic API, a nie dosłowny fragment bieżących wersji engine_comparison_examples.ipynb ani 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()

Siłą DataFusion jest jego modułowość. API rozszerzeń obejmuje niestandardowe katalogi, dostawców tabel, reguły optymalizatora i plany wykonania. Dlatego DataFusion pojawia się wszędzie tam, gdzie ktoś buduje własną platformę danych albo osadza silnik zapytań we własnym produkcie.

Daft dla danych multimodalnych

Daft włącza obrazy, dźwięk, wideo i embeddings do przepływu pracy z DataFrame. Udostępnia natywne wyrażenia dla operacji takich jak dekodowanie obrazów i pobieranie z URL, dzięki czemu typowe ścieżki preprocessing nie wymagają pętli Pythona wykonywanej wiersz po wierszu. Ta integracja jest głównym powodem, by wybrać Daft zamiast ogólnego silnika tabelarycznego.

Poniżej znajduje się ilustracyjny szkic API, a nie dosłowny fragment bieżących wersji engine_comparison_examples.ipynb ani 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()
)

Jeśli znasz Pandas, API wyda Ci się znajome, ale silnik korzysta z tego samego stosu Rust + Arrow co Polars i DataFusion. Daft przynosi największe korzyści wtedy, gdy „wierszami” w DataFrame są obrazy, PDF-y lub tensory.

Benchmark na pojedynczym węźle

Uruchomiłem Pandas, Polars, DataFusion i Daft, a także natywny Rust za pośrednictwem Polars-rs, na około 41 mln przejazdów żółtych taksówek w Nowym Jorku z całego 2024 roku. Spark i Ray pojawiają się w przykładach rozproszonych, a nie w tym benchmarku jednowęzłowym. Pełne wyniki znajdują się w repozytorium demo:

Wykres stanowi kontekst dla benchmarku towarzyszącego, a nie uczciwy ranking silników porównanych 1:1: Pandas ładuje dane przejazdów i stref przed mierzoną funkcją, podczas gdy Polars, DataFusion, Daft i Rust obejmują w sekcji pomiarowej różne operacje odczytu plików, rejestracji lub konfiguracji zapytania. Skrypt powtarza każdą operację trzy razy i raportuje medianę po załadowaniu importów; stan cache systemu plików, rozgrzewka, CPU i inne warunki uruchomienia mogą zmienić wartości. Przed uznaniem kolejności za wynik wydajności ujednolić granicę pomiaru i ponownie uruchomić testy w kontrolowanych warunkach.

Przy tym rozmiarze (~660 MB Parquet) trzy nowsze silniki Pythona kończą pracę szybko. Podana wyżej wartość 94x dla Polars względem Pandas pochodzi z PDS-H; moje obciążenie taksówkowe obejmujące 41 mln wierszy dało mniejszą różnicę. W tym pomiarze o mieszanym zakresie Polars, DataFusion, Daft i Polars-rs znalazły się w podobnym przedziale, a DataFusion odnotował najniższy raportowany czas. Ta kolejność dotyczy tego zapytania i tej granicy pomiaru, a nie silników jako takich. Gdy wyniki są tak zbliżone, decyzję powinny determinować dopasowanie API i powtarzane pomiary na reprezentatywnych danych.

Część 2: dane multimodalne

Tabelaryczny ETL zwykle zmniejsza ilość danych: filtrujesz, agregujesz i zapisujesz mniej, niż odczytujesz. Pipeline’y multimodalnego AI działają odwrotnie. Pojedyncza ścieżka dokumentu może rozgałęzić się na dziesiątki fragmentów tekstu i wektorów embedding.

W konwencjonalnym pipeline’ie PySpark obrazy i dźwięk często trafiają jako dane binarne do bibliotek Pythona w celu dekodowania lub transformacji. Przekraczanie granicy JVM/Python i serializacja wyników może stanowić dużą część czasu zegarowego. Spark obsługuje ścieżki oparte na Arrow, wektoryzowane UDF-y i integracje z akceleratorami, ale rozwiązania te wymagają jawnego zaprojektowania i pomiarów.

Wykonanie potokowe i wykorzystanie GPU

Spark planuje pracę w etapach rozdzielonych granicami shuffle. Prosta implementacja może umieścić pobieranie, dekodowanie i inferencję w sekwencji, w której CPU albo GPU pozostaje bezczynne. Spark obsługuje planowanie zasobów GPU i pluginy, więc bezczynność nie jest nieunikniona. Jednak nakładanie się operacji i backpressure nie są automatycznymi właściwościami zwykłego zadania PySpark DataFrame.

Wykonanie potokowe stanowi alternatywę. Zamiast uruchamiać etapy jeden po drugim, silnik nakłada na siebie I/O, pracę CPU i inferencję na GPU. Gdy etapy i bufory są zbalansowane, takie nakładanie może ograniczyć czas bezczynności i zwiększyć przepustowość; uruchamianie i opróżnianie pipeline’u, backpressure albo wolniejszy etap nadal mogą powodować oczekiwanie zasobów.

Benchmark przetwarzania obrazów

Wykonałem benchmark przetwarzania obrazów na 500 rzeczywistych fotografiach ze zbioru Food-101. Wyniki z repozytorium demo:

Polars i DataFusion nie występują w tym zestawieniu, ponieważ test dotyczy natywnych operacji na obrazach, a nie wyrażeń tabelarycznych. W tym uruchomieniu na 500 obrazach ścieżka obrazowa Daft była około 3,7x szybsza niż bazowa implementacja Pandas + Pillow. Dokładna wartość multimodal_results.json równa 1.0148517909983639s podzielona przez wartość rust_multimodal_results.json równą 0.235865209s daje 4.30267692, więc implementacja Rust była 4,3x szybsza. Towarzyszący README pokazuje zaokrąglone wartości 1.01s i 0.24s, których stosunek wynosi 4,2x; jest to efekt zaokrąglenia wyświetlanych wartości, podczas gdy wartości JSON są źródłem prawdy. Próbka dobrze ujawnia narzut orkiestracji, ale sama jest zbyt mała, by przewidywać produkcyjną przepustowość.

Część 3: przetwarzanie rozproszone

Gdy jedna maszyna przestaje wystarczać, trzeba zdecydować, jak rozłożyć pracę.

Apache Spark

Spark to dojrzały wybór dla dużych tabelarycznych pipeline’ów ETL, złączeń intensywnie wykorzystujących shuffle oraz organizacji, które już operują w jego ekosystemie. Odzyskiwanie na podstawie lineage i szerokie wsparcie platformowe mają znaczenie, gdy niezawodność i znajomość operacyjna są ważniejsze niż lokalna szybkość. Pipeline’y mieszane CPU/GPU wymagają bardziej świadomej konfiguracji zasobów i projektu niż ścieżka SQL — właśnie tutaj Ray Data i Daft oferują bardziej wyspecjalizowaną abstrakcję.

Poniżej znajduje się ilustracyjny szkic API Spark, a nie dosłowny fragment bieżących wersji distributed_spark.ipynb ani 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 dla heterogenicznych zasobów obliczeniowych

Ray Data został zaprojektowany z myślą o obciążeniach AI. Zamiast barier między etapami, znanych ze Sparka, wykorzystuje model strumieniowy, który utrzymuje GPU w stanie ciągłej pracy. Najważniejszą funkcją dla pipeline’ów AI jest planowanie mieszanych zasobów: możesz zadeklarować, że jeden aktor potrzebuje „1 GPU, 4 CPU”, a inny tylko CPU, po czym Ray zajmie się resztą.

Amazon poinformował o ponad $120 mln rocznych oszczędności po przeniesieniu wybranych wewnętrznych obciążeń przetwarzania danych ze Sparka do Raya. Raport wskazuje na o 91% lepszą efektywność kosztową w proof of concept i o 82% w produkcji, w przeliczeniu na GiB danych wejściowych z S3. Skala jest znacząca, ale jest to studium przypadku migracji dla określonych obciążeń, a nie oczekiwany poziom oszczędności przy typowym wdrożeniu Raya.

Poniżej znajduje się ilustracyjny szkic API Ray Data, a nie dosłowny fragment bieżących wersji distributed_ray.ipynb ani 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/")

Warto zwrócić uwagę na proporcję CPU do GPU. W benchmarkach Anyscale Ray Data nadal skalował się wraz ze wzrostem liczby CPU przypadających na GPU, podczas gdy inne testowane silniki osiągały plateau. Źródło informuje o przyspieszeniu inferencji obrazów sięgającym 3x w miarę ograniczania zagłodzenia CPU, ponieważ dostarczające dane CPU mogą nadążyć za GPU.

Rozproszony Daft

Daft skaluje się poziomo za pośrednictwem swojego silnika Flotilla: na każdym węźle działa jeden worker Swordfish, a Flotilla zajmuje się planowaniem w całym klastrze. Swordfish obsługuje lokalne wykonanie w Rust i potokuje I/O z obliczeniami za pomocą strumieniowania małych batchy, dzięki czemu każdy węzeł pozostaje zajęty i nie czeka na kolejny etap.

W benchmarkach opublikowanych przez Daft Flotilla działał 4–18x szybciej niż testowane implementacje Sparka w czterech obciążeniach multimodalnych. Największą różnicę odnotowano w przypadku wykrywania obiektów w wideo. Potraktuj ten wynik jako dowód, że projekt wykonania ma znaczenie dla takich pipeline’ów, a następnie porównaj go z konkurencyjnymi wynikami Anyscale poniżej oraz z testem lokalnym.

Anyscale, firma stojąca za Ray, opublikowała konkurencyjny benchmark, w którym Ray Data zmniejszył lub odwrócił różnicę na instancjach z dużą liczbą CPU po dostrojeniu. Oba badania dostawców są powodem, by testować reprezentatywne operatory i proporcje CPU do GPU, a nie wyciągać wnioski o trwałym rankingu.

Poniżej znajduje się ilustracyjny szkic API Daft, a nie dosłowny fragment bieżących wersji distributed_daft.ipynb ani 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/")

Porównanie rozproszone

Część 4: rola Rust i Arrow

Pod powierzchnią tej rywalizacji ciekawsza jest konwergencja. Polars, Daft i DataFusion szeroko wykorzystują Rust, a Ray zawiera komponenty natywne obok swoich API Pythona. Wszystkie mogą wymieniać dane za pośrednictwem części ekosystemu Apache Arrow.

Interfejs Arrow PyCapsule (protokół __arrow_c_stream__) zapewnia zgodnym bibliotekom standardowy sposób wymiany strumieni Arrow. Transfer może uniknąć serializacji wiersz po wierszu i ponownie wykorzystać bufory, jeśli schematy oraz układy pamięci są zgodne. Materializacja, zmiana chunków, konwersja typów lub transfer między urządzeniami nadal mogą powodować kopiowanie danych, dlatego należy zweryfikować przekazanie za pomocą profilowania, zamiast zakładać, że jest bezkosztowe.

Poniżej znajduje się ilustracyjny szkic API, a nie samodzielny skrypt: zakłada on, że events.parquet już istnieje. Towarzyszący engine_comparison_examples.ipynb tworzy .data/events.parquet w komórkach konfiguracyjnych przed przykładem Arrow.

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

Dlaczego te nowsze silniki wykorzystują Rust do natywnego wykonania?

1. Brak śledzącego garbage collectora w kodzie Rust. Ownership zapewnia natywnym operatorom bardziej bezpośrednią kontrolę nad alokacją i zwalnianiem pamięci. Może to ograniczyć opóźnienia związane z GC, ale nie zapobiega presji na pamięć ani błędom out-of-memory.

2. Kompaktowe natywne reprezentacje. Struktury Rust nie zawierają nagłówków obiektów Javy. Praktyczna korzyść zależy od układu używanego przez silnik: kolumnowe systemy JVM również unikają reprezentowania każdej wartości jako osobnego obiektu.

3. Silniejsze kontrole współbieżności. Reguły bezpiecznego Rust eliminują wiele race condition na etapie kompilacji. Kod silnika nadal może zawierać bloki unsafe i logiczne błędy współbieżności, ale język ogranicza obszar potencjalnych awarii.

W połączeniu z kolumnowym układem pamięci Arrow decyzje te mogą ograniczyć narzut alokacji i serializacji. Nie eliminują go jednak na każdej granicy; nadal znaczenie mają obiekty Pythona, transport sieciowy, niezgodne schematy i przemieszczanie danych między urządzeniami.

Część 5: korzystanie z wielu silników

Nie musisz przepuszczać całej platformy przez jeden silnik. Kompozycja jest użyteczna, gdy przekazanie danych między silnikami kosztuje mniej niż zmuszenie jednego systemu do obsługi każdego obciążenia.

Jedna z możliwych architektur wykorzystuje Spark do istniejących złączeń lakehouse, zapisuje otwarty format tabeli lub pliku, a następnie przekazuje wynik do Ray Data albo Daft w celu wykonania inferencji na GPU lub transformacji multimodalnych. Polars albo DuckDB mogą służyć do lokalnej analizy tych samych plików. Mniejszy zespół może potrzebować tylko jednego z tych silników; kolejny warto dodać dopiero wtedy, gdy zmierzony bottleneck uzasadnia granicę operacyjną.

Tę kompozycję umożliwia interoperacyjność. W tych przykładach Parquet jest wspólnym formatem przekazywania plików, a Arrow może zapewniać wymianę danych w pamięci. Obsługa Delta i Iceberg zależy od silnika i konektora, dlatego należy zweryfikować każdą granicę. Przekazanie pliku również wiąże się z kosztami I/O i konwersji; nie jest automatycznie bezkosztowe.

Wybór silnika według przypadku ewaluacji

Skalowanie diagonalne

Przez długi czas wybór sprowadzał się do skalowania pionowego (większa maszyna) albo poziomego (więcej maszyn). Polars Cloud robi coś pomiędzy, określając to jako skalowanie diagonalne. Skaluje poziomo podczas odczytu z chmurowego storage’u, aby maksymalnie wykorzystać I/O, a następnie przechodzi do jednego dużego węzła, gdy filtry i agregacje zmniejszą ilość danych. Rozproszony shuffle nigdy się nie odbywa.

Szersza lekcja brzmi: porównuj koszt zakończonego sukcesem zadania, a nie godzinową cenę instancji. Większy węzeł może być tańszy, gdy skrócenie czasu działania przewyższa wyższą stawkę, ale wynik zależy od I/O, presji na pamięć i ilości pracy wyeliminowanej przed shuffle.

Odtwarzanie porównania

Towarzyszące repozytorium demo zawiera skrypty, notebooki i środowisko użyte do porównań. Poniższa szybka komenda korzysta z domyślnego zbioru przejazdów taksówek z całego 2024 roku, zgodnego z większym uruchomieniem pokazanym wcześniej.

# 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 ..

Repozytorium zawiera również notebooki z porównaniami API side-by-side oraz konfiguracje Docker Compose do lokalnych uruchomień rozproszonych Spark, Ray i Daft.

Najważniejsze wnioski

1. Udowodnij, że jedna maszyna nie wystarcza. Wyniki skalowania pionowego pokazują, że część dużych skanów analitycznych mieści się na pojedynczym węźle. Przed zaakceptowaniem narzutu klastra zmierz wymagania dotyczące pamięci, I/O, spill i odzyskiwania po awariach.

2. Modele wykonania są ważniejsze niż znajoma składnia. Leniwe planowanie, pushdown, strumieniowanie i równoległe operatory wyjaśniają dużą część różnicy między zachłannym skryptem DataFrame a silnikiem analitycznym.

3. Mierz cały pipeline akceleratora. Dekodowanie, sieciowe I/O, batching, serializacja i backpressure decydują o tym, czy GPU pozostaje zajęte. Ray Data i Daft traktują takie nakładanie operacji jako centralną abstrakcję; Spark może je obsłużyć dzięki dodatkowemu projektowaniu i narzędziom.

4. Twórz shortlistę według obciążenia, a następnie wykonuj benchmarki. Dopasowanie do ekosystemu wykorzystaj do zawężenia wyboru, a reprezentatywne zadanie end-to-end — do podjęcia decyzji. Wyniki benchmarków dostawców są hipotezami, nie gwarancjami.

5. Otwarte formaty ułatwiają odwracalność wyboru. Interfejsy zgodne z Arrow, Parquet i otwarte formaty tabel mogą ograniczyć koszty przekazywania danych. Sprawdź, czy konkretna konwersja ponownie wykorzystuje bufory, czy je kopiuje.

Jeśli potrzebujesz punktu wyjścia, wypróbuj Polars dla lokalnego pipeline’u tabelarycznego oraz Daft lub Ray Data dla pipeline’u mieszanego CPU/GPU. Wybierz Spark, gdy jego wykonanie rozproszone, ekosystem lub istniejąca baza operacyjna spełnia konkretne wymaganie. Sam rozmiar zbioru danych jest słabym powodem.

Bibliografia

Benchmarki i dane dotyczące wydajności

• Benchmarki Polars PDS-H — strumieniowanie Polars względem Pandas przy SF-10 (94x) i SF-100 (6,4x względem wykonania w pamięci)
• Wyniki DataFusion ClickBench (listopad 2024) — opublikowane przez projekt wyniki na pojedynczym węźle dla Parquet
• Embucket: TPC-H SF-1000 na DataFusion — studium przypadku dostawcy dotyczące r6gd.metal z pomiarami na rozgrzanym cache; Q18 i Q21 przepisano po tym, jak pierwotne uruchomienie zakończyło się niepowodzeniem lub trwało nadmiernie długo
• Migracja Amazon ze Spark do Ray — $120 mln rocznych oszczędności, 82% efektywności kosztowej w produkcji
• Benchmarki Daft Flotilla — 4–18x szybciej niż Spark w obciążeniach multimodalnych
• Benchmarki Anyscale Ray względem Daft — konkurencyjny benchmark dostawcy dla obciążeń multimodalnych

Silniki i frameworki

• Polars — biblioteka DataFrame w Rust z leniwym wykonaniem
• Apache DataFusion — osadzany silnik zapytań w Rust (GitHub)
• Daft — rozproszony DataFrame natywny dla danych multimodalnych
• Ray — framework obliczeń rozproszonych dla AI (Data Internals)
• Apache Spark — rozproszony silnik ETL i SQL
• DuckDB — analityczny silnik SQL działający w procesie
• Dask — równoległe obliczenia natywne dla Pythona

Architektura i ekosystem

• Polars Cloud: skalowanie diagonalne — dynamiczne skalowanie pionowe i poziome
• Architektura Daft Flotilla — rozproszony silnik Swordfish + Flotilla
• Apache DataFusion Comet — akcelerator Spark pierwotnie opracowany przez Apple
• Apache Arrow — kolumnowy format danych w pamięci (Flight RPC, interfejs PyCapsule)
• InfluxDB 3.0 + DataFusion — baza danych szeregów czasowych zbudowana na DataFusion
• GreptimeDB — baza danych do observability wykorzystująca DataFusion

Repozytorium demo

• engine-comparison-demo — kod towarzyszący artykułowi: benchmarki, notebooki i Docker Compose dla Spark/Ray/Daft

Zbiory danych

• NYC Taxi Trip Records — dane o żółtych taksówkach NYC TLC (Parquet, ~2,9 mln wierszy miesięcznie)
• Food-101 Dataset — 101 tys. obrazów jedzenia z ETH Zurich (Bossard i in., ECCV 2014)