Ray vs Dask vs Spark: Choosing a Distributed Compute Engine for Python-First Teams

ray vs dask vs spark is the pick-one architectural decision that quietly sets the ceiling on how fast a Python-first data team can ship — because the distributed compute engine you standardise on decides whether a data scientist can scale a notebook to a cluster in an afternoon, whether your ETL bill is dominated by a JVM you can't profile, and whether batch inference over a hundred million rows is a one-liner or a two-week platform project. Every workload your team runs — a nightly Parquet aggregation, a feature-engineering job that has outgrown a single machine's RAM, a GPU batch-scoring pass over an embeddings table — has to run somewhere, and the three engines that dominate that "somewhere" in 2026 were each born to solve a different problem. Picking well is not about which project has the most GitHub stars; it is about matching the engine's mental model to the shape of the work.

This guide is the walkthrough you wished existed the first time an interviewer asked "compare Spark, Dask, and Ray and tell me when each one wins," or "your data scientists live in pandas and just blew past a machine's memory — Spark or Dask?", or "why did the ML platform team standardise on Ray for batch inference instead of PySpark?" It walks through the three origins and mental models — Spark as a mature JVM SQL engine with a distributed dataframes API, Dask as a pure-Python task-graph scheduler that scales pandas and NumPy with the lightest possible lift, and Ray as a distributed-futures runtime with stateful actors and a ray data layer built for batch inference — the four axes interviewers actually probe (workload shape, the JVM language boundary, the scheduler model, operational surface), the same groupby aggregation written three ways, and the shuffle and partitioning behaviour that decides your cluster bill. Each section pairs a teaching block with a Solution-Tail interview answer — runnable code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the data-processing practice library →, rehearse on the ETL practice library →, and sharpen the systems axis with the design practice library →.

---

On this page

• Why the engine choice defines your team velocity
• Spark — the mature JVM lakehouse standard
• Dask — pandas at scale, task graphs
• Ray — distributed futures and Ray Data
• Head-to-head — decision matrix and interview signals
• Cheat sheet — distributed compute engine recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why the engine choice defines your team velocity

Three engines, three origin stories — the mental model you pick binds your platform for years

The one-sentence invariant: choosing a distributed compute engine is a picking exercise between a JVM SQL engine that treats every job as an optimised query plan (Spark), a pure-Python scheduler that treats every job as a graph of pandas/NumPy tasks (Dask), and a distributed-futures runtime that treats every job as a mesh of remote function calls and stateful actors (Ray) — and each engine trades away something the other two keep, in a way your data scientists, your ops team, and your finance team will all feel for years. The engine you standardise on in quarter one becomes the engine every downstream job, every CI pipeline, every on-call runbook, and every new hire's mental model is built around. Migrating off it later is not a config change; it is a rewrite.

The three origins in one line each.

• Spark — born to replace MapReduce for SQL/ETL at scale. It is a JVM engine. Its native tongue is a query plan: you describe what you want with a DataFrame or SQL, and Catalyst decides how. pyspark is a thin Python wrapper over that JVM engine — powerful, mature, and separated from your Python code by a process boundary.
• Dask — born to make pandas and NumPy bigger than one machine. It is pure Python. Its native tongue is a task graph: every .groupby() or .mean() builds a lazy DAG of Python callables that a scheduler runs across threads, processes, or a dask distributed cluster. If your team already thinks in pandas, Dask is the smallest conceptual jump.
• Ray — born to make distributed Python for ML feel like local Python. Its native tongue is the future: f.remote(x) schedules a function on the cluster and hands you back a reference. Add stateful actors for models and services, and ray data on top for streaming datasets, and you have the runtime the modern ML/inference stack is built on.

The four axes interviewers actually probe.

• Workload shape. Is the job SQL/ETL (scan Parquet, join, aggregate, write a table) or ML/inference (featurise, train, score a model over billions of rows)? Spark dominates the first; Ray dominates the second; Dask straddles the middle for pandas-shaped analytics. Naming which axis a workload sits on is the first senior signal.
• The language boundary (JVM vs pure Python). Spark runs on the JVM; your Python UDFs cross a serialization boundary into a JVM executor, which is where the profiling gets hard and the performance cliffs hide. Dask and Ray are pure Python — your stack trace is your stack trace, and a NumPy-heavy UDF runs at native speed without a JVM in the middle.
• The scheduler model. Spark schedules stages separated by shuffles (a bulk-synchronous, all-or-nothing model). Dask schedules a fine-grained task graph (millions of small tasks, dynamically). Ray schedules tasks and actors against a shared object store (the most flexible, and the only one with first-class stateful workers). The scheduler model dictates what kinds of jobs feel natural and which fight the engine.
• Operational surface. Spark means running (or renting) a JVM cluster — Databricks, EMR, or self-managed YARN/K8s. Dask is a lightweight Python cluster you can stand up from a notebook. Ray is a Python cluster with a richer control plane (the GCS, the dashboard, autoscaling for heterogeneous CPU+GPU nodes). More power usually means more to operate.

The 2026 reality — no universal winner, three clear lanes.

• Spark owns the lakehouse. For batch SQL/ETL over Parquet/Delta/Iceberg at terabyte-to-petabyte scale, Spark is the default and it is not close. Catalyst, Adaptive Query Execution, and a decade of production hardening make it the safe, boring, correct choice for the warehouse feed.
• Dask is the lightest lift for Python analytics. When a pandas/NumPy/scikit-learn workflow outgrows one machine but the code and the team's mental model should not have to change, Dask scales it with the same API. It is the shortest path from "notebook" to "cluster."
• Ray owns ML and inference. Distributed training, hyperparameter search, reinforcement learning, model serving, and especially batch inference over huge datasets are where Ray's futures-plus-actors model and Ray Data's streaming execution pull decisively ahead. The modern LLM and recommendation stacks are built on it.
• They interoperate. This is not a cage match. RayDP runs Spark on Ray; Dask-on-Ray uses Ray as the scheduler; Spark can hand a DataFrame to Ray Data for the inference tail. Senior teams often run two engines — Spark for the warehouse, Ray for inference — and glue them.

What interviewers listen for.

• Do you name all three engines and their origin problem without prompting? — senior signal.
• Do you say "Spark is a JVM SQL engine; pyspark is a wrapper" rather than treating PySpark as native Python? — required answer.
• Do you route by workload shape ("SQL/ETL → Spark, pandas-at-scale → Dask, batch inference → Ray") rather than by popularity? — senior signal.
• Do you name shuffle as the shared cost centre across all three, not a Spark-only concept? — required answer.
• Do you mention that they interoperate (RayDP, Dask-on-Ray) instead of framing it as mutually exclusive? — senior signal.

Worked example — the three-engine comparison table

Detailed explanation. The single most useful artifact for an engine-selection interview is a memorised comparison table. Every senior discussion converges on this table within the first ten minutes; having it in your head is what separates a fluent answer from a stumbling one. Walk through building the table for a hypothetical platform team that runs both a nightly warehouse ETL job and a daily batch-inference pass.

• Warehouse job. Read 4 TB of Parquet, join to a dimension table, aggregate to daily rollups, write a Delta table.
• Inference job. Score a 300-million-row embeddings table through a GPU model, write predictions back.
• Team. Eight Python-first engineers; nobody wants to write Scala; two data scientists live in pandas.
• Constraint. One platform if possible; two engines only if the workload demands it.

Question. Build the three-engine comparison and map each of the two jobs to the engine that fits it best.

Input.

Axis	Spark	Dask	Ray
Origin problem	SQL/ETL at scale (post-MapReduce)	pandas/NumPy bigger than RAM	distributed Python for ML
Runtime	JVM (pyspark is a wrapper)	pure Python	pure Python
Native abstraction	DataFrame / SQL query plan	task graph over pandas partitions	futures + actors + Ray Data
Scheduler	stages split by shuffle	fine-grained task DAG	tasks + actors on object store
Sweet spot	lakehouse ETL, big joins	pandas-shaped analytics at scale	training + batch inference
GPU story	possible, awkward	possible, manual	first-class

Code.

# A tiny "shape detector" that maps a workload description to an engine.
# Illustrative — the point is the routing logic, not production code.
from dataclasses import dataclass

@dataclass
class Workload:
    kind: str            # "sql_etl" | "pandas_analytics" | "ml_inference" | "training"
    needs_gpu: bool
    team_language: str   # "python" | "scala"
    data_scale: str      # "gb" | "tb" | "pb"

def pick_engine(w: Workload) -> str:
    if w.kind == "sql_etl" and w.data_scale in ("tb", "pb"):
        return "spark"                       # lakehouse ETL is Spark's home turf
    if w.kind == "pandas_analytics":
        return "dask"                        # keep the pandas API, scale it out
    if w.kind in ("ml_inference", "training") or w.needs_gpu:
        return "ray"                         # futures + actors + Ray Data
    # Fallback: small-to-medium SQL/ETL for a pure-Python team can go either way
    return "dask" if w.team_language == "python" else "spark"

warehouse = Workload("sql_etl", needs_gpu=False, team_language="python", data_scale="tb")
inference = Workload("ml_inference", needs_gpu=True, team_language="python", data_scale="tb")

print(pick_engine(warehouse))   # -> spark
print(pick_engine(inference))   # -> ray

Step-by-step explanation.

• The router keys on workload kind first, scale second. A 4 TB SQL/ETL job routes to Spark because the join-and-aggregate-over-Parquet workload is exactly what Catalyst and AQE are tuned for, and the JVM boundary barely matters when the logic is pure SQL.
• The inference job routes to Ray because needs_gpu plus ml_inference is the Ray Data lane: streaming batches into GPU actors is a first-class primitive there and an awkward bolt-on everywhere else.
• The fallback branch is where Dask earns its place. For a pure-Python team doing medium-scale, pandas-shaped analytics, Dask wins on conceptual overhead — the same code that ran on a laptop runs on the cluster.
• The table, not the code, is the interview artifact. The router just encodes the table's logic; in the room you draw the table and talk through the two mappings.

Output.

Job	Winning engine	Why
Nightly 4 TB Parquet ETL	Spark	Catalyst-optimised joins + AQE + mature Delta support
300M-row GPU batch inference	Ray	Ray Data streams batches into GPU actors natively
(hypothetical) pandas feature job	Dask	keep the pandas API; scale past one machine's RAM

Rule of thumb. Never pick a distributed compute engine by popularity. Pick it by workload shape (SQL/ETL vs pandas-analytics vs ML/inference), the language boundary your team can tolerate, and whether GPUs are in the picture. Draw the three-engine table first; the mapping falls out of the constraints.

Worked example — the JVM boundary that trips up Python teams

Detailed explanation. The single most misunderstood thing about pyspark is that it is not Python running your logic — it is a Python client driving a JVM engine. Understanding where the process boundary sits explains most of Spark's performance surprises and most of the reasons a pure-Python team reaches for Dask or Ray. Walk through what happens when you call a Python UDF in PySpark versus the same function in Dask.

• PySpark path. Your Python row-UDF is pickled, shipped to a JVM executor, which spins up a Python worker process, serializes each row out of the JVM, runs your function, and serializes the result back — per row, across the boundary.
• Dask path. Your function is a plain Python callable running in a Python worker on a partition that is already a pandas DataFrame. No cross-language serialization; NumPy stays in NumPy.
• The consequence. A NumPy-vectorised transform can be an order of magnitude slower as a PySpark row-UDF than as the same code in Dask, purely from boundary-crossing overhead — which is why Spark pushes you toward built-in SQL functions and Pandas/Arrow UDFs.

Question. Explain, with code, why the same Python transform behaves so differently on the two engines, and what the Spark-native fix is.

Input.

Engine	Where your Python runs	Cross-language cost
PySpark row UDF	Python worker beside a JVM executor	serialize every row JVM <-> Python
PySpark Pandas UDF (Arrow)	Python worker, batched via Arrow	serialize per Arrow batch (cheap)
Dask	Python worker on a pandas partition	none (already Python/NumPy)

Code.

# --- PySpark: the slow path (row-at-a-time Python UDF) ---
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, pandas_udf, col
from pyspark.sql.types import DoubleType
import pandas as pd

spark = SparkSession.builder.appName("boundary-demo").getOrCreate()
df = spark.range(0, 50_000_000).withColumnRenamed("id", "x")

# Slow: each row crosses the JVM<->Python boundary individually
@udf(DoubleType())
def slow_scale(x):
    return float(x) * 1.5

# Fast: Arrow batches an entire column of rows across the boundary once
@pandas_udf(DoubleType())
def fast_scale(x: pd.Series) -> pd.Series:
    return x * 1.5           # vectorised NumPy inside a Python worker

df.select(slow_scale(col("x"))).write.mode("overwrite").format("noop").save()
df.select(fast_scale(col("x"))).write.mode("overwrite").format("noop").save()

# --- Dask: the same transform, no boundary at all ---
import dask.dataframe as dd
import numpy as np

ddf = dd.from_pandas(pd.DataFrame({"x": np.arange(50_000_000)}), npartitions=64)
ddf["scaled"] = ddf["x"] * 1.5      # runs as native NumPy on each pandas partition
ddf["scaled"].sum().compute()

Step-by-step explanation.

• slow_scale is a per-row Python UDF. Spark cannot see inside it, so it falls back to shipping every row out of the JVM into a Python worker, one value at a time. On 50M rows this serialization dominates the runtime.
• fast_scale is a Pandas (Arrow) UDF. Spark ships whole columns as Arrow record batches, your NumPy vectorises over the batch, and the result returns as a batch. Same Python, a fraction of the boundary cost — this is the Spark-native fix.
• The Dask version never crosses a boundary. Each partition is already a pandas DataFrame in a Python process, so ddf["x"] * 1.5 is plain vectorised NumPy. There is no JVM, so there is nothing to serialize.
• The lesson is not "Spark is slow." For SQL/ETL expressed in built-in functions, Spark is extremely fast because everything stays in the JVM/Tungsten. The boundary only bites when your Python logic is on the hot path — which is exactly when pandas-native Dask or Ray becomes attractive.

Output.

Transform	Engine	Relative cost	Why
slow_scale row UDF	PySpark	slowest	per-row JVM <-> Python serialization
fast_scale Pandas UDF	PySpark	fast	Arrow-batched crossing; vectorised
built-in col("x") * 1.5	PySpark	fastest	stays in JVM/Tungsten, no Python
ddf["x"] * 1.5	Dask	fast	native NumPy, no boundary

Rule of thumb. On Spark, keep your hot path in built-in SQL functions or Arrow-based Pandas UDFs; a row-at-a-time Python UDF is a JVM-boundary tax. When your logic is fundamentally Python/NumPy and must stay on the hot path, that tax is the strongest argument for Dask or Ray.

Worked example — the decision tree every senior engineer runs in their head

Detailed explanation. Given a new workload, the senior architect runs a short decision tree before naming an engine. Codifying it makes the interview answer reproducible — an interviewer can hand you any scenario and you can walk the tree out loud. Walk it with three canonical scenarios: a warehouse rollup, a pandas feature job that outgrew RAM, and a GPU batch-scoring pass.

• Q1. Is the core of the job SQL/ETL over lake tables (join, aggregate, write)? → yes = Spark is the default.
• Q2. Is the core an existing pandas/NumPy/scikit workflow that just needs to scale, with no rewrite? → yes = Dask.
• Q3. Is the core ML — distributed training, hyperparameter search, or batch inference (especially GPU)? → yes = Ray.
• Q4. Mixed platform? → run the primary lane's engine, and glue a second engine only where its lane clearly wins (e.g. Spark ETL feeding Ray inference).

Question. Walk the decision tree for the three scenarios and record the engine each ends up with.

Input.

Scenario	Q1 (SQL/ETL?)	Q2 (scale pandas?)	Q3 (ML/GPU?)
Nightly warehouse rollup	yes	—	—
Feature job that blew past RAM	no	yes	no
GPU batch scoring of 300M rows	no	no	yes

Code.

def route(sql_etl: bool, scale_pandas: bool, ml_or_gpu: bool) -> str:
    if sql_etl:
        return "Spark (lakehouse ETL default)"
    if scale_pandas:
        return "Dask (keep pandas API, scale out)"
    if ml_or_gpu:
        return "Ray (training / batch inference)"
    return "Dask or Spark — decide by team language + scale"

print(route(True,  False, False))   # Nightly warehouse rollup
# -> Spark (lakehouse ETL default)

print(route(False, True,  False))   # Feature job past RAM
# -> Dask (keep pandas API, scale out)

print(route(False, False, True))    # GPU batch scoring
# -> Ray (training / batch inference)

Step-by-step explanation.

• Scenario 1 short-circuits at Q1. A join-aggregate-write over lake tables is Spark's exact design centre; Catalyst optimises the plan and AQE fixes skew at runtime. No reason to reach further.
• Scenario 2 fails Q1 but passes Q2. The logic is already pandas; rewriting it in Spark SQL would be a project. Dask runs the same code across partitions, so the mapping is "scale, don't rewrite."
• Scenario 3 passes only Q3. GPU batch inference over hundreds of millions of rows is where Ray Data's streaming execution and GPU actors are native; Spark and Dask can be coerced into it but fight you on GPU scheduling and pipelining.
• The Q4 branch is the senior move. Real platforms are mixed. The mature answer is not "one engine forever" but "the right engine per lane, glued at the boundary" — Spark writes the feature table, Ray Data reads it and scores it.

Output.

Scenario	Engine	One-line justification
Nightly warehouse rollup	Spark	lake ETL is Catalyst's home turf
Feature job past RAM	Dask	scale pandas without a rewrite
GPU batch scoring	Ray	Ray Data + GPU actors are native

Rule of thumb. The engine decision tree is a whiteboard-friendly answer: SQL/ETL → Spark, scale-pandas → Dask, ML/GPU/inference → Ray, mixed → glue at the lane boundary. Practice walking it so an interviewer can hand you any scenario and get an engine name in under 60 seconds.

Senior interview question on engine selection

A senior interviewer often opens with: "Your company runs a nightly lakehouse ETL job in Spark, and a new ML team wants to batch-score a 300-million-row table through a GPU model daily. The ML team is pure Python and finds PySpark UDFs painful. Do you make them use Spark for consistency, or introduce a second engine — and how do you justify the operational cost of running two?"

Solution Using a two-engine platform with Spark for ETL and Ray for inference

# Stage 1 — Spark writes the governed feature table (its home turf)
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("nightly-features").getOrCreate()

events = spark.read.format("delta").load("s3://lake/events")
dims   = spark.read.format("delta").load("s3://lake/dim_user")

features = (
    events
    .join(F.broadcast(dims), "user_id")          # small dim -> broadcast join
    .groupBy("user_id")
    .agg(
        F.count("*").alias("event_count"),
        F.avg("amount").alias("avg_amount"),
        F.max("ts").alias("last_seen"),
    )
)
features.write.format("delta").mode("overwrite").save("s3://lake/user_features")

# Stage 2 — Ray Data reads that table and streams it through a GPU model
import ray
import numpy as np

ray.init(address="auto")                          # attach to the Ray cluster

ds = ray.data.read_parquet("s3://lake/user_features")

class Scorer:
    def __init__(self):
        import torch
        self.model = torch.jit.load("model.pt").eval().cuda()

    def __call__(self, batch: dict) -> dict:
        import torch
        x = torch.as_tensor(
            np.stack([batch["event_count"], batch["avg_amount"]], axis=1),
            dtype=torch.float32,
        ).cuda()
        with torch.no_grad():
            batch["score"] = self.model(x).cpu().numpy().ravel()
        return batch

scored = ds.map_batches(
    Scorer,
    batch_size=4096,
    num_gpus=1,                                   # each Scorer actor gets a GPU
    concurrency=4,                                # 4 GPU actors in parallel
)
scored.write_parquet("s3://lake/user_scores")

Step-by-step trace.

Step	Spark stage (ETL)	Ray stage (inference)
Read	Delta events + dim tables	Parquet feature table
Core op	broadcast join + groupBy agg	map_batches into GPU actors
Where Python runs	mostly JVM built-ins (fast)	pure Python + PyTorch on GPU
Parallelism unit	stages split by shuffle	streaming batches across 4 actors
Output	governed Delta feature table	Parquet score table
Handoff	writes user_features	reads user_features

The two engines meet at exactly one artifact — the user_features Delta/Parquet table. Spark does what it is best at (optimised joins and aggregations over lake tables, all in the JVM), and Ray does what it is best at (streaming batches into GPU actors with native GPU scheduling). Neither engine is bent out of its lane, and the only shared surface is a table contract.

Output:

Concern	Single-engine (Spark only)	Two-engine (Spark + Ray)
ETL performance	excellent	excellent (unchanged)
GPU batch inference	awkward (UDF + manual GPU)	native (Ray Data + GPU actors)
ML team ergonomics	painful (PySpark UDFs)	pure Python, no JVM boundary
Operational surface	one cluster	two clusters (real added cost)
Coupling	tight (one job)	loose (table contract)

Why this works — concept by concept:

• Lane separation — the ETL job and the inference job sit on different axes (SQL/ETL vs ML/inference), so forcing one engine to serve both means one of the two lanes runs against the grain. Splitting by lane lets each engine operate in its design centre.
• Table-contract handoff — the engines are coupled only through a governed feature table, not a shared runtime. This is the loosest possible coupling: either engine can be swapped without touching the other, and the contract is a schema, not an API.
• Broadcast join + groupBy in Spark — the small dimension table is broadcast so the join needs no shuffle, and the aggregation is a built-in Catalyst operation, so almost nothing leaves the JVM. This is why Spark's ETL cost stays low.
• map_batches with GPU actors in Ray — Ray Data streams fixed-size batches into a pool of stateful Scorer actors that each hold a GPU-resident model. The model loads once per actor (not once per batch), and num_gpus=1 lets Ray's scheduler place actors on GPU nodes automatically.
• Cost — the honest cost of the two-engine answer is a second cluster to operate (a real O(ops) tax). It is justified only because the inference lane's productivity and GPU-utilisation gains exceed that tax; if the inference job were tiny, the correct answer would be "coerce it into Spark and keep one cluster."

Python
Topic — data-processing
Data-processing problems on distributed engines

Practice →

Design Topic — design Design problems on compute-engine platforms

Practice →

---

2. Spark — the mature JVM lakehouse standard

A distributed SQL engine with Catalyst and Tungsten under a DataFrame API — the safe default for lakehouse ETL

The mental model in one line: Spark is a JVM query engine where you describe a computation as a DataFrame or SQL expression, the Catalyst optimizer rewrites it into an efficient physical plan of stages separated by shuffles, and the Tungsten execution engine runs those stages over off-heap memory — so pyspark is best understood as a Python remote control for a highly-optimised JVM database, unbeatable for SQL/ETL over lake tables and awkward exactly when your own Python logic has to sit on the hot path. Every senior data engineer has shipped a Spark pipeline; the ones who ship fast ones understand that the DataFrame API is a query builder, not a Python loop.

The four axes for Spark.

• Workload shape. SQL/ETL is Spark's home: scan columnar files, filter, join, aggregate, window, write. When the computation can be expressed in DataFrame/SQL operations, Spark is exceptional. Row-by-row Python logic is its weak spot.
• Language boundary. Spark is JVM. pyspark drives it from Python, but your data lives in the JVM and your Python UDFs cross a boundary. Keep logic in built-in functions and Arrow-based Pandas UDFs to stay JVM-side.
• Scheduler model. Spark builds a DAG of stages; a shuffle (any wide dependency like a join or groupBy) marks a stage boundary where data is repartitioned across the network. Stages run bulk-synchronously; understanding shuffles is understanding Spark performance.
• Operational surface. Spark means a JVM cluster — Databricks, EMR, Dataproc, or self-managed on YARN/Kubernetes. It is heavier to operate than a Dask cluster but has the deepest managed-service ecosystem.

Catalyst and Tungsten — why Spark is fast for SQL.

• Catalyst optimizer. Turns your logical plan into an optimised physical plan: predicate pushdown, projection pruning, join reordering, constant folding. You write declaratively; Catalyst decides execution.
• Tungsten execution. Off-heap memory management, cache-aware binary row formats, and whole-stage code generation that compiles a chain of operators into a single tight JVM loop.
• Adaptive Query Execution (AQE). Re-optimises the plan at runtime using actual shuffle statistics — coalescing small partitions, switching join strategies, and splitting skewed partitions. AQE is why modern Spark handles skew better than the Spark of five years ago.

The shuffle — the tax you must respect.

• What it is. Any wide transformation (join on a key, groupBy, distinct, repartition) requires records with the same key to end up on the same executor. Spark writes intermediate data to disk and moves it across the network — the shuffle.
• Why it dominates cost. Shuffles serialize, spill, and network-transfer data; a poorly-partitioned join can move terabytes. Most Spark tuning is shuffle-avoidance: broadcast the small side of a join, pre-partition by the join key, and let AQE coalesce.
• Skew. When one key holds a disproportionate share of rows, one shuffle partition becomes a straggler that stalls the whole stage. The classic fix is salting the skewed key; AQE's skew-join handling automates much of it.

Common beginner mistakes.

• Treating a Python UDF as free. A row-at-a-time UDF drags data across the JVM boundary; prefer built-in functions or Pandas UDFs.
• Calling .collect() on a big DataFrame. This pulls the entire result to the driver's memory and OOMs it. Use .write, .take(n), or aggregate first.
• Ignoring partition count. Too few partitions underuses the cluster; too many floods the scheduler with tiny tasks. Tune spark.sql.shuffle.partitions (or trust AQE).
• Shuffling when you could broadcast. Joining a huge fact table to a small dimension without broadcast() triggers a full shuffle join instead of a cheap broadcast hash join.

Worked example — the reference groupby aggregation in PySpark

Detailed explanation. The canonical SQL/ETL task: read a transactions table, aggregate total revenue and transaction count per product category, and write the result. This is the exact same computation we will write in Dask and Ray in the next sections, so you can compare the three engines head-to-head on one workload.

• Input. A Parquet transactions table with category, amount, user_id.
• Operation. Group by category; sum amount; count rows.
• Output. A small per-category rollup table.

Question. Write the PySpark job that produces per-category revenue and transaction counts from a Parquet source.

Input.

Column	Type	Example
category	string	"books"
amount	double	12.50
user_id	bigint	1001

Code.

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

spark = (
    SparkSession.builder
    .appName("category-revenue")
    .config("spark.sql.shuffle.partitions", "200")   # tune shuffle width
    .getOrCreate()
)

tx = spark.read.parquet("s3://lake/transactions")

rollup = (
    tx.groupBy("category")                # wide dependency -> a shuffle
      .agg(
          F.sum("amount").alias("revenue"),
          F.count("*").alias("txn_count"),
      )
      .orderBy(F.col("revenue").desc())
)

rollup.write.mode("overwrite").parquet("s3://lake/category_rollup")
rollup.show(truncate=False)

Step-by-step explanation.

• read.parquet is lazy. Nothing executes yet; Spark records the read as the root of a logical plan and reads only the columns the downstream plan needs (projection pushdown).
• groupBy("category").agg(...) introduces a shuffle. Rows must be repartitioned so that all rows of a category land together. Spark first does a partial aggregation on each partition (a map-side combine), then shuffles the small partials, then finalises — so the network only carries per-partition partials, not raw rows.
• orderBy adds a second shuffle (a range partition) to sort globally. If you only need top-N, orderBy(...).limit(n) lets Catalyst push a partial top-N and avoid a full sort.
• write.parquet triggers execution. This is the action that forces the whole plan to run. Up to here everything was a plan; the write is what actually reads, shuffles, aggregates, and materialises.

Output.

category	revenue	txn_count
electronics	4,812,004.50	128,004
books	1,204,551.75	512,880
grocery	990,220.10	1,004,552

Rule of thumb. In PySpark, express the whole pipeline as chained DataFrame operations and end with a single action (write/show). The aggregation shuffles per-partition partials, not raw rows — which is why a groupBy over a billion rows can still be cheap.

Worked example — Catalyst, partition pruning, and the broadcast join

Detailed explanation. The biggest Spark performance wins come from letting Catalyst avoid work: reading fewer files (partition pruning), scanning fewer columns (projection pushdown), and avoiding shuffles (broadcast joins). Walk through a job that joins a huge fact table to a small dimension and see how to keep it shuffle-free.

• Partition pruning. A filter on a partition column (e.g. dt = '2026-08-01') lets Spark skip entire directories of files.
• Broadcast join. When one join side is small, Spark ships it to every executor and joins locally — no shuffle of the big side.
• explain() is your friend. The physical plan tells you whether you got a BroadcastHashJoin (good) or a SortMergeJoin (a full shuffle).

Question. Join a partitioned fact table to a small dimension table without shuffling the fact table, and confirm the plan.

Input.

Table	Rows	Partitioned by	Join key
fact_sales	8,000,000,000	dt	product_id
dim_product	40,000	—	product_id

Code.

from pyspark.sql import functions as F

fact = spark.read.format("delta").load("s3://lake/fact_sales")
dim  = spark.read.format("delta").load("s3://lake/dim_product")

result = (
    fact
    .where(F.col("dt") == "2026-08-01")          # partition pruning: skip other days
    .select("product_id", "amount")              # projection pushdown: 2 columns only
    .join(F.broadcast(dim), "product_id")        # broadcast the 40k-row dimension
    .groupBy("category")
    .agg(F.sum("amount").alias("revenue"))
)

result.explain(mode="formatted")                 # look for BroadcastHashJoin
result.write.mode("overwrite").parquet("s3://lake/daily_category_revenue")

Step-by-step explanation.

• The where(dt == ...) prunes partitions before any data is read. Because fact_sales is physically partitioned by dt, Spark reads only that one day's directory — a filter that eliminates 99%+ of the files without opening them.
• The select triggers projection pushdown. Only product_id and amount are read from the Parquet/Delta files; the other columns are never decoded, cutting I/O.
• F.broadcast(dim) forces a broadcast hash join. The 40k-row dimension is collected to the driver, shipped to every executor, and held in a hash map; each fact partition joins locally. The 8-billion-row fact table never shuffles.
• explain() confirms the plan. You want BroadcastHashJoin in the physical plan. If you see SortMergeJoin, the broadcast hint was ignored (usually because the "small" side exceeded spark.sql.autoBroadcastJoinThreshold), and you are paying for a full shuffle.

Output.

Plan node	Present?	Meaning
PartitionFilters: dt = 2026-08-01	yes	only one day's files read
ReadSchema: product_id, amount	yes	projection pushdown worked
BroadcastHashJoin	yes	no shuffle of the fact table
Exchange (shuffle)	only for groupBy	join added zero shuffles

Rule of thumb. The three cheapest Spark wins are partition pruning (filter on the partition column), projection pushdown (select only needed columns), and broadcast joins (hint the small side). Read explain() and hunt for BroadcastHashJoin; a surprise SortMergeJoin is a shuffle you did not budget for.

Worked example — diagnosing a shuffle and letting AQE fix skew

Detailed explanation. A join keyed on a column where one value dominates (a NULL user, a "guest" account, a mega-merchant) creates a skewed shuffle: one partition gets billions of rows while the rest finish in seconds, and the whole stage waits on the straggler. Walk through spotting it and the two fixes — Adaptive Query Execution and manual salting.

• Symptom. One task in a stage runs 50x longer than the median; the stage is "99% done" for an hour.
• AQE fix. spark.sql.adaptive.enabled=true plus spark.sql.adaptive.skewJoin.enabled=true lets Spark split the skewed partition into sub-partitions at runtime.
• Manual fix. Salt the skewed key: append a random suffix so the hot key spreads across many partitions, join, then aggregate away the salt.

Question. Turn on AQE skew handling for a skewed join, and show the manual salting fallback for the worst hot keys.

Input.

Setting	Value	Effect
spark.sql.adaptive.enabled	true	AQE on
spark.sql.adaptive.skewJoin.enabled	true	split skewed partitions
salt buckets (manual)	16	spread the hottest key

Code.

# --- Fix 1: let AQE handle skew automatically (preferred) ---
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

joined = big_fact.join(other_fact, "user_id")     # AQE splits skewed user_id partitions

# --- Fix 2: manual salting when a few keys are pathologically hot ---
from pyspark.sql import functions as F

SALT = 16
salted_left = (
    big_fact
    .withColumn("salt", (F.rand() * SALT).cast("int"))
    .withColumn("join_key", F.concat_ws("_", "user_id", "salt"))
)
salted_right = (
    other_fact
    .withColumn("salt", F.explode(F.array([F.lit(i) for i in range(SALT)])))
    .withColumn("join_key", F.concat_ws("_", "user_id", "salt"))
)
result = (
    salted_left.join(salted_right, "join_key")
    .groupBy("user_id")                            # aggregate the salt back away
    .agg(F.sum("amount").alias("total"))
)

Step-by-step explanation.

• AQE is the first thing to try. With skew-join enabled, Spark inspects real shuffle-partition sizes at runtime and splits any partition that is much larger than the median into several tasks, so the hot key's rows spread across workers. No code change beyond the config.
• Salting is the manual escape hatch for cases AQE cannot fully absorb. On the big side you attach a random salt 0..15 to the hot key; on the small side you replicate each row 16 times, once per salt value, using explode. Now the hot key becomes 16 keys, joined across 16 partitions.
• The salt is joined away at the end. After the join spreads the load, you groupBy the original user_id to recombine the 16 salted buckets back into one row per user — the salt exists only to break the skew during the shuffle.
• The cost of salting is replication on the small side. Replicating the dimension 16x is cheap when it is small; it would be expensive if both sides were large, which is exactly when AQE (not salting) is the right tool.

Output.

Approach	Straggler task	Code change	When to use
No fix	50x median (stalls stage)	none	never (baseline)
AQE skew join	near-median	one config flag	first choice
Manual salting	near-median	rewrite the join	a few pathological keys

Rule of thumb. Reach for AQE skew handling first — it is a config flag and it covers most skew. Fall back to salting only for a handful of pathologically hot keys, and always aggregate the salt away afterward so the result is unchanged.

Senior interview question on Spark

A senior interviewer might ask: "You have a Spark job joining an 8-billion-row clicks table to a 3-billion-row impressions table on user_id, and the stage is stuck at 99% for 40 minutes because a handful of bot users have billions of rows each. Walk me through diagnosing the skew and fixing it, and explain why a broadcast join is not an option here."

Solution Using AQE skew handling plus targeted salting of the hottest keys

from pyspark.sql import functions as F

# 1. Diagnose: find the skewed keys before touching the join
skew = (
    clicks.groupBy("user_id")
    .count()
    .orderBy(F.col("count").desc())
    .limit(20)
)
skew.show()   # a few bot user_ids with billions of rows each

# 2. Turn on AQE skew join — the first line of defense
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256m")

# 3. For the worst offenders AQE cannot fully split, salt just those keys
HOT = [111, 222, 333]          # bot user_ids from step 1
SALT = 32

def salt_hot(df):
    is_hot = F.col("user_id").isin(HOT)
    return (
        df.withColumn(
            "salt",
            F.when(is_hot, (F.rand() * SALT).cast("int")).otherwise(F.lit(0)),
        )
        .withColumn("jk", F.concat_ws("_", "user_id", "salt"))
    )

left = salt_hot(clicks)
# replicate only the hot keys on the right side, salt 0 for the rest
right_hot = (
    impressions.where(F.col("user_id").isin(HOT))
    .withColumn("salt", F.explode(F.array([F.lit(i) for i in range(SALT)])))
    .withColumn("jk", F.concat_ws("_", "user_id", "salt"))
)
right_cold = (
    impressions.where(~F.col("user_id").isin(HOT))
    .withColumn("salt", F.lit(0))
    .withColumn("jk", F.concat_ws("_", "user_id", "salt"))
)
right = right_hot.unionByName(right_cold)

joined = (
    left.join(right, "jk")
    .groupBy("user_id")
    .agg(F.count("*").alias("matched"))
)
joined.write.mode("overwrite").parquet("s3://lake/click_impression_join")

Step-by-step trace.

Step	Action	Effect
1	count rows per user_id	identifies the 3 bot keys causing skew
2	enable AQE skew join	Spark splits oversized shuffle partitions at runtime
3	salt only the hot keys	3 hot keys become 3x32 keys spread across partitions
3b	replicate hot keys on the right	preserves join correctness after salting
4	groupBy user_id	recombines salted buckets; salt disappears from output

Because both sides are multi-billion-row tables, a broadcast join is impossible — neither side fits in an executor's memory, so Spark must shuffle. AQE splits the merely-large partitions automatically; the three pathological bot keys, which are too hot even for AQE's default splitting, get targeted salting that scatters their rows across 32 sub-keys while the cold majority pays no overhead (salt 0).

Output:

Metric	Before	After
Straggler task runtime	~40 min	~90 s
Stage wall-clock	~45 min	~4 min
Data shuffled	unchanged	unchanged
Extra work	none	32x replication of 3 hot keys only
Result correctness	correct	identical

Why this works — concept by concept:

• Diagnose before you fix — counting rows per key turns "the stage is stuck" into "three specific bot user_ids are the problem." You cannot fix skew you have not measured, and the fix (AQE vs salting) depends on how many keys are hot.
• Broadcast is off the table — a broadcast join requires one side to fit in executor memory. With both sides in the billions, Spark must do a shuffle (sort-merge) join, which is precisely why skew bites.
• AQE skew join — Adaptive Query Execution reads real shuffle-partition sizes at runtime and splits any partition exceeding the skew threshold into multiple tasks, so a large-but-not-pathological key stops being a straggler with zero code change.
• Targeted salting — for the three keys too hot even for AQE, salting turns each into 32 sub-keys; replicating only those keys on the right side keeps the join correct while the cold majority (salt 0) pays nothing. The final groupBy user_id erases the salt.
• Cost — the added work is a 32x row replication of exactly three keys (negligible against billions of rows) plus one config change. The eliminated cost is a 40-minute straggler that idles the entire cluster. Net: minutes instead of near-an-hour, at O(hot-keys) extra work, not O(rows).

Python
Topic — data-processing
Data-processing problems on Spark aggregations and joins

Practice →

Perf Topic — optimization Optimization problems on shuffle and skew

Practice →

---

3. Dask — pandas at scale, task graphs

A pure-Python scheduler that parallelises pandas and NumPy across a task graph — the lightest lift from notebook to cluster

The mental model in one line: Dask is a pure-Python parallel-computing library where a high-level collection like dask.dataframe mirrors the pandas API, every operation builds a lazy task graph of small Python functions over partitioned pandas frames, and a scheduler (threads, processes, or a dask distributed cluster) executes that graph — so the whole value proposition is "your pandas/NumPy code, scaled past one machine's memory, with no rewrite and no JVM". Every data scientist who has hit MemoryError in pandas has reached for Dask; the ones who use it well understand that a Dask DataFrame is a lazy graph of pandas DataFrames, not a magic infinite-memory pandas.

The four axes for Dask.

• Workload shape. Pandas-shaped analytics at scale: groupbys, joins, rolling windows, and NumPy array math over data that is bigger than RAM but not necessarily petabyte-scale. Also excellent for embarrassingly-parallel custom Python (dask.delayed, dask.bag).
• Language boundary. None. Dask is pure Python; a partition is a real pandas DataFrame in a Python process. Your NumPy runs at native speed and your stack traces are Python stack traces.
• Scheduler model. A fine-grained task graph: each operation decomposes into many small tasks with explicit data dependencies, and a dynamic scheduler runs them, moving intermediate results between workers. This is more flexible than Spark's stage model but gives you less automatic query optimisation.
• Operational surface. The lightest of the three. LocalCluster spins up in a notebook; dask.distributed scales to many machines with a scheduler and workers; Coiled/managed options exist. No JVM, small dependency footprint.

The dask.dataframe model — a lazy graph of pandas frames.

• Partitions. A Dask DataFrame is a sequence of pandas DataFrames split along the index. Each partition should fit comfortably in a worker's memory (a common target is ~100 MB–1 GB per partition).
• Laziness. Operations build a graph; nothing runs until you call .compute() (materialise to pandas) or .persist() (keep in distributed memory). This is the same lazy/action split as Spark.
• The pandas API subset. Most common pandas operations work; some (like arbitrary index-shuffling reindexes) are expensive or unsupported because they would require a full shuffle.

Shuffles in Dask — the same tax, a different lever.

• When they happen. A groupby on a non-index column, a merge on a non-index column, or set_index all trigger a shuffle: partitions are re-split so matching keys land together.
• The control you have. set_index on the join/group key up front makes subsequent operations shuffle-free (Dask knows the data is already partitioned by that key). split_out controls how many output partitions a groupby produces.
• Why it can hurt more than Spark. Dask has less automatic query optimisation than Catalyst, so a naive pipeline can shuffle more than necessary. The fix is usually explicit: set the index, or restructure so the expensive shuffle happens once.

Common beginner mistakes.

• Too many or too few partitions. Thousands of tiny partitions flood the scheduler; a handful of giant ones OOM workers. Aim for partitions in the ~100 MB range and repartition when needed.
• Calling .compute() inside a loop. Each call runs the whole graph from scratch. Build one big graph, or persist() shared intermediates once.
• Expecting a full query optimiser. Dask does not reorder your joins the way Catalyst does; you must structure the pipeline (e.g. filter before merge, set the index before groupby) yourself.
• Ignoring the dashboard. The Dask dashboard shows task streams, memory, and shuffles in real time; skipping it means flying blind on performance.

Worked example — the reference groupby aggregation in dask.dataframe

Detailed explanation. The same per-category revenue rollup we wrote in PySpark, now in Dask. Note how close the code is to plain pandas — that closeness is the entire point of Dask. This is the second of our three head-to-head implementations of the identical workload.

• Input. The same Parquet transactions table.
• Operation. Group by category; sum amount; count rows.
• Output. A small per-category pandas DataFrame.

Question. Write the Dask job that produces per-category revenue and transaction counts, and explain where it shuffles.

Input.

Column	Type	Example
category	string	"books"
amount	float64	12.50
user_id	int64	1001

Code.

import dask.dataframe as dd
from dask.distributed import Client

client = Client(n_workers=8, threads_per_worker=2, memory_limit="8GB")  # local cluster

# Lazy read: one pandas partition per Parquet row-group / file chunk
tx = dd.read_parquet("s3://lake/transactions", columns=["category", "amount"])

rollup = (
    tx.groupby("category")               # triggers a shuffle across partitions
      .agg({"amount": "sum", "user_id": "count"})
      .rename(columns={"amount": "revenue", "user_id": "txn_count"})
)

result = rollup.compute()                # runs the graph; returns a pandas DataFrame
result = result.sort_values("revenue", ascending=False)
result.to_parquet("category_rollup.parquet")
print(result)

Step-by-step explanation.

• read_parquet builds a lazy graph with one task per file/row-group; columns=[...] is Dask's projection pushdown, so only two columns are read.
• groupby("category").agg(...) compiles to a shuffle. Dask does a partial aggregation per partition (a groupby on each pandas frame), then a tree reduction or shuffle to combine partials by category. Because the number of categories is small, this combine is cheap — the network carries per-partition partials, not raw rows, exactly like Spark's map-side combine.
• .compute() is the action that executes the whole graph and pulls the small result back to the client as a real pandas DataFrame. The sort happens in pandas on that small result, so it costs nothing at scale.
• The code reads like pandas because it is pandas per partition. That is the migration story: a data scientist changes import pandas as pd to import dask.dataframe as dd and read_parquet(...) and the aggregation is unchanged.

Output.

category	revenue	txn_count
electronics	4,812,004.50	128,004
books	1,204,551.75	512,880
grocery	990,220.10	1,004,552

Rule of thumb. In Dask, write the pipeline in the pandas API, keep partitions around 100 MB, and end with a single .compute(). A low-cardinality groupby combines cheaply because Dask (like Spark) aggregates per-partition partials before combining.

Worked example — reading the task graph and using persist

Detailed explanation. Dask's superpower and its footgun are both the task graph. Seeing the graph explains why an operation is slow, and persist() is the lever that stops you from recomputing shared work. Walk through visualising a graph and caching an intermediate.

• .visualize() renders the DAG so you can see the fan-out and the shuffle.
• .persist() computes a collection now and keeps it in distributed worker memory, so downstream operations reuse it instead of recomputing from the source.
• The rule. Persist an intermediate that is used more than once; do not persist everything (worker memory is finite).

Question. A pipeline filters a large frame, then computes both a groupby and a describe on the filtered result. Avoid reading and filtering the source twice.

Input.

Step	Reused?	Persist?
read + filter	used by 2 branches	yes
groupby branch	terminal	no
describe branch	terminal	no

Code.

import dask.dataframe as dd

tx = dd.read_parquet("s3://lake/transactions")

# Expensive shared prefix: read + filter, used by two downstream branches
recent = tx[tx["ts"] >= "2026-07-01"]

# Without persist, BOTH of the next lines re-read and re-filter the source.
recent = recent.persist()                # compute once, keep in cluster memory

by_cat   = recent.groupby("category")["amount"].sum().compute()
overall  = recent["amount"].describe().compute()

# Inspect the graph of a single branch to see the shuffle and fan-out
recent.groupby("category")["amount"].sum().visualize(filename="graph.png")

Step-by-step explanation.

• recent is a lazy sub-graph (read + filter). Without intervention, each terminal .compute() walks back to the source, so the read-and-filter would run twice — once for the groupby, once for the describe.
• recent.persist() executes that sub-graph once and pins the resulting partitions in worker memory. Both branches now start from the cached partitions instead of the Parquet files.
• The two .compute() calls run their own tails (a shuffle for the groupby, a reduction for the describe) but share the persisted prefix — the read and filter happen exactly once.
• .visualize() draws the DAG so you can literally see the partition fan-out and the shuffle node; when a job is mysteriously slow, the graph usually shows an unexpected shuffle or a partition explosion.

Output.

Metric	Without persist	With persist
Source reads	2	1
Filter passes	2	1
Peak memory	lower	higher (caches recent)
Wall-clock	slower	faster

Rule of thumb. Persist a Dask collection that feeds two or more downstream computations, and use .visualize() when a job is slow — the graph almost always reveals a redundant shuffle or a shared prefix you should have persisted.

Worked example — the distributed cluster and raw futures

Detailed explanation. Beyond DataFrames, Dask exposes a lower-level futures API (client.submit, client.map) that is the closest analogue to Ray's model — arbitrary Python functions scheduled across the cluster. This is how you parallelise custom Python that does not fit the DataFrame mould. Walk through submitting tasks and gathering results.

• client.submit(fn, x) schedules fn(x) on a worker and returns a Future immediately (lazy, asynchronous).
• client.map(fn, xs) submits one task per element.
• client.gather(futures) blocks and collects the results.

Question. Parallelise a custom per-file parsing function across a list of input files using Dask futures.

Input.

Component	Value
Cluster	dask.distributed, 8 workers
Task	parse_file(path) -> row_count
Inputs	500 file paths

Code.

from dask.distributed import Client, as_completed

client = Client("tcp://scheduler:8786")     # connect to a distributed cluster

def parse_file(path: str) -> int:
    import pandas as pd
    df = pd.read_json(path, lines=True)     # arbitrary custom Python per file
    return len(df.dropna(subset=["user_id"]))

paths = [f"s3://raw/events/{i:05d}.jsonl" for i in range(500)]

# Fan out: one task per file, all scheduled immediately
futures = client.map(parse_file, paths)

# Stream results as they finish (no need to wait for the slowest)
total = 0
for fut in as_completed(futures):
    total += fut.result()

print(f"total valid rows across {len(paths)} files: {total}")

Step-by-step explanation.

• client.map(parse_file, paths) submits 500 tasks at once and returns 500 futures without blocking. The scheduler distributes them across the 8 workers, keeping all cores busy.
• Each task is arbitrary Python — reading JSON, dropping nulls, counting — running natively in a worker process. This is the embarrassingly-parallel pattern the DataFrame API is clumsy for.
• as_completed yields futures in finish order, so fast files' results are accumulated immediately instead of waiting for the slowest file. This keeps the client responsive and the reduction incremental.
• This is Dask's Ray-like face. The futures API is a general task scheduler; the difference from Ray is that Dask has no first-class stateful actor and a less GPU-aware scheduler, which is exactly where Ray pulls ahead for ML.

Output.

Aspect	Behaviour
Tasks submitted	500, immediately
Blocking	only at fut.result()
Result order	completion order (via as_completed)
Worker utilisation	all 8 workers saturated

Rule of thumb. Use dask.dataframe for pandas-shaped analytics and drop to client.submit/client.map for custom embarrassingly-parallel Python. Dask's futures are its most Ray-like feature — but for stateful, GPU-heavy ML, that is exactly the boundary where Ray becomes the better tool.

Senior interview question on Dask

A senior interviewer might ask: "A data scientist has a pandas pipeline that groups a 400 GB event log by user_id and computes per-user features, and it OOMs on a single 64 GB machine. They do not want to rewrite it in Spark SQL. Walk me through scaling it with Dask — the partitioning, the index strategy, and how you keep the user_id groupby from shuffling the entire dataset twice."

Solution Using set_index on the group key plus tuned partitions and a single persist

import dask.dataframe as dd
from dask.distributed import Client

client = Client(n_workers=16, threads_per_worker=2, memory_limit="16GB")

# 1. Read with sane partition sizes (~256 MB each) and only needed columns
ddf = dd.read_parquet(
    "s3://lake/event_log",
    columns=["user_id", "event_type", "amount", "ts"],
    blocksize="256MB",
)

# 2. Set the index to the group key ONCE. This is the single expensive shuffle;
#    afterwards Dask knows the data is partitioned by user_id.
ddf = ddf.set_index("user_id", shuffle="tasks")

# 3. Persist the indexed frame so the shuffle is not repeated per feature
ddf = ddf.persist()

# 4. Now multiple groupby-apply features are shuffle-free (already partitioned by user_id)
features = ddf.groupby("user_id").agg(
    txn_count=("amount", "count"),
    total_spend=("amount", "sum"),
    avg_spend=("amount", "mean"),
    last_seen=("ts", "max"),
)

# 5. Custom per-user feature via map_partitions (runs native pandas per partition)
def sessions(pdf):
    pdf = pdf.sort_values("ts")
    gap = pdf["ts"].diff() > "30min"
    return pdf.assign(session=gap.cumsum()).groupby("user_id")["session"].nunique()

session_counts = ddf.map_partitions(sessions)

out = features.join(session_counts.rename("session_count"))
out.to_parquet("s3://lake/user_features", write_index=True)

Step-by-step trace.

Step	Action	Shuffle?
1	read with 256 MB partitions, 4 columns	no (projection pushdown)
2	set_index("user_id")	yes — the one deliberate shuffle
3	persist the indexed frame	no (materialises step 2 once)
4	groupby("user_id").agg	no (already partitioned by key)
5	map_partitions custom feature	no (per-partition native pandas)

The whole strategy hinges on paying for exactly one shuffle. set_index("user_id") repartitions the 400 GB dataset so each partition owns a disjoint range of user_id values; once persisted, every subsequent groupby("user_id") and map_partitions operates within a partition and needs no further data movement. The single-machine OOM is solved because no partition ever holds the whole dataset — each worker holds only its ~256 MB slices.

Output:

Metric	Naive pandas	Dask (indexed)
Fits in memory	no (400 GB > 64 GB)	yes (per-partition)
Shuffles	n/a (crashes)	exactly 1 (set_index)
Groupby features	crash	shuffle-free after index
Rewrite required	none	none (same pandas API)
Cluster	1 machine	16 workers

Why this works — concept by concept:

• Partition sizing — blocksize="256MB" splits the 400 GB log into partitions that comfortably fit a worker, so no single pandas frame is ever the whole dataset. This is the direct fix for the single-machine OOM.
• set_index as the one shuffle — grouping by a non-index column shuffles every time; setting the index to user_id once pays that shuffle a single time and leaves the data physically partitioned by the group key, so later groupbys are local.
• persist after the index — without persist, each downstream feature would re-trigger the expensive set_index shuffle. Persisting materialises the indexed frame in cluster memory so the shuffle happens exactly once.
• map_partitions for custom logic — the session-counting feature is arbitrary pandas that has no DataFrame-API equivalent; map_partitions runs it as native pandas on each already-keyed partition, no rewrite and no boundary.
• Cost — one O(N) shuffle for set_index, then O(N) local aggregation with zero further data movement, all in pure Python with no JVM. The alternative (rewriting in Spark SQL) buys a better optimiser but forces the team off its pandas mental model — the wrong trade when the code already works and just needs to scale.

Python
Topic — data-processing
Data-processing problems on partitioned aggregation

Practice →

ETL Topic — etl ETL problems on out-of-core pandas pipelines

Practice →

---

4. Ray — distributed futures and Ray Data

Distributed futures plus stateful actors, with Ray Data on top — the ML and batch-inference native

The mental model in one line: Ray is a distributed-execution runtime whose two primitives are the task (a stateless remote function call via @ray.remote, returning a future) and the actor (a stateful remote object, e.g. a loaded model), coordinated through a shared distributed object store — and ray data layers a streaming dataset abstraction on top so that batch inference over huge datasets becomes ds.map_batches(model), which is why the modern training, serving, and inference stack is built on Ray rather than on a SQL engine. Every ML platform engineer who has fought to get a GPU model to score a billion rows efficiently has met Ray; the ones who use it well understand that Ray is a compute runtime first and a data library second.

The four axes for Ray.

• Workload shape. ML end to end: distributed training (Ray Train), hyperparameter search (Ray Tune), reinforcement learning (RLlib), serving (Ray Serve), and — the focus here — batch inference over large datasets (Ray Data). Also general distributed Python that needs state.
• Language boundary. None. Ray is pure Python; tasks and actors are your functions and classes. GPUs, PyTorch, and CUDA live directly in your actor code with no serialization gymnastics.
• Scheduler model. Tasks and actors scheduled against a shared object store, with locality-aware placement and heterogeneous resource requests (num_cpus, num_gpus, custom resources). This is the only one of the three with first-class stateful workers and native GPU scheduling.
• Operational surface. A Ray cluster (the head node runs the GCS control plane; worker nodes can be CPU or GPU) with autoscaling, a rich dashboard, and KubeRay for Kubernetes. Heavier than Dask, lighter than a full Spark platform, and uniquely good at mixed CPU+GPU fleets.

Tasks and actors — the two primitives.

• Tasks (@ray.remote on a function). Stateless. f.remote(x) schedules f(x) somewhere on the cluster and returns an ObjectRef (a future). ray.get(ref) blocks for the value. This is the futures model — like Dask's submit, but with a shared object store underneath.
• Actors (@ray.remote on a class). Stateful. A.remote() creates a long-lived worker holding state (e.g. a model in GPU memory); method calls a.method.remote(x) run on that worker. Actors are how you load a model once and reuse it across thousands of batches.
• The object store. A shared-memory store on each node holds large objects (Arrow/NumPy) zero-copy; results move between nodes only when needed. This is what makes passing big arrays between tasks cheap.

Ray Data — the streaming dataset for batch inference.

• Streaming execution. A ray.data.Dataset is processed in blocks that stream through a pipeline of operations, so you can score a dataset far larger than cluster memory without materialising it all at once.
• map_batches. The core operation: apply a function (or a stateful actor class) to fixed-size batches. Pass a class plus num_gpus=1 and Ray runs a pool of GPU actors that load the model once and score batch after batch.
• Why it beats Spark/Dask for inference. The model loads once per actor (not per batch or per partition), GPU scheduling is native, and CPU preprocessing overlaps with GPU scoring because the pipeline streams — three things that are manual or awkward on the other engines.

Common beginner mistakes.

• Loading the model inside the batch function. Reloading a model per batch destroys throughput; load it once in an actor's __init__ and reuse it.
• ray.get in a loop. Calling ray.get on each future serially defeats parallelism; submit all tasks first, then gather, or use ray.wait to stream completions.
• Wrong batch size. Too-small batches underfill the GPU; too-large batches OOM it. Tune batch_size to the model and hardware.
• Treating Ray as a SQL engine. Ray Data is for the compute-heavy last mile (inference, featurisation), not for terabyte joins — hand the heavy relational work to Spark and let Ray do the model pass.

Worked example — the reference groupby aggregation via Ray Data

Detailed explanation. The same per-category revenue rollup, now in Ray Data. This is the least natural of the three for a pure aggregation (Ray is a compute runtime, not a SQL engine), which is itself instructive — it shows where Ray is not the right tool. This completes our three head-to-head implementations.

• Input. The same Parquet transactions table.
• Operation. Group by category; sum amount; count rows.
• Output. A small per-category result.

Question. Write the Ray Data job that produces per-category revenue and counts, and note why this is not Ray's sweet spot.

Input.

Column	Type	Example
category	string	"books"
amount	float64	12.50
user_id	int64	1001

Code.

import ray

ray.init()

ds = ray.data.read_parquet("s3://lake/transactions", columns=["category", "amount"])

# Ray Data supports grouped aggregations directly
from ray.data.aggregate import Sum, Count

rollup = (
    ds.groupby("category")
      .aggregate(
          Sum("amount"),
          Count(),
      )
)

result = rollup.take_all()      # small result -> pull to the driver
for row in sorted(result, key=lambda r: -r["sum(amount)"]):
    print(row)

Step-by-step explanation.

• read_parquet creates a streaming Dataset of blocks (Arrow tables), reading only the two requested columns.
• groupby("category").aggregate(Sum, Count) shuffles by category and combines partial aggregates per block, then merges — the same partial-then-combine pattern as Spark and Dask. Ray Data can do this, but it is a plainer implementation than Catalyst's.
• take_all() pulls the small grouped result to the driver. Because the output is one row per category, this is cheap.
• The tell is that this is unremarkable. A pure relational aggregation runs fine on Ray Data but gains nothing over Spark — no Catalyst, no AQE. Ray earns its place on the next two examples (actors and GPU batch inference), not on this one.

Output.

category	sum(amount)	count()
electronics	4,812,004.50	128,004
books	1,204,551.75	512,880
grocery	990,220.10	1,004,552

Rule of thumb. Ray Data can do groupby aggregations, but a pure SQL/ETL rollup is not where it shines — there is no Catalyst underneath. Use Ray Data when the per-batch work is heavy Python/GPU compute; hand plain relational aggregation to Spark.

Worked example — tasks, actors, and the object store

Detailed explanation. The two primitives that make Ray a compute runtime: stateless tasks for fan-out and stateful actors for reused state. Walk through a Monte-Carlo-style fan-out with tasks and a shared-model pattern with an actor, and see how the object store passes big arrays cheaply.

• Tasks. @ray.remote function; f.remote(x) returns a future; ray.get collects.
• Actors. @ray.remote class; state lives in the worker across calls.
• Object store. ray.put(big_array) stores once; every task that references it reads zero-copy from shared memory.

Question. Fan out a CPU-heavy simulation across the cluster with tasks, and load a model once in an actor to score inputs.

Input.

Primitive	Use
task	stateless parallel simulate(seed)
actor	stateful Model holding weights
object store	share a large lookup array without recopying

Code.

import ray
import numpy as np

ray.init(address="auto")

# --- Tasks: stateless fan-out ---
@ray.remote
def simulate(seed: int, table_ref) -> float:
    table = table_ref                       # zero-copy read from the object store
    rng = np.random.default_rng(seed)
    return float((rng.standard_normal(1_000_000) @ table).mean())

big_table = np.random.rand(1_000_000)
table_ref = ray.put(big_table)              # store ONCE; shared across all tasks

futures = [simulate.remote(s, table_ref) for s in range(1000)]
results = ray.get(futures)                  # gather all 1000 in parallel
print("mean of means:", np.mean(results))

# --- Actor: load a model once, reuse across calls ---
@ray.remote(num_gpus=1)
class Model:
    def __init__(self, path: str):
        import torch
        self.net = torch.jit.load(path).eval().cuda()   # loaded ONCE per actor

    def predict(self, x: np.ndarray) -> np.ndarray:
        import torch
        with torch.no_grad():
            t = torch.as_tensor(x, dtype=torch.float32).cuda()
            return self.net(t).cpu().numpy()

model = Model.remote("model.pt")
pred = ray.get(model.predict.remote(np.random.rand(256, 8).astype("float32")))
print(pred.shape)

Step-by-step explanation.

• ray.put(big_table) stores the array once in the distributed object store and returns a reference. All 1000 simulate tasks read it zero-copy from shared memory instead of each pickling a fresh 1M-element copy — the object store is what makes big-array passing cheap.
• simulate.remote(...) fans out 1000 futures immediately; Ray schedules them across all cores/nodes. ray.get(futures) then blocks once for the whole batch rather than serially — the correct way to gather.
• The Model actor loads the network once in __init__ and holds it in GPU memory. Every predict call reuses that resident model; num_gpus=1 tells Ray to place the actor on a GPU. This "load once, score many" pattern is the foundation of efficient batch inference.
• Tasks vs actors is the key distinction. Tasks are for stateless parallelism; actors are for expensive state (a model, a DB connection pool) that must survive across calls. No other engine here gives you first-class actors.

Output.

Primitive	What ran	Cost saved
ray.put + tasks	1000 sims sharing one array	999 array copies avoided
Model actor	model loaded once, many predicts	per-call model reload avoided
ray.get(list)	one parallel gather	serial round-trips avoided

Rule of thumb. Use ray.put for any large object shared by many tasks, gather with a single ray.get(list_of_futures) rather than in a loop, and put expensive state (models, connections) in an actor so it loads once and is reused across every call.

Worked example — Ray Data batch inference with map_batches and GPU actors

Detailed explanation. This is Ray's headline workload and the reason ML platforms pick it: scoring a dataset larger than memory through a GPU model, with the model loaded once per GPU, CPU preprocessing overlapping GPU compute, and automatic batching. Walk through a full map_batches inference pipeline.

• Read. Stream the dataset in blocks (no full materialisation).
• Preprocess. A stateless CPU map_batches (tokenise / normalise) runs on CPU workers.
• Infer. A stateful actor-class map_batches with num_gpus=1 and concurrency=N runs N GPU actors, each loading the model once.
• Write. Stream predictions out to Parquet.

Question. Build a streaming batch-inference pipeline that preprocesses on CPU and scores on GPU, using a pool of GPU actors.

Input.

Stage	Compute	Ray config
read_parquet	I/O	streaming blocks
preprocess	CPU	stateless fn map_batches
score	GPU	actor class, num_gpus=1, concurrency=4
write	I/O	write_parquet

Code.

import ray
import numpy as np

ray.init(address="auto")

ds = ray.data.read_parquet("s3://lake/embeddings")     # streamed, not fully loaded

# 1. CPU preprocessing — stateless, runs on CPU workers
def normalize(batch: dict) -> dict:
    x = np.stack(batch["embedding"]).astype("float32")
    norms = np.linalg.norm(x, axis=1, keepdims=True)
    batch["embedding"] = (x / np.clip(norms, 1e-6, None))
    return batch

ds = ds.map_batches(normalize, batch_size=8192)         # CPU stage

# 2. GPU inference — stateful actor loads the model once, scores many batches
class Classifier:
    def __init__(self):
        import torch
        self.model = torch.jit.load("classifier.pt").eval().cuda()

    def __call__(self, batch: dict) -> dict:
        import torch
        x = torch.as_tensor(batch["embedding"]).cuda()
        with torch.no_grad():
            logits = self.model(x)
            batch["label"] = logits.argmax(dim=1).cpu().numpy()
        return batch

scored = ds.map_batches(
    Classifier,
    batch_size=4096,
    num_gpus=1,          # each Classifier actor reserves one GPU
    concurrency=4,       # 4 GPU actors process batches in parallel
)

# 3. Stream results out — never materialises the full dataset in memory
scored.write_parquet("s3://lake/predictions")

Step-by-step explanation.

• The dataset streams in blocks, so a 500 GB embeddings table never has to fit in cluster memory — blocks flow through the pipeline and are released after they are written.
• The CPU normalize stage runs as a stateless function on CPU workers. Because Ray Data pipelines are streaming, this CPU work overlaps the GPU inference downstream — while the GPU scores batch N, the CPUs are normalising batch N+1.
• Classifier is a stateful actor class, so passing the class (not an instance) to map_batches with concurrency=4 makes Ray spin up 4 GPU actors, each loading classifier.pt exactly once in __init__. Every batch reuses the resident model — the load cost is paid 4 times total, not once per batch.
• num_gpus=1 lets Ray's scheduler place the actors on GPU nodes in a mixed fleet automatically. You never hand-assign GPUs; you declare the resource and Ray solves placement.
• write_parquet streams output, so the pipeline has bounded memory from read to write regardless of dataset size. This end-to-end streaming with per-actor model reuse is exactly what Spark UDFs and Dask make you build by hand.

Output.

Property	Value
Dataset size vs memory	500 GB streamed on a smaller cluster
Model loads	4 (once per GPU actor)
CPU/GPU overlap	yes (streaming pipeline)
GPU placement	automatic (num_gpus=1)
Peak memory	bounded (block streaming)

Rule of thumb. For GPU batch inference, use a stateful actor class in map_batches with num_gpus=1 and a tuned concurrency, so the model loads once per GPU and CPU preprocessing overlaps GPU scoring. This "load once, stream batches" pattern is Ray's decisive edge over Spark and Dask for inference.

Senior interview question on Ray

A senior interviewer might ask: "You need to run daily batch inference: score a 1-billion-row feature table through a PyTorch model on a fleet of 8 GPUs, then write predictions to Parquet. The data is far larger than cluster RAM and the model takes 30 seconds to load. Design the Ray Data pipeline, explain how you keep all 8 GPUs busy, and how you avoid reloading the model per batch."

Solution Using Ray Data streaming with a GPU actor pool and overlapped preprocessing

import ray
import numpy as np

ray.init(address="auto")

# 1. Stream the billion-row table in blocks; override block size for GPU-sized batches
ds = ray.data.read_parquet(
    "s3://lake/features",
    columns=["id", "features"],
).map_batches(
    lambda b: {                                   # CPU preprocessing, overlaps GPU
        "id": b["id"],
        "x": np.stack(b["features"]).astype("float32"),
    },
    batch_size=16384,
    num_cpus=1,
)

# 2. GPU actor: model loaded ONCE in __init__ (the 30s cost paid 8x total, not per batch)
class Scorer:
    def __init__(self):
        import torch
        self.model = torch.jit.load("model.pt").eval().cuda()

    def __call__(self, batch: dict) -> dict:
        import torch
        x = torch.as_tensor(batch["x"]).cuda()
        with torch.no_grad():
            out = self.model(x).cpu().numpy()
        return {"id": batch["id"], "score": out.ravel()}

# 3. Pool of 8 GPU actors; Ray streams batches to whichever is free
scored = ds.map_batches(
    Scorer,
    batch_size=8192,          # tuned to fill one GPU
    num_gpus=1,               # one GPU per actor
    concurrency=8,            # 8 actors -> all 8 GPUs busy
)

# 4. Stream predictions out with bounded memory
scored.write_parquet("s3://lake/predictions")

Step-by-step trace.

Stage	Resource	Model loads	Memory behaviour
read_parquet	I/O	—	streaming blocks
preprocess (CPU)	1 CPU/task	—	overlaps GPU downstream
score (8 GPU actors)	8 GPUs	8 total (once each)	resident model, reused
write_parquet	I/O	—	bounded (streaming out)

The design keeps all 8 GPUs busy because concurrency=8 creates eight Scorer actors and Ray's streaming scheduler dispatches each ready batch to whichever actor is idle — there is no barrier that makes fast GPUs wait for slow ones. The 30-second model load is paid exactly once per actor (8 times total for the whole run), because the model lives in the actor's __init__ and persists across every batch that actor scores. The billion rows never sit in memory at once: blocks stream from S3, through CPU normalisation, into a GPU, and out to Parquet, so peak memory is a function of block size and concurrency, not dataset size.

Output:

Metric	Naive (UDF-style)	Ray Data actor pool
Model loads	per batch (thousands)	8 (once per GPU)
GPU utilisation	low (load-dominated)	high (score-dominated)
Dataset in memory	must fit	streamed
CPU/GPU overlap	no	yes
Rewrite for scale	large	none (change concurrency)

Why this works — concept by concept:

• Streaming execution — Ray Data processes the billion rows as a stream of blocks, so the dataset never has to fit in cluster memory; peak memory depends on block size and concurrency, not total size. This is the direct answer to "data far larger than RAM."
• Stateful GPU actors — passing the Scorer class with concurrency=8 creates 8 long-lived actors that each load the model once in __init__. The 30-second load is amortised over millions of rows instead of paid per batch — the single biggest throughput lever.
• Declarative GPU placement — num_gpus=1 per actor lets Ray's scheduler pin each actor to a distinct GPU on the fleet automatically; you never hand-assign device IDs, and autoscaling can add GPU nodes if configured.
• CPU/GPU overlap — the preprocessing map_batches runs on CPUs concurrently with GPU scoring because the pipeline streams, so GPUs are fed continuously instead of stalling while CPUs prepare the next batch.
• Cost — the run costs 8 model loads plus streaming I/O, and scaling to 16 GPUs is a one-word change (concurrency=16). The eliminated cost is the naive pattern's per-batch model reload, which would make load time — not compute — dominate the bill. Net: GPU-bound throughput at O(actors) load cost, not O(batches).

Python
Topic — data-processing
Data-processing problems on batch inference pipelines

Practice →

Design Topic — design Design problems on ML inference systems

Practice →

---

5. Head-to-head — decision matrix and interview signals

Match the engine to the workload shape — there is no universal winner, only lane winners

The mental model in one line: the ray vs dask vs spark decision is not a ranking but a routing — Spark wins the SQL/ETL-over-lake-tables lane on the strength of Catalyst and a decade of hardening, Dask wins the scale-my-pandas-without-a-rewrite lane on the strength of API fidelity and zero JVM, and Ray wins the ML/training/batch-inference lane on the strength of futures, actors, native GPU scheduling, and Ray Data's streaming — and the senior answer names the lane before it names the engine. Every distributed-compute interview eventually asks you to compare all three; the candidates who route by workload shape score highest, the ones who declare a single "best" score lowest.

SQL/ETL vs ML/inference — the primary split.

• SQL/ETL (scan, join, aggregate, write lake tables). Spark first, always, at TB–PB scale — Catalyst optimises the plan and AQE fixes skew at runtime. Dask is a viable second for pandas-shaped ETL at GB–low-TB scale when the team is pure Python. Ray is the wrong tool for pure relational ETL.
• ML / batch inference (featurise, train, score a model). Ray first — Ray Data streaming + GPU actors are purpose-built for it. Dask can do CPU-bound scikit workflows. Spark can bolt on inference via Pandas UDFs but fights you on GPUs.
• The mixed reality. Most platforms do both. The mature pattern is Spark for the warehouse feed and Ray for the inference tail, glued by a table contract — not one engine forced to do everything.

Cost and operations — the axes finance and on-call feel.

• Operational weight. Dask is the lightest (pure-Python cluster from a notebook). Ray is medium (a richer control plane, GPU-aware, KubeRay). Spark is the heaviest to self-manage but has the deepest managed ecosystem (Databricks, EMR, Dataproc) that absorbs the weight for money.
• Compute efficiency. For SQL/ETL, Spark's Tungsten/whole-stage codegen makes it very cost-efficient per row. For Python/NumPy hot paths, Dask and Ray avoid the JVM boundary tax. For GPU inference, Ray's model-reuse and streaming maximise GPU utilisation, which is where the money is.
• Hiring and mental model. A pure-Python team is productive on Dask and Ray on day one; PySpark demands understanding the JVM boundary, the DataFrame-as-query-builder model, and shuffle tuning. That learning curve is a real (if unpriced) cost.

The scheduler comparison — why each feels the way it does.

• Spark: stages split by shuffle. Bulk-synchronous; great for wide relational operations; every wide op is a stage boundary. You reason in stages and shuffles.
• Dask: fine-grained task graph. Millions of small tasks, dynamic; flexible for custom Python; less automatic optimisation, so you structure the pipeline. You reason in partitions and the graph.
• Ray: tasks + actors on an object store. The most general; first-class state and GPUs; you compose your own dataflow. You reason in futures, actors, and resources.

Common beginner mistakes (in the comparison itself).

• Declaring a single winner. "Spark is best" or "Ray is the future" both fail the interview; the answer is lane-dependent.
• Ignoring interoperability. RayDP (Spark on Ray) and Dask-on-Ray exist; the engines are not mutually exclusive, and pretending they are misses the mature multi-engine architecture.
• Forgetting the JVM boundary in cost. Comparing raw benchmarks without noting that PySpark UDF-heavy code pays a boundary tax gives a misleading picture.
• Conflating Dask and Ray. Both are pure-Python and both have futures, but Ray adds stateful actors and native GPU scheduling that make it the ML tool; Dask is the pandas-scaling tool.

Worked example — the full decision matrix

Detailed explanation. The artifact to memorise: a matrix scoring all three engines across the axes interviewers probe. Build it once, carry it into every interview, and read the winners off the columns.

• Rows. The axes: SQL/ETL, ML/inference, GPU, language boundary, ops weight, scale ceiling.
• Columns. Spark, Dask, Ray.
• Cells. Which engine wins the row and why.

Question. Fill in the decision matrix and state the one-line rule each row implies.

Input.

Axis	What it measures
SQL/ETL at scale	lake-table join/aggregate performance
ML / batch inference	model training + scoring throughput
GPU support	native GPU scheduling
Language boundary	JVM tax on Python logic
Ops weight	effort to run the cluster

Code.

MATRIX = {
    #                      Spark            Dask             Ray
    "sql_etl_scale":     ("best",          "ok (pandas)",   "poor"),
    "ml_batch_inference":("bolt-on (UDF)", "cpu only",      "best"),
    "gpu_support":       ("awkward",       "manual",        "native"),
    "language_boundary": ("JVM tax",       "none",          "none"),
    "ops_weight":        ("heavy/managed", "lightest",      "medium"),
    "scale_ceiling":     ("PB",            "TB",            "PB (compute)"),
}

def winner(axis: str) -> str:
    ranking = {"best": 3, "native": 3, "none": 2, "ok (pandas)": 2,
               "lightest": 3, "medium": 2, "PB": 3, "PB (compute)": 3}
    spark, dask, ray = MATRIX[axis]
    scores = {"spark": ranking.get(spark, 1),
              "dask": ranking.get(dask, 1),
              "ray": ranking.get(ray, 1)}
    return max(scores, key=scores.get)

for axis in MATRIX:
    print(f"{axis:22} -> {winner(axis)}")

Step-by-step explanation.

• SQL/ETL at scale → Spark. Catalyst + AQE + Tungsten make lake-table joins and aggregates fastest and cheapest per row; nothing else is close at PB scale.
• ML/batch inference → Ray. Ray Data streaming + GPU actors + model reuse win decisively; Spark's UDF path and Dask's CPU-only story are both weaker.
• Language boundary → Dask/Ray (tie). Both are pure Python with no JVM tax; Spark pays the boundary whenever your Python logic is on the hot path.
• Ops weight → Dask. The lightest to stand up; Ray is medium; self-managed Spark is heaviest (though managed services trade that weight for cost).
• The matrix routes, it does not rank. Read down a column and you see each engine's shape; read across a row and you see the lane winner. That two-way read is the whole skill.

Output.

Axis	Winner	Rule it implies
SQL/ETL at scale	Spark	lake ETL → Spark
ML/batch inference	Ray	inference/training → Ray
GPU support	Ray	GPUs → Ray
Language boundary	Dask/Ray	pure-Python hot path → not Spark
Ops weight	Dask	lightest cluster → Dask

Rule of thumb. Carry the matrix into the interview and read the winner off the row that matches the workload. The moment you can say "this workload sits on the ML/inference row, so Ray" you have given the senior answer.

Worked example — the same pipeline three ways, side by side

Detailed explanation. Nothing clarifies the engines like one task in all three dialects. We have built the per-category rollup in each section; here they sit together so the differences in feel are obvious at a glance.

• Spark. DataFrame query, JVM execution, Catalyst plan.
• Dask. pandas API, task graph, pure Python.
• Ray. Dataset + aggregate, streaming, compute-runtime.

Question. Place the three implementations side by side and name the one-line character of each.

Input.

Engine	Import	Action verb
Spark	pyspark.sql	write / show
Dask	dask.dataframe	compute
Ray	ray.data	take_all / write

Code.

# ---- Spark (JVM SQL engine) ----
from pyspark.sql import functions as F
(spark.read.parquet("s3://lake/transactions")
      .groupBy("category")
      .agg(F.sum("amount").alias("revenue"), F.count("*").alias("txn_count"))
      .write.mode("overwrite").parquet("out_spark"))

# ---- Dask (pandas at scale) ----
import dask.dataframe as dd
(dd.read_parquet("s3://lake/transactions", columns=["category", "amount"])
   .groupby("category")["amount"].sum()
   .compute()
   .to_frame("revenue").to_parquet("out_dask.parquet"))

# ---- Ray (compute runtime) ----
import ray
from ray.data.aggregate import Sum, Count
(ray.data.read_parquet("s3://lake/transactions", columns=["category", "amount"])
    .groupby("category").aggregate(Sum("amount"), Count())
    .write_parquet("out_ray"))

Step-by-step explanation.

• Spark reads like SQL because it is a query builder over a JVM engine; the .write is the action that triggers the optimised plan. This is the most declarative of the three.
• Dask reads like pandas because each partition is pandas; .compute() is the action. A pandas user recognises every line — that familiarity is the migration story.
• Ray reads like a data pipeline built from explicit operators; .write_parquet streams the result. It is fine here but unremarkable — Ray's value is elsewhere (actors, GPUs), not in relational rollups.
• Same result, three characters. Declarative-SQL (Spark), scaled-pandas (Dask), streaming-compute (Ray). The syntax differences are shallow; the fit differences (which we saw in the interview solutions) are deep.

Output.

Engine	Character	Best when
Spark	declarative SQL over JVM	SQL/ETL at scale
Dask	scaled pandas, pure Python	pandas outgrew RAM
Ray	streaming compute runtime	ML / GPU inference

Rule of thumb. For a plain rollup, all three are a few lines and all three work. The engine choice is decided not by this easy case but by the hard cases — skewed joins (Spark), no-rewrite pandas scaling (Dask), GPU batch inference (Ray).

Senior interview question on engine selection for a mixed platform

A senior interviewer might ask: "You are the platform lead for a company that runs (a) a 20 TB nightly lakehouse ETL, (b) an analyst team living in pandas on ~500 GB datasets, and (c) a daily GPU batch-inference job over 1 billion rows. Leadership wants 'one engine to reduce complexity.' Do you comply, and if not, how do you justify the engine set you actually recommend?"

Solution Using a lane-based engine set with table-contract handoffs

# Recommended engine set — one per lane, glued by governed tables.
PLATFORM = {
    "lakehouse_etl_20tb": {
        "engine": "spark",
        "why": "Catalyst + AQE + Delta; PB-ready; cheapest per row for SQL/ETL",
        "handoff": "writes governed Delta tables (the contract)",
    },
    "analyst_pandas_500gb": {
        "engine": "dask",
        "why": "scale the existing pandas code with no rewrite; pure Python; lightest ops",
        "handoff": "reads the same Delta/Parquet tables Spark writes",
    },
    "gpu_batch_inference_1b": {
        "engine": "ray",
        "why": "Ray Data streaming + GPU actor pool; model loads once; GPU-bound throughput",
        "handoff": "reads the feature table Spark writes; writes a score table",
    },
}

def justify(one_engine_mandate: bool) -> str:
    if not one_engine_mandate:
        return "Use the lane-based set: Spark + Dask + Ray, glued by tables."
    # If forced to one engine, the least-bad single choice and its costs:
    return ("If truly forced to one: Spark. It covers ETL natively, can run "
            "analyst workloads (with a JVM-boundary tax on pandas), and can bolt "
            "on inference via Pandas UDFs (losing native GPU efficiency). "
            "The hidden cost is analyst friction + degraded GPU utilisation.")

for lane, spec in PLATFORM.items():
    print(lane, "->", spec["engine"], "::", spec["why"])
print(justify(one_engine_mandate=False))

Step-by-step trace.

Lane	Engine	Handoff artifact	Why not a different engine
20 TB ETL	Spark	governed Delta tables	Dask lacks Catalyst; Ray is not a SQL engine
500 GB pandas	Dask	reads Spark's tables	Spark taxes pandas UDFs; Ray is overkill for CPU analytics
1B GPU inference	Ray	reads feature table, writes scores	Spark UDF fights GPUs; Dask is CPU-oriented

The recommendation resists the "one engine" mandate with a cost argument, not a preference. Each lane sits on a different axis of the decision matrix, so a single engine necessarily runs two of the three lanes against its grain: forcing everything onto Spark taxes the analysts' pandas code across the JVM boundary and cripples GPU utilisation on inference; forcing everything onto Ray throws away Catalyst for the 20 TB ETL. The three engines couple only through governed tables, so "three engines" does not mean "three tangled runtimes" — it means three tools each in its lane, meeting at schema contracts. The honest cost (a second and third cluster to operate) is named explicitly and weighed against the productivity and GPU-efficiency it buys.

Output:

Approach	ETL cost	Analyst productivity	Inference GPU util	Ops surface
One engine (Spark forced)	low	low (JVM UDF tax)	poor	one cluster
One engine (Ray forced)	high (no Catalyst)	medium	best	one cluster
Lane-based (Spark+Dask+Ray)	low	high	best	three clusters

Why this works — concept by concept:

• Lane-based routing — each of the three workloads sits on a different decision-matrix axis (SQL/ETL, pandas-analytics, ML/inference), so matching an engine per lane keeps every workload in an engine's design centre instead of fighting one engine on two fronts.
• Table-contract coupling — the engines meet only at governed Delta/Parquet tables, so "three engines" is loose coupling by schema, not a tangled shared runtime; any engine can be swapped without touching the others.
• Resisting the one-engine mandate with cost, not taste — the argument against consolidation is quantified (JVM-boundary tax on analysts, degraded GPU utilisation), which is what makes it a senior answer rather than a preference.
• Naming the least-bad single choice — if genuinely forced to one engine, Spark is the least-bad because it can technically cover all three lanes; stating this and its hidden costs shows you engaged with the constraint instead of dodging it.
• Cost — the recommended set costs additional operational surface (two extra clusters) but buys native performance in every lane; the alternative saves ops surface at the price of analyst friction and wasted GPU spend. The trade is only worth it because all three lanes are large — if any lane were tiny, you would fold it into a neighbour's engine.

Design
Topic — design
Design problems on multi-engine data platforms

Practice →

Perf
Topic — optimization
Optimization problems on engine and cost trade-offs

Practice →

---

Cheat sheet — distributed compute engine recipes

• Which engine when. SQL/ETL over lake tables at TB–PB scale → Spark (Catalyst, AQE, Delta). Existing pandas/NumPy workflow that outgrew one machine and must not be rewritten → Dask (same API, pure Python, lightest ops). Distributed training, hyperparameter search, or GPU batch inference → Ray (futures + actors + Ray Data streaming). Mixed platform → run the primary engine per lane and glue with a governed table contract; do not force one engine to serve every lane.
• The JVM boundary rule (Spark). pyspark drives a JVM engine; a row-at-a-time Python UDF serializes every row across the boundary. Keep the hot path in built-in SQL functions or Arrow-based Pandas UDFs (@pandas_udf); reserve raw Python UDFs for logic that genuinely cannot be expressed otherwise. When Python/NumPy must be on the hot path, that boundary tax is the argument for Dask or Ray.
• PySpark ETL template. spark.read.format("delta").load(...) → filter on the partition column (partition pruning) → .select(needed_cols) (projection pushdown) → .join(F.broadcast(small_dim), key) (broadcast, no shuffle) → .groupBy(...).agg(...) → single .write. Confirm the plan with .explain(mode="formatted") and look for BroadcastHashJoin, not SortMergeJoin.
• Spark shuffle + skew. A shuffle is any wide dependency (join/groupBy/distinct/repartition). Reduce it by broadcasting small sides and pre-partitioning by key. For skew, enable spark.sql.adaptive.enabled=true + spark.sql.adaptive.skewJoin.enabled=true first; fall back to salting only the handful of pathologically hot keys, replicating them on the small side and aggregating the salt away afterward.
• Dask dataframe template. Client(...) → dd.read_parquet(path, columns=[...], blocksize="256MB") → set_index(group_key) once (the one deliberate shuffle) → .persist() → many groupby(group_key).agg(...) / map_partitions(fn) that are now shuffle-free → single .compute() or .to_parquet(). Target ~100–256 MB partitions; use .visualize() and the dashboard when a job is slow.
• Dask futures template. For embarrassingly-parallel custom Python: futures = client.map(fn, items) then stream results with for f in as_completed(futures): use(f.result()). This is Dask's Ray-like face; it lacks stateful actors and native GPU scheduling, which is the boundary where Ray wins.
• Ray tasks + actors. Stateless fan-out: @ray.remote def f(...), submit all with [f.remote(x) for x in xs], gather once with ray.get(list). Share big objects with ray.put(obj) (zero-copy). Expensive state (a model, a pool): @ray.remote class A — load once in __init__, reuse across method calls. Never ray.get in a loop; never reload a model per call.
• Ray Data batch inference template. ray.data.read_parquet(...) → CPU map_batches(preprocess_fn, batch_size=...) → GPU map_batches(ScorerClass, batch_size=..., num_gpus=1, concurrency=N) → write_parquet(...). The model loads once per GPU actor (N loads total, not per batch); streaming gives bounded memory and CPU/GPU overlap. This is Ray's decisive edge for inference.
• Partition / batch sizing. Spark: tune spark.sql.shuffle.partitions or trust AQE coalescing; aim for ~128 MB shuffle partitions. Dask: ~100–256 MB per partition; repartition if the count explodes or collapses. Ray: batch_size sized to fill the GPU (too small underfills, too large OOMs); concurrency = number of GPUs/actors.
• The scheduler mental models. Spark = stages split by shuffle (reason in stages). Dask = fine-grained task graph over pandas partitions (reason in partitions and the DAG). Ray = tasks + actors on a shared object store (reason in futures, actors, resources). Match the workload to the model that fits it: relational → Spark, pandas → Dask, stateful/GPU compute → Ray.
• Interoperability. They are not mutually exclusive: RayDP runs Spark on Ray; Dask-on-Ray uses Ray as the scheduler; a Spark DataFrame can be handed to Ray Data for the inference tail. The mature architecture is often multi-engine glued at table or dataset boundaries, not a single winner.
• The decision matrix (memorise). SQL/ETL at scale → Spark. ML/training/GPU inference → Ray. Scale-my-pandas-no-rewrite → Dask. Language boundary: Spark pays a JVM tax on Python hot paths; Dask and Ray do not. Ops weight: Dask lightest, Ray medium, Spark heaviest (or managed). No universal winner — route by workload shape, name the lane before the engine.

Frequently asked questions

Ray vs Dask vs Spark — what is the one-sentence difference?

Spark is a mature JVM SQL engine (driven from Python via pyspark) that turns your DataFrame/SQL into an optimised plan of stages and is the default for lakehouse SQL/ETL at scale; Dask is a pure-Python scheduler that parallelises the pandas and NumPy APIs across a task graph, making it the lightest way to scale existing pandas code past one machine's memory; and Ray is a pure-Python distributed-execution runtime built on futures and stateful actors with a ray data layer for streaming batch inference, making it the native choice for ML, training, and GPU workloads. The routing rule is simple: SQL/ETL → Spark, scale-my-pandas → Dask, ML/inference → Ray. They interoperate, so a real platform often uses two of them glued by a table contract rather than crowning a single winner.

When should I pick PySpark over Dask?

Pick PySpark when the workload is SQL/ETL over lake tables (Parquet/Delta/Iceberg) at terabyte-to-petabyte scale, where Catalyst's query optimisation, Adaptive Query Execution's runtime skew handling, and a decade of production hardening matter more than the JVM boundary — big joins, wide aggregations, and warehouse feeds are Spark's home turf. Pick Dask when you already have a pandas/NumPy/scikit workflow that simply outgrew a single machine's RAM and you want to scale it without a rewrite and without a JVM, especially for a pure-Python team doing GB-to-low-TB analytics. The deciding questions are: is the logic naturally relational (favor Spark) or naturally pandas (favor Dask), and is the data volume so large that Catalyst's optimiser pays for the JVM boundary (favor Spark)?

Is Ray a replacement for Spark?

No — they solve different problems and largely occupy different lanes. Spark is a distributed SQL engine optimised for relational ETL over lake tables; Ray is a general distributed-compute runtime optimised for ML — distributed training, hyperparameter tuning, reinforcement learning, serving, and batch inference over huge datasets via Ray Data. Ray is not a great pure SQL/ETL engine (it has no Catalyst), and Spark is not a great GPU-inference engine (its Python UDF path and GPU scheduling fight you). The mature pattern is complementary, not substitutive: Spark writes the governed feature table and Ray Data reads it to run the GPU model pass. Projects like RayDP even run Spark on Ray, which underscores that they compose rather than compete.

Which engine is best for batch inference?

Ray, specifically Ray Data with map_batches, is purpose-built for batch inference and generally wins decisively. The reasons are structural: you pass a stateful actor class so the model loads once per GPU (not once per batch), num_gpus=1 gives native GPU scheduling on a mixed fleet, concurrency=N runs N GPU actors so all your GPUs stay busy, and the streaming execution overlaps CPU preprocessing with GPU scoring while keeping memory bounded regardless of dataset size. Spark can bolt on inference through Pandas/Arrow UDFs but has no first-class GPU actor and makes model-lifecycle and GPU placement manual; Dask can do CPU-bound scoring but is not GPU-native. If the job is "score a billion rows through a GPU model," Ray Data is the tool.

Can Ray, Dask, and Spark interoperate?

Yes, and mature platforms rely on it. RayDP runs Apache Spark on top of a Ray cluster, so you get Spark's DataFrame API and Ray's actor/GPU ecosystem in one runtime; Dask-on-Ray swaps Dask's scheduler for Ray's so Dask collections execute on a Ray cluster; and at the data level a Spark DataFrame or Delta table can be read directly by Ray Data for the inference tail, or a Dask DataFrame can hand off to Ray. Even without those bridges, the loosest and most common interop is a table contract: one engine writes a governed Parquet/Delta table and another reads it. So the practical answer to "which one" is often "the right one per lane, connected at a dataset boundary," not a single engine for everything.

Which engine is easiest for a pure-Python team with no JVM experience?

Dask is usually the gentlest on-ramp because its high-level collections mirror the pandas and NumPy APIs a Python team already knows — often the only change is swapping import pandas as pd for import dask.dataframe as dd and adding a .compute(). Ray is also pure Python and very approachable for anyone comfortable with functions, classes, and async-style thinking, and it becomes essential the moment ML, GPUs, or stateful workers enter the picture. PySpark is the steepest for a JVM-naive team: you must internalise that it drives a JVM engine, that the DataFrame API is a query builder (not a Python loop), and that shuffles and the serialization boundary dominate performance. For pandas-shaped analytics, start with Dask; for ML and inference, reach for Ray; adopt PySpark when the SQL/ETL scale genuinely demands Catalyst.

Practice on PipeCode

• Drill the data-processing practice library → for the partitioned-aggregation, groupby, and batch-processing problems that Spark, Dask, and Ray workloads live and die on.
• Rehearse on the ETL practice library → for the incremental-load, out-of-core, and lakehouse-feed patterns that decide when Spark beats Dask and vice versa.
• Sharpen the systems axis with the design practice library → for the multi-engine platform, batch-inference topology, and engine-selection trade-off questions senior interviewers open with.
• Tune the cost axis on the optimization practice library → for the shuffle, skew, partition-sizing, and GPU-utilisation problems that separate a fluent engine answer from a stumbling one.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the three-engine decision matrix against real graded inputs.

Lock in distributed-engine muscle memory

Docs explain engines. PipeCode drills explain the decision — when Spark's Catalyst earns the JVM boundary, when Dask scales pandas without a rewrite, when Ray Data's GPU actors own batch inference, and when a mixed platform beats forcing one engine everywhere. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs Python-first data engineers actually face.

Practice data-processing problems →
Practice design problems →