One line of SQL, two terabytes
The question fits on one line: count yesterday's plays that lasted at least 30 seconds, per track, and turn the counts into money per rights holder. A music service asks it every day to pay the labels. The input is what makes it a project: 4,000 JSON files in object storage, 2 TB, about 1.8 billion plays.
Answering it is a batch job: a program that runs over a bounded input, a dataset with a known end, computes its answer and stops. Because the input is complete, the job can count without guessing what has not arrived yet, and running it again over the same input should give the same answer.
On one machine that parses about 100 MB a second, 2 TB takes 20,000 seconds, five and a half hours, and a crash in the fifth hour starts it over. So the work is spread across many machines.
Split the input, then meet by key
The job splits its input into partitions, here the 4,000 files of 500 MB each, and runs one map task per partition. A map task reads only its own file, drops plays under 30 seconds and emits one record per qualifying play: the track id and a 1. Map tasks share nothing, so a cluster can run 400 at once and, in theory, read the day in under a minute.
Counting per track breaks that independence: a track's plays are scattered across all 4,000 files, and every record for it has to reach the same place. That step is the shuffle. Each map task splits its output into 200 buckets by a hash of the track id; each of 200 reduce tasks fetches its bucket from every map task, groups the records by track and sums them, and the totals, priced per rights holder in a last small step, are written as the day's output.
Each map task reads one input file. The shuffle sends every record of a track to the reduce task its hash picks, so each reduce task reads from every map task.
Google's 2004 MapReduce paper gave this shape its name. Apache Spark, the usual engine for it today, chains many such steps in one job instead of writing each result back to storage, but the rule still holds in Spark: a job splits into stages at each shuffle, and a reduce task cannot finish until every map task has delivered its bucket.
The shuffle is also where the job pays. Every map task writes its buckets out, every reduce task pulls from every map task across the network, and records are serialized and sorted or hashed at both ends. The cheapest shuffle is a smaller one: each map task can add up its own plays first and send one partial count per track it saw, a map-side combine. A map task over a file of about 450,000 plays covering 60,000 tracks then sends 60,000 records instead of up to 450,000. Partial counts can be added together in any order; per-file medians cannot, and neither can per-file counts of distinct listeners, because one listener can appear in many files.
One song, one slow task
The labels also want unique listeners per track, which, done directly, needs every (track, listener) pair of a track in one reduce task. Then a major album comes out, and its lead single gets 90 million plays from 30 million listeners in a day, while an average reduce task handles about 9 million plays in total. Map tasks can drop repeated pairs only within their own file, so the task that owns the single receives 30 to 90 million pairs, and the other 199 finish and wait. This is skew: a few keys carry far more data than the rest, and the task that owns one becomes the straggler the stage waits for. More machines do not help; the hot key still lands on one task.
The usual fix is to split the hot key on purpose with a salt: send the single's pairs to 16 tasks by a hash of the listener id, count distinct listeners in each, and add the 16 counts in a second, small step. The sum is right because a listener always hashes to the same salt.
A task dies halfway
Out of 4,200 tasks a day, some fail: a machine is reclaimed, a disk fills. The engine does not restart the job. It runs the failed task again, usually on another machine, from the same input partition, and reruns any map task whose stored buckets were lost with its machine. A failure costs a few tasks, not five hours, and the answer stays right under two conditions:
- The task is deterministic. Run twice on the same partition, it emits the same records. A task that reads the clock, draws a random number or looks up a table that changes during the run can give its retry a different result from the attempt other tasks already read, and the job ends up holding parts of both.
- Its output appears all at once or not at all. A task writes to a temporary location, and the engine commits that output, by a rename where the storage renames atomically or by a manifest, only when the attempt succeeds. The engine relies on this when it starts a second copy of a straggling task, speculative execution, and commits only the copy that finishes first.
A payment or an email sent from inside a task goes out again on every retry and every speculative copy. Such side effects belong outside the tasks or must be safe to repeat, the problem Idempotency & exactly-once illusions takes apart.
Running yesterday again
Whole runs repeat too: a rerun after a fix, a scheduler that fires twice. A job that appends its totals to a table doubles that day's royalties on the second run. So the output is laid out like the input, one output partition per day such as royalties/2026-10-22/, and each run replaces its day instead of adding rows to it: an idempotent partition overwrite. Run it once or five times and the day holds one set of totals, from whichever run committed last, so two runs of the same day should not overlap.
Readers must see the old day or the new one, never half of each. On plain object storage, deleting a directory and writing new files are separate steps, so the job writes the new day to a fresh location and then points the table's partition at it, a single metadata change. Lake table formats make that switch one commit for any set of files in a table; Analytics data: OLTP vs OLAP & the warehouse explains them.
"Finished" is a decision, not a fact. Plays are filed under the day they happened, and a phone that played offline can upload them two days later, into a day already computed. Because a day reruns safely, the usual answer is to compute recent days again a few days later and let the overwrite replace the earlier totals.
Backfilling three months
In late October the service finds that since 1 August its web player has logged a play twice, under the same play id, whenever a listener resumed after a pause. Making the job drop repeated play ids, which needs its own shuffle by play id, unless the input is already partitioned that way, because the two copies can sit in different files, stops the overcount from now on; the 83 days already computed still overpay.
Running the corrected job over those historical partitions is a backfill, and with the job built as above it is the daily job with a different date. Each run reads one day of raw plays (partitioned by the day a play started, so both copies of a play share a day) and replaces that day's output, so the days are independent: run a few at a time, rerun the ones that fail, and readers see each day switch to the corrected totals as its commit lands. Nothing is counted twice, because nothing is added.
Each run reads one day of raw plays and replaces that day's output partition, so the 83 days can run in any order and any of them can run again.
A backfill is only as good as what it reads:
- The raw input must still exist. With 60 days of raw plays kept, the days before late August are out of the backfill's reach unless another copy of their input survives.
- Reference data must be read as it was. Joining August's plays with today's rights-holder table pays a catalogue sold in September to its new owner for August too.
- Old events may have an older shape. August's plays lack a field the player added in October; reading both shapes is what Schema evolution: event formats & compatibility is about.
- Outputs built on those days must follow. Monthly statements are rerun after the days they read, an ordering left to the orchestrator in ETL/ELT & data orchestration.
Waiting for the whole day, on purpose
A royalty statement can wait hours for a number of record over input that has stopped changing, one it can recompute. That is the trade batch makes: latency for a simpler job that carries no state between runs. A per-minute chart of plays during the album's launch wants the opposite, and many systems run both over the same raw events. The stream side, with its own rules for when a stretch of time is complete, is in Stream processing: windows, watermarks & state.
A brief that asks for a backfill
One requirement of Analytics Lakehouse reads: backfill historical partitions after transform logic changes. A design that meets it shows each modelled table's partition key, how a rerun replaces a partition, and how long raw data stays in the lake, which sets how far back a backfill can reach. Ad Click Aggregation depends on the same thing for its accurate batch totals: raw clicks kept where a batch job can read a whole day again.