Distributed Data Processing
Splitting one computation across many machines, and the three things that buys you — memory, disk bandwidth and cores — against the one thing it costs: a network in the middle of your query.
Who needs this, what one row is, and why the obvious build breaks
Every lesson starts from the consumer, because designing from the source outward is this domain's characteristic mistake.
At what point does a computation stop fitting on one machine, and what does moving it to a cluster actually buy?
The models downstream of this job and the people waiting for them: a nightly fct_orders rebuild that has to land before the European morning, a feature table a training run depends on, a mart that eighty dashboards read. None of them care how many machines were involved. They care that the table is complete and that it appeared before they needed it.
The unit of distributed processing is the partition — a bounded slice of rows that one task can process alone, in memory, without talking to any other task. Everything in this module is a consequence of that unit: how many there are, how evenly they are filled, and what has to happen when a computation needs rows from more than one of them.
Load the table into one process and compute. A single machine with a lot of memory, a columnar engine and a well-written query goes remarkably far — much further than the tooling conversation suggests (DuckDB Concepts) — and it has no network, no scheduler, no serialization and no partial failure. For a very large share of real analytical work this is the correct answer and reaching for a cluster is the mistake.
The dataset stops fitting in memory. The process does not fail cleanly: it starts spilling, then swapping, and a job that took minutes takes hours while every profile shows the CPU idle and the disk saturated (Disk and Storage: Latency, Throughput, IOPS and the fsync Tax).
- The dataset stops fitting in memory. The process does not fail cleanly: it starts spilling, then swapping, and a job that took minutes takes hours while every profile shows the CPU idle and the disk saturated (Disk and Storage: Latency, Throughput, IOPS and the fsync Tax).
- The job takes six hours and dies at hour five. There is no checkpoint, no partial result and no way to resume, so the only option is to start again and hope (Checkpointing).
- A single machine can only read so many bytes per second from storage. Once the query is bound by that, a faster CPU changes nothing and the only remaining lever is reading from more machines at once (Object Storage as Data Infrastructure).
- The input arrives as ten thousand files in object storage. One process opens them sequentially; the per-object round trip dominates, and the job is limited by request latency rather than by data volume (File Size and the Small-Files Problem).
What is actually happening
- A distributed engine splits the input into partitions, then ships the code to the data rather than the data to the code. Each task runs the same closure over a different partition, on a machine that can read that partition cheaply, and produces a partial result.
- Anything that can be computed from one partition alone — a filter, a projection, a per-row expression — needs no coordination at all and scales almost linearly with the number of cores you point at it (Narrow and Wide Transformations).
- Anything that needs rows grouped by a key that is spread across partitions — a
GROUP BY, a join, aDISTINCT, a global sort — requires the data to be redistributed across the network first. That redistribution is the shuffle, and it is the boundary between what is cheap and what is not (The Shuffle). - Partial failure is normal rather than exceptional. With enough machines, something is always dying, so the engine tracks how each partition was derived and recomputes lost work from its inputs. That only works if a task is a deterministic function of its input, which is why non-deterministic transformations are a correctness problem here and not a style one (Determinism: Same Input, Same Output?).
- The scheduler is therefore doing two jobs at once: cutting the computation into tasks small enough to retry and large enough to be worth scheduling, and placing those tasks where their input already is (Stages and Tasks).
Four reasons to distribute, only one of which is volume
The conversation about distributed processing usually starts with size, which is the least useful place to start. A hundred gigabytes of Parquet with a selective predicate and two projected columns is a small job on one machine; a hundred gigabytes joined to itself on a high-cardinality key is not a small job anywhere.
It is more productive to name the specific constraint that a single process is failing to satisfy, because each one has a different remedy and only one of them requires a cluster. Memory can be bought. Restartability can be designed. Read bandwidth genuinely requires more machines. Volume, on its own, requires nothing at all.
The decision below is worth making explicitly and writing down, because the alternative is a platform whose compute layer was chosen by whichever engineer joined most recently.
Which of these is true today, and which will be true in a year?
when The data fits in memory on a machine you can rent, the query returns in acceptable time, and the job is short enough that restarting it is not painful.
cost Nothing. A single-node columnar engine on a large instance is the cheapest and most debuggable answer available, and it stops being the answer at a predictable point (DuckDB Concepts).
when The working set exceeds what one machine holds — usually because of a join or a wide aggregation rather than the raw input.
cost A cluster, or a job restructured to stream through the data in bounded chunks. The second is often possible and rarely attempted.
when The job is bound by how fast bytes arrive from object storage, and the CPU is idle waiting.
cost More machines reading in parallel, which genuinely helps here — this is the constraint distribution is best at (Object Storage).
when The run is long enough that a mid-run failure is unacceptable, so the work must be divided into units that can be retried independently.
cost The task and lineage machinery of a distributed engine, or a hand-rolled chunking scheme that you then have to maintain (Partial Failure).
when The workload is bursty — a nightly rebuild and nothing for twenty hours — and holding a machine large enough for the peak wastes the rest of the day.
cost Startup latency per run, and a cost model where a badly shaped job scales its bill with its inefficiency (Separating Storage from Compute).
Ship the code to the data
The organising idea is old and still the whole trick: it is cheaper to send a few kilobytes of compiled plan to the machine holding the data than to send the data to the machine holding the plan. Every distributed data engine is an elaboration of that sentence, and every performance problem in one is a place where data had to move anyway.
The pipeline below is what a job actually does, with the promise each phase makes. Read the guarantees column: almost every phase promises something local and conditional, and the only global promise — that all rows with the same key are together — is bought by the one phase that uses the network.
Note where correctness is and is not addressed. No phase in this pipeline checks that the answer is right. The engine is a very sophisticated machine for executing exactly what you declared, including when what you declared is at the wrong grain (Grain: What Does One Row Represent?).
- 1Plan
Resolves the query against the catalog, rewrites it, and chooses physical operators and a join strategy.
guarantees The plan is semantically equivalent to what you wrote — under the engine's assumptions about your functions being deterministic and side-effect free.
fails by Choosing a plan from stale or missing statistics, most visibly by not broadcasting a side that would have fitted (Query Optimizers).
- 2Split
Divides the input into partitions, usually from file and row-group boundaries.
guarantees Every input row belongs to exactly one partition, and each partition is independently readable.
fails by Producing far too few partitions from a handful of large non-splittable files, so most of the cluster has nothing to do (Partitions: the Unit of Parallelism).
- 3Local compute
Runs filters, projections and per-row expressions inside each task, with no coordination.
guarantees Linear scaling with available cores, as long as each partition fits in the memory the task was given.
fails by Spilling to local disk when a partition does not fit, which turns a CPU-bound stage into a disk-bound one (Disk and Storage: Latency, Throughput, IOPS and the fsync Tax).
- 4Shuffle
Repartitions rows by key across the network so that grouping and joining become local problems.
guarantees All rows sharing a key land in the same output partition, for a fixed partition count and a deterministic partitioner.
fails by Being the barrier the whole stage waits on, and by concentrating a hot key into one partition (The Shuffle).
- 5Reduce
Aggregates or joins within each redistributed partition.
guarantees Correct per-key results, given that the shuffle placed every row of that key here.
fails by Running out of memory on the one partition that received a dominant key (Data Skew).
- 6Commit
Publishes the output files and makes them visible to readers.
guarantees Atomic visibility only if the sink provides it — a table format with a metadata swap does; a directory of files written by many tasks does not.
fails by Leaving partially written output visible when a job fails between task completion and commit (Atomic Publish).
Only one phase uses the network for data, and it is the phase every performance conversation ends up at. The rest of this module is largely about arranging for it to move less.
What the cluster is actually spending its time on
A distributed job spends its wall clock in a small number of places, and the ordering between them is stable enough to reason about even though the magnitudes never are. Knowing the ordering is what lets you predict which change will matter before you make it.
The weights below are relative and unitless — they establish an order, not a measurement, and any specific job will differ. What transfers is the shape: the two largest terms are decided before the job starts, by how much data the layout forces it to read and by how much the query shape forces it to move.
The last driver is the one people optimise first, because it is the one that appears in a configuration file. Executor count and instance type are the easiest knobs to turn and among the least effective, which is a reliable property of easy knobs.
- Reading less is always better than processing faster, because bytes never read cost nothing at every subsequent stage.
- Moving less is the second lever, and it is a property of the query plan rather than the cluster (Narrow and Wide Transformations).
- Evening out the work is the third, and it is the one that adding machines cannot substitute for (Data Skew).
Decided by partition pruning, column projection and file layout — that is, by decisions made long before the job ran (Partition Pruning).
Driven by the number and shape of wide operations. A join at the wrong grain moves far more than the input contains.
Grows directly with skew. A cluster waiting on one task is paying for every core it is not using (Straggler Tasks).
Appears when partitions do not fit the memory their task was given, and turns compute into I/O (Memory Pressure, Swap and the OOM Killer).
Per-task fixed cost. Invisible with thousands of rows per task, dominant with a hundred — which is what over-partitioning produces.
Fixed per run and independent of data size, which is why frequent tiny jobs are inefficient in a way large ones are not.
Relative weights, not measurements. The ordering is what transfers between platforms; the magnitudes do not, and this domain never publishes a price.
Relative weights for a typical shuffle-heavy batch job, shown to fix an ordering rather than to predict a runtime. The teaching is that the top two are set by layout and query shape, and the bottom two are the ones with configuration flags attached.
How to build it
Most important first.
- Establish the single-machine baseline before distributing anything. If a columnar engine on one large instance answers the query, the cluster buys you operational complexity and nothing else (Measure Before You Optimize).
- Reduce before you redistribute. Filters, projections and partial aggregations applied before a shuffle shrink the thing that has to cross the network, and that ordering is the highest-leverage decision in most jobs (Predicate Pushdown).
- Make the layout do the work the engine otherwise has to. Partition pruning and column pruning mean fewer bytes ever enter the job, which is strictly better than processing them quickly (Partition Pruning).
- Write the job so that re-running any bounded range of it is safe. Distributed jobs fail in the middle, and the difference between a ten-minute recovery and a data incident is whether the write replaces a partition or appends to it (Idempotent Data Pipelines).
- Keep the transformations pure. A task that reads the wall clock, a mutable dimension or a random seed produces different answers on retry, and the engine will retry (Reprocessing vs Retrying).
What this actually promises
Naming the guarantee you do not have is worth more than naming the one you do — everything downstream inherits the weakest promise in the chain.
- The engine guarantees that, given deterministic tasks and a source it can re-read, the declared plan eventually completes despite individual machine failures. It does not guarantee that it completes within your schedule.
- It guarantees no ordering of output rows unless you asked for one. Output file boundaries, file counts and row order within a file are implementation details that change between runs and between versions (Atomic Publish).
- It guarantees nothing about correctness. A distributed engine will execute a wrong join across a thousand cores and return the wrong answer faster than a single machine could (The Pipeline Succeeded. The Data Is Wrong.).
- Task-level retries mean a task's side effects may happen more than once. Only the final commit protocol makes the *published* result appear once, and only for sinks that support it (Exactly-Once: Input Consumption, State Update, Output Write).
Can I trust it?
A green pipeline is evidence that code ran. These four fields are the evidence that the data is right.
- Assert row-count conservation across the stages you expect to conserve it. A filter should reduce rows, a projection should not change them, and a join should be checked against an expected fan-out — a join that silently multiplies rows is the most common way a distributed transformation produces a confidently wrong number (Grain: What Does One Row Represent?).
- Reconcile the job's output aggregate against the source for a closed period. This is the only check that observes the whole computation rather than one stage of it (Reconciliation).
- Both miss the failure that distribution specifically introduces: a non-deterministic task that was retried, so the published result is internally consistent, reproducible-looking and different from what a re-run produces.
- Distribution buys throughput, not latency. It shortens a job that processes a lot of data; it does nothing for the fixed costs of starting a cluster, planning a query and committing an output, which for small inputs can dominate the whole run.
- That fixed cost sets a floor on how often a batch job can usefully run. A pipeline scheduled more frequently than its own startup and commit overhead spends most of its life on ceremony (Cost vs Freshness).
- A consumer of a distributed batch job experiences freshness as the schedule interval plus the run duration plus the commit — and the run duration is set by the slowest task, not the average one (Straggler Tasks).
- A column added upstream costs a distributed job nothing if it projects explicitly and everything if it does not —
SELECT *widens every partition, every shuffle and every spill at once (Projection Pushdown). - A type change upstream usually surfaces as a cast that produces nulls rather than an error, and the job succeeds at full parallelism while producing an empty measure (Nullability & Defaults).
- Changing the number of shuffle partitions changes which key lands where. Nothing about the result changes, but the output file layout does, which matters to every downstream reader that depended on it (Bucketing).
- Within a run, the engine recovers by recomputing lost partitions from their inputs. That recovery is free to you and invisible — right up to the point where the lost partition was the expensive one, and its recomputation extends the stage.
- Across runs, recovery is a data problem rather than an engine one. Re-run a bounded range, write to a location no consumer is reading, validate, then publish by swapping (Planning a Backfill).
- A job that appends is not re-runnable. A job that replaces a partition, or merges on a business key, is (Upserts and Merges).
What can go wrong
- The job runs at full parallelism and produces a wrong answer, which is the failure mode distribution does nothing to prevent and slightly obscures.
- One partition is far larger than the rest, so 999 tasks finish and one keeps running while the cluster idles (Data Skew).
- The driver runs out of memory pulling results back to itself, taking the whole application down after every task succeeded (The Spark Execution Model).
- A machine is lost mid-stage and takes its shuffle output with it, forcing the previous stage to be recomputed rather than just the current one (The Shuffle).
- The mitigation fails too: adding workers to a job bound by one task, or by object-storage request latency, changes the bill and not the runtime (The Bottleneck Moves After Every Fix).
- "Distributed means fast." It means parallel. A distributed engine executing a badly shaped query moves more bytes over a network to reach the same wrong answer more slowly than one machine would.
- "Just add more workers." Workers help exactly one condition: every task is roughly equal and there are more tasks than slots. They do nothing for a job bound by one large partition, by request latency, or by the driver (Straggler Tasks).
- "We are big data now." Volume is one of four reasons to distribute, and usually the least common. The others are memory, restartability and read bandwidth, and each has cheaper answers (Horizontal vs Vertical Scaling).
- "The engine will optimise it." The optimiser rewrites what it can see. It cannot see through a user-defined function, and it cannot fix a join at the wrong grain (Query Optimizers).
Operating it
- The distribution of task durations per stage, not the mean. One chart of max versus median task time explains more distributed-job pathology than any other single signal (Tail Latency: Why p50 Being Fine Does Not Help).
- Bytes shuffled read and written per stage. It is the clearest proxy for what the job is actually doing and the number that moves when a join plan changes (Pipeline Metrics).
- Spill bytes and peak executor memory. Spilling is the transition between a job that is bound by CPU and one that is bound by local disk, and it is usually invisible in wall-clock terms until it is severe (Memory Pressure, Swap and the OOM Killer).
- At 10x volume a well-shaped job changes almost nothing about its structure — more partitions, more tasks, the same plan. That is the property distribution actually buys, and it is worth a lot.
- At 100x, the shuffle stops being a stage and starts being the job. Decisions that were stylistic — join order, pre-aggregation, whether one side can be broadcast — become the difference between running and not (Broadcast Joins).
- Cardinality scales worse than volume. Ten times the rows is a bigger job; ten times the distinct keys is a bigger shuffle, more state, and a worse skew profile (Partition Cardinality).
- Cost is cluster-seconds, and cluster-seconds are *worker count x wall clock* — including every worker that sat idle waiting for the last task. A cluster that is 90% idle for the final third of a job is paying full price for it (Compute Waste).
- Bytes read from storage, decided by layout and projection before the job starts. This is the driver that is set months earlier and is almost always the largest (Scan Cost).
- Bytes shuffled, which is driven by join and aggregation shape rather than by input size, and which is the term that grows fastest as a query gets more sophisticated (The Shuffle).
- Repeated work: the same source scanned by four jobs because no one materialised the intermediate, or a full rebuild each night when only one day changed (Incremental Processing).
- A cluster costs you determinism of experience: the same job, the same data and the same code can take twice as long because a machine was slow, and there is no local reproduction of that.
- Debugging moves from a stack trace to a UI full of stages. The mistake is usually in one task out of thousands, and finding it requires reading the shape of a job rather than stepping through code (Self Time, Total Time, and Where the CPU Went).
- Serialization, scheduling and coordination are pure overhead. Below a certain input size distribution is strictly slower than not distributing, and the crossover point is far higher than most teams assume (Parallel Overhead).
Where this applies
Almost nothing here is universal. These labels say what each claim is specific to, and where a different engine, format, warehouse or scale would differ.
- GENERALThe partition-and-shuffle structure is shared by every distributed data engine — Spark, Flink, Trino, Dremio and the distributed warehouses. What differs is who schedules the tasks and whether the shuffle goes through local disk, memory or a separate service.
- SCALE-SPECIFICBelow the point where data stops fitting in one machine's memory, single-node columnar engines usually win on wall clock and always win on operational cost. The advice inverts once the input exceeds memory, once read bandwidth is the limit, or once a run is long enough that mid-run failure must be survivable.
- SIMPLIFIEDPresenting a job as scan, shuffle, aggregate, write leaves out broadcast exchanges, adaptive re-planning at runtime, and engines that pipeline across stage boundaries instead of materialising. Those change the details of every stage and none of the reasoning about which operations need redistribution.
Where the depth lives
This domain teaches how data moves and how you know it arrived intact. It hands the rest off by name.
- — Distributed Systems owns why partial failure is the normal case, what a retry can and cannot promise, and the consensus and membership machinery a cluster manager is built on. This module assumes those results and reasons about their consequences for a data job.
- — DevOps / Production Engineering owns how the job binary and its dependencies are built, versioned and deployed onto the cluster, and how a bad version is rolled back after it has already written output.