Module 4 ended with a table showing where each piece of Kilometre Zero data lives: orders in Cassandra, stock in PostgreSQL, the catalogue in Redis, photos in MinIO and, in the HDFS data lake, one file per day holding the hundreds of thousands of events from orders.events and the millions of clicks from the website. Spreading the data out solved the problem of storing it; now the next problem appears: computing something with it. Adding up Grape Harvest Week sales by producer and market over seven days of events, or training recommendations on a year of clicks, does not fit on one machine, and even if it did it would take hours. This lesson explains what changes when the thing being distributed is a computation and not just data: why it pays to take the code to where the data is, what ways there are to split up the work (scatter/gather, divide and conquer, work queues, BSP, dataflow, actors), how batch, streaming and interactive queries differ, and what new problems show up (the cost of moving data between nodes, skew, stragglers, re-execution after a failure, the limits of scaling). Everything that follows in this module (MapReduce, Spark, Flink, Airflow) is a concrete implementation of these ideas, so it is worth understanding them first with single-machine Python code, which is what we will do in simulations/.

1. Distributing a computation: taking the code to the data
2. Data parallelism and task parallelism
3. Distributed computing patterns
4. Batch, streaming and interactive
5. The cost of communication: the shuffle
6. Data skew
7. Stragglers and speculative execution
8. Fault tolerance through deterministic re-execution
9. Scalability: Amdahl and Gustafson
10. Hands-on: scatter/gather, work queue and skew in simulations/
11. Common Mistakes and Tips
12. Exercises
13. Conclusion

When the catalogue fitted in a PostgreSQL table, "compute sales by producer" was a SQL query: the engine read the rows from its local disk, grouped them and returned twenty numbers. The data and the computation were on the same machine. In the data lake from 04-02 that is no longer the case: /km0/events/2026-09-14/orders.jsonl is 150 MB in two blocks living on three different DataNodes, and the seven days of Grape Harvest Week add up to more than a gigabyte spread across the whole cluster. There are two ways to compute over that:

• Take the data to the code. A process in analytics downloads the gigabyte over the network, scans it and adds things up. It works, but the network is the slowest resource we have (fallacy 3 from 01-04, "bandwidth is infinite"): at 1 Gbit/s, moving 1 GB takes about 10 seconds in transfer alone, and with a year of clicks (terabytes) the approach is simply not viable. On top of that, there is only one process doing the adding, so the computation does not scale.
• Take the code to the data (data locality). Send each DataNode a small program (a few kilobytes) that reads the block it already has on its local disk, computes a partial result (a few kilobytes: sales by producer for that block) and returns only that. The code and the results, which are tiny, move; the data, which is huge, stays put.

This inversion is the central idea of all modern distributed computing and the reason HDFS and MapReduce were born together: the file system exposes where each block lives (hdfs fsck -locations showed it in 04-02) precisely so that the compute scheduler can run each task on the node that holds its block, or at least in the same rack. When locality is not possible (the node is busy, or the data sits in an object store such as MinIO that exposes no locality), you pay for the network, which is why modern platforms separate compute from storage but make up for it with 25–100 Gbit/s networks and columnar formats that read only the columns they need (05-03).

The second change is that the computation becomes many independent tasks plus a combining phase. Instead of one program that scans everything, we write one function that processes a chunk and another that combines partial results. That decomposition is not free: some computations decompose naturally (adding up sales by producer) and others do not (sorting all events by amount, computing an exact median, traversing a graph of "customers who bought the same things"). The patterns in section 3 are the known ways of decomposing.

There are two ways to split a job across several nodes, and it is worth telling them apart because they call for different infrastructure:

Kilometre Zero's analytics is almost entirely data parallelism, which is why this module focuses on it. But the two are combined: the daily pipeline in 05-05 is task parallelism (validate, aggregate, load, notify) in which the "aggregate" task is, on the inside, data parallelism in Spark. And the law that governs scaling (section 9) is different in each case: data parallelism comes close to linear scaling because the sequential fraction is small; task parallelism is limited by the longest chain of dependencies.

This is the simplest pattern and the one that already appeared, unnamed, in 04-01 when we discussed secondary indexes in Cassandra: a coordinator splits the work among N workers (scatter), each one computes over its share, and the coordinator collects and combines the partial results (gather).

It works when the computation is associative and commutative: sum, count, maximum, minimum, set union. It makes no difference in what order or in what grouping the partials are combined; the result is the same. An average is not directly combinable (the average of averages is wrong if the chunks differ in size), but it becomes combinable by carrying (sum, count) instead of the average; an exact median does not, and you have to approximate it or pay for a global sort. This is the first question to ask of any distributed computation: what partial result does each chunk return, and how are two partials combined?

The limit of scatter/gather is that there is a single coordinator: it splits, waits for everyone and combines. If combining is expensive (millions of distinct keys) or the partials are large, the coordinator becomes the bottleneck. MapReduce (05-02) solves exactly this by distributing the combining phase as well.

This is scatter/gather applied recursively: a problem is split into subproblems, each subproblem is split again until it fits on one node, and the results are combined on the way back up the tree. A distributed sort works like this: each node sorts its chunk (a local mergesort), and successive combinations merge sorted lists. Frameworks use it for reduce operations across many nodes: instead of one coordinator adding up 1,000 partials, they are added up in a tree (treeReduce in Spark), with 1,000 → 32 → 1 combinations in parallel.

In scatter/gather the coordinator decides in advance which chunk goes to which worker. With a work queue it does not decide: it publishes all the tasks to a queue and each worker takes the next one when it finishes the previous one. This is the competing consumers pattern from 02-04 applied to computation, and it has two big advantages:

• Dynamic balancing. If a worker is slow (an old machine, a big task), it simply takes fewer tasks; the fast ones absorb the rest. Nobody sits idle waiting.
• Natural fault tolerance. If a worker dies halfway through a task, the task goes back to the queue (through the lease or the pending ack) and another worker runs it. For that to be correct, the task must be idempotent: running it twice must give the same result as running it once (section 8).

This is the internal model of nearly every scheduler: YARN, Spark and Flink keep queues of pending tasks and assign them to executors as they become free, with a preference for the node that holds the data. We will see it in simulations/work_queue.py.

Some computations are not done in a single pass but in iterations where each step depends on the previous one: PageRank over a graph, training a model by gradient descent, propagating "customers who bought the same things" through the order graph for Kilometre Zero's recommendations. The BSP model (Valiant, 1990) organises those computations into supersteps:

1. Local computation: each node works only with its own data and the messages it received in the previous superstep.
2. Communication: each node sends messages to the others (for example, a graph vertex sends its value to its neighbours).
3. Barrier: nobody starts the next superstep until everyone has finished the current one and all messages have arrived.

The barrier is what makes the model easy to reason about (within a superstep there are no races: each node only sees messages from the previous superstep) and also what makes it sensitive to stragglers (section 7): the superstep lasts as long as the slowest node. Google's Pregel, and its descendants Apache Giraph and Spark's GraphX, are BSP that "thinks like a vertex": each vertex of the graph receives messages, updates its value and sends messages to its neighbours, superstep after superstep until no vertex changes. At Kilometre Zero, "products often bought together" is a graph where the vertices are products and the edges are weighted by the number of shared orders; two or three supersteps of propagation are enough to find that aged-cheese and crianza-wine are closer than their categories suggest.

The dataflow model describes the computation as a directed acyclic graph (DAG) of operators: read, filter, transform, group, join, write. Each operator receives data from the ones before it and emits data to the ones after it; data flows along the edges. The system decides how to parallelise each operator (how many instances, on which nodes), how to chain operators that do not need to redistribute data (filter and transform can run in the same process, row by row), and where redistribution is needed (grouping by producer forces all the rows for one producer to reach the same instance: that is the shuffle of section 5).

This is the model of Spark (05-03) and Flink (05-04), and also, under a different vocabulary, of distributed SQL engines and pipeline schedulers (05-05, where the nodes of the DAG are whole jobs rather than operators). Compared with MapReduce, which forces everything to be expressed as chained map/reduce pairs, dataflow lets the programmer write the complete transformation and leaves it to the optimiser to decide the stages. Compared with BSP, dataflow has no global barriers: an operator processes as soon as it has data, and only the shuffle synchronises.

The actor model (Hewitt, 1973) takes a different route: it neither splits data nor describes a graph, but models the system as many small objects (actors) that communicate only through asynchronous messages, each with its own private state and mailbox, processing one message at a time. There is no shared memory and there are no locks: an actor receives a message, changes its state, sends messages to other actors or creates new actors. Erlang has had it built into the language since the 1980s (Ericsson's telephone exchanges; RabbitMQ from 02-04 is written in Erlang) and Akka brought it to the JVM. It is a natural fit for per-entity state: one actor per courier, such as van-3, that receives its positions and maintains its route, or one actor per order that runs its saga (03-05). It is less suited to massive computation over historical data, which is what this module is about, so we will leave it at a mention: its place comes up again when per-key state takes centre stage, and in fact Flink's stateful operators (05-04) look a lot like actors partitioned by key.

Regardless of the pattern, there are three processing modes, depending on when the data arrives and when the answer is needed:

The three modes share the patterns of section 3 and the problems of sections 5–8, but each suffers them differently: skew in a batch makes a night longer; in a stream it clogs a partition for good. And the boundary between batch and streaming is blurrier than it seems: a batch of "the events of 14 September" is a stream that has been given a beginning and an end; a stream processed in one-second micro-batches is a series of very small batches. That idea, that one engine can treat both modes with the same DAG, is the one Spark and Flink exploit and the one 05-04 will develop.

Distributing a computation has a cost that does not exist on one machine: moving data between nodes when the next step needs it grouped differently. Adding up sales by producer requires all of Montblanc Dairy's lines to end up on the same node, and those lines are scattered across every block on every node. The operation that brings them together is called the shuffle, and it is by far the most expensive phase of any distributed job, because:

• It is all-to-all: every node sends part of its data to every other node. With N nodes that is N² network flows.
• It usually goes through disk: the data is serialised, written out sorted by destination key, transferred and read back in. In MapReduce every shuffle is a full write to disk (05-02); Spark keeps it in memory when it can (05-03).
• It cannot fully overlap with the computation: the receiver needs all its data before grouping, so the shuffle is an implicit barrier.

The rule of thumb is to minimise what crosses the shuffle: reduce before you move. If each node adds up its lines by producer locally before sending them (a combiner, in the vocabulary of 05-02; a partial aggregation, in Spark), then instead of moving 250,000 lines it moves 20 partials per node. That is the reason for insisting that the computation be associative: only then can you pre-reduce. And it explains the advice in 04-01 about tables materialised by market and producer: somebody had paid for the shuffle once, at write time, so as not to pay for it on every query.

A computation that needs no shuffle (filtering, transforming row by row, adding up a global total that combines into a single number) is embarrassingly parallel and scales almost linearly. One that needs several chained shuffles (group by producer, join with the catalogue, regroup by market) is limited by the network and by the worst of the following sections.

The ideal split gives every node the same amount of work. The real split depends on the distribution of the keys, and real distributions are uneven. During "Artisan Cheese Week" Montblanc Dairy accounts for half of Kilometre Zero's order lines; if the shuffle groups by producer, the node that receives montblanc-dairy processes half the data while the other seven share the other half. The job takes as long as that node takes: with 8 nodes and one key that weighs 50%, the maximum speedup is 2, not 8. This is data skew, and it is the most common cause of distributed jobs that "don't scale even though I add machines".

You detect it by looking at the duration of the tasks within one stage: if most finish in 20 s and one takes 3 min, there is a hot key. The solutions, all of them ways of breaking up the fat key, are covered in the hands-on part (section 10.3) and will be picked up again in 05-03 under the name Spark gives them, salting:

• Pre-reduce before the shuffle: if each node adds up its Montblanc Dairy lines, what crosses the network is one partial per node, not half the data.
• Spread the hot key by adding a random suffix (montblanc-dairy#0 … #7), group by the suffixed key and then regroup the eight partials in a second, much smaller step.
• Choose a different partition key when the computation allows it: grouping by (producer, market) spreads Montblanc Dairy across four markets.
• Handle known hot keys separately (filter them out, process them with more parallelism) and join afterwards.

Skew also exists in the input: if the file for 14 September is two blocks and the one for the 15th is ten, the per-day jobs will have very different durations. And it exists in time: Kafka events cluster around midday and late afternoon, so a stream partitioned by hour has fat hours.

Even with a perfect split, some task will end up taking far longer than the rest without the data justifying it: the machine has a degraded disk, another job is competing for its CPU, its rack's network is saturated, the JVM is in a garbage collection pause. These are stragglers. In a job with 1,000 tasks the probability that one of them lands on a troubled machine is high, and since the job finishes when the last task finishes, a single straggler lengthens the whole job. In BSP the effect is multiplied by the number of supersteps.

The classic solution, introduced by MapReduce and present in Spark (spark.speculation) and in Hadoop, is speculative execution: when a stage is nearly finished and one task has been running far longer than the median of the others, the scheduler launches a copy of that task on another node; whichever finishes first wins, and the other is cancelled. It is an expense (redundant work gets run) that pays off because the cost of one extra task is far lower than that of the whole cluster waiting. It is only possible because tasks are deterministic and idempotent, which is the subject of the next section: if the copy and the original both wrote their result, you would have to guarantee that only one gets written (atomic output, 05-02).

On one machine, if the program fails halfway through, you relaunch it from the start. With 1,000 nodes and a three-hour job, the probability that some node fails during the job is practically 1 (fallacy 1 from 01-04), and relaunching everything each time is unacceptable. The strategy of the frameworks in this module is fine-grained re-execution: if a task fails, that task is re-run, not the job. For that to be correct, three properties we already know from 02-05 are needed:

1. Immutable input. The task reads a chunk of data that does not change (an HDFS block, a range of Kafka offsets, a partition of an RDD). Re-running reads the same thing.
2. Deterministic computation. The same input produces the same output. No unseeded random numbers, no dependence on the current time or on the order of arrival over the network. If randomness is needed, it is derived from the key or from a fixed seed.
3. Idempotent or atomic output. Either writing twice is harmless (a PUT of an object with the same name in MinIO, an INSERT ... ON CONFLICT DO NOTHING with the task id), or the output is written to a temporary location and published in one go on completion (renaming a file, committing a transaction). That way a task that died halfway leaves no partial output to confuse its re-execution or the speculative execution.

With those three properties, a node failure boils down to "its tasks go back to the queue", which is what the work queue of section 3.3 was already doing. What changes with each framework is how it rebuilds the input of a task when that input was the result of an earlier task: MapReduce always writes it to disk (HDFS or local), Spark remembers how it was computed and recomputes it (the lineage of 05-03), Flink saves periodic checkpoints of the state (05-04). And what does not change is that everything rests on idempotent tasks: the idempotent consumer from 02-05 and a Spark task are the same idea at a different scale.

In 01-03 we saw Amdahl's law: if a fraction 1 − p of the work is sequential, the speedup with N nodes is bounded by 1 / (1 − p) no matter how many nodes are added. In a distributed job the sequential part is whatever is not split: reading the list of blocks, scheduling tasks, the final gather, writing the result to a single file, and above all the shuffle and the wait for stragglers, which, even though they run in parallel, behave like a barrier. With p = 0.95 the ceiling is 20 even if 1,000 nodes are used, which seems to say that large-scale distributed computing is not worth it.

John Gustafson's answer (1988) is that Amdahl assumes a fixed-size problem, and that is not how clusters are used: nobody buys 1,000 nodes to compute one day's sales faster, but to compute a year's sales, or all the clicks, in the same time. Gustafson's law measures the scaled speedup: if with N nodes the job takes a time T of which a fraction s is sequential and 1 − s is parallel, that same job on a single node would have taken s + (1 − s) · N times T, so:

With s = 0.05 and N = 1,000, the scaled speedup is 950: almost linear, because as the problem grows the sequential fraction (scheduling, gathering 20 numbers) stays the same while the parallel one (reading terabytes) grows with N. Both laws are true, they measure different things, and together they give this module's design rule:

The practical consequence: parallelism pays off when the work per node is large compared with the fixed cost of starting up and coordinating. Launching 1,000 tasks of 100 ms each on a cluster with 2 s of scheduling latency is worse than 10 tasks of 10 s. We will check this by measuring in the hands-on part.

The whole hands-on part runs on one machine with multiprocessing, which splits work among processes just as a cluster splits it among nodes, with the difference that the "network" is local memory. That is enough to see the patterns, measure speedups and reproduce skew. The input file is a small version of /km0/events/2026-09-14/orders.jsonl, with the event envelope from 02-05 and one order.created event per line:

These three lines are enough to read the code; to measure anything you need more volume, so the first script includes a generator that creates hundreds of thousands of events with the same shape and a distribution skewed towards Montblanc Dairy.

Points worth understanding in the code:

• The coordinator does not read the data. chunks_by_bytes only computes byte ranges, with a seek and a readline to align each cut to the start of a line (without that, a line would be split between two workers and both would either discard it or count it wrongly). This is exactly what Hadoop does with input splits, and why JSON Lines or CSV are "splittable" whereas a JSON file with one giant array is not.
• Each worker opens the file on its own and reads only its range. In a cluster, that worker would run on the node that holds the block (locality); here they all share one disk.
• The partial is small: a Counter with three keys, regardless of whether the chunk has a thousand lines or a million. What crosses the "network" (Pool's internal queue) is three numbers per worker.
• reduce is trivial because addition is associative. If the computation were "average amount per order", the partial would have to be (sum, count) per producer and the reduction would divide at the end.

A run on an 8-core laptop with 400,000 orders (about 130 MB):

The result is identical with any number of workers (the reduction does not depend on the split) and the speedup drifts away from the ideal as the processes grow: with 8 we get 5.8, and with 16 (more processes than cores) nothing more. This is Amdahl in action: starting the Pool costs a fixed 100 ms or so, the gather and the chunk alignment are sequential, and the disk is shared. With a file ten times larger the sequential fraction is diluted and the speedup with 8 approaches 7.5: Gustafson.

The second script turns the computation into idempotent tasks on a queue, with workers that take them on demand and one that dies halfway through a task. Each task is "compute sales by producer for one market and one day", and its output is a file output/<day>-<market>.json written atomically.

What the run teaches us:

• Dynamic balancing: w0 finishes Girona and takes Valencia without anybody assigning it; with markets of uneven size, the fast workers absorb more tasks.
• Detection and re-execution: the coordinator receives no error message from w1 (it died with os._exit, like a machine being switched off); it detects the death because the process is no longer alive, just as YARN detects a NodeManager through missed heartbeats. It requeues the task as is and launches a replacement (on another run it may be w2, if it becomes free first, that takes Lleida: the queue does not assign, the workers compete).
• Idempotency: w1 died after starting and perhaps with the temporary file half-written; w1' runs the same task, writes its own temporary file and renames it. The orphaned temporary file from w1 stays in output/ with nobody reading it (a real job would clean it up). Had the output been an INSERT into PostgreSQL with no unique key, the re-execution would have duplicated rows: it is the same problem as the consumer in 02-05.
• The result is the same as that of the scatter/gather, which is the very definition of the failure having been tolerated.

The third experiment reuses sales_scatter_gather.py but changes the split: instead of byte chunks, it splits by producer, which is what a naive "group by producer" shuffle would do. Add this function and the call to the script:

(For the skew to show up in the time, and not only in the amount of data, each worker in a real shuffle would receive only its own lines; in this simulation each one filters the complete file, so add to the inner loop a small piece of work proportional to the worker's own lines, for example hashlib.md5(line).hexdigest() only when ln["producer"] == producer.) The output shows the problem:

Three workers, and the job lasts 3.52 s: what Montblanc Dairy takes. The other two sit idle half the time. The fix is to spread the hot key over subkeys and do a second reduction:

Now the longest task processes an eighth of the data instead of half, and the whole job drops to a little over 1 s with six processes. Note that Python's hash() is randomised per process for strings (PYTHONHASHSEED), so for the subkey to be deterministic across runs and re-executions you have to fix the seed or use zlib.crc32(order_id.encode()) % n_sub; it is a small example of how the non-determinism of section 8 sneaks in. Spark will do this very thing with two groupBy calls and a salt column in 05-03.

• Moving the data to the computation out of habit. The first instinct of anyone coming from the monolith is "I'll download the file and process it". With gigabytes that works badly and with terabytes it does not work at all. Always ask where the data is and whether the computation can go there.
• Designing a partial result that does not combine. Averages, percentiles, distinct counts (count distinct) and medians cannot be added up. Carry (sum, count), use approximate structures (HyperLogLog for distinct counts, t-digest for percentiles) or accept a full shuffle.
• Ignoring the shuffle. A groupBy on a high-cardinality key is an all-to-all operation. Pre-reduce before moving, choose keys with reasonable cardinality and look at the shuffle bytes in the framework's UI (05-02 and 05-03 show them).
• Blaming the cluster for skew. When it "doesn't scale", look at the duration of the tasks in the slow stage. If one takes five times the median, you are not short of machines: you have one key too many.
• Non-idempotent tasks. An INSERT with no key, an incremented counter, an append to a file: any of these turns re-execution (and speculative execution) into a duplicate. Write to a temporary file and rename, use natural keys, or write whole partitions with overwrite (05-05).
• Hidden non-determinism. Python's hash() with a random seed, datetime.now(), unseeded random, the ordering of a dict in an old version, the order of arrival of messages. Re-running a task must produce identical bytes.
• Lots of tiny tasks. The scheduler has a per-task cost (milliseconds to seconds). 100,000 tasks of 50 ms are worse than 1,000 of 5 s. As a guide, between 2 and 4 tasks per core and stage, with durations of seconds to minutes.
• Extrapolating Amdahl to Gustafson (or the other way round). If the problem is fixed, extra nodes do not help; if the problem grows, they do. Before asking for more machines, decide which of the two cases is yours.

For each of these computations over the order.created events of Grape Harvest Week, state (a) what each worker returns as a partial result and (b) how two partials are combined, or why they cannot be combined and what the alternative is:

1. Total amount sold by Roble Alto Winery.
2. Average amount per order in each market.
3. Number of distinct customers who bought crianza-wine.
4. The 10 best-selling products by quantity.
5. The median order amount.

In work_queue.py, replace write_atomically with a direct write (open(path, "w") and json.dump) and make the worker die during the write (for example, write half of the JSON, call f.flush() and then os._exit(1)). Describe what happens in the coordinator at the end, how you would detect it in production and why os.replace prevents it. Then propose how you would write the output if, instead of a file, it were a PostgreSQL table market_daily_sales(day, market, producer, amount), so that re-execution would still be safe.

Using the timings from the run in section 10.1 (1 worker: 6.84 s; 8 workers: 1.18 s), estimate the script's sequential fraction according to Amdahl. With that fraction, what speedup would you get with 64 workers on the same file? And what would Gustafson's scaled speedup be with 64 workers if the file grew 64-fold? Finally: a real cluster's scheduler adds 1.5 s per task between assignment and start-up; if the 130 MB file is divided into 1,000 chunks, how long does the job take with 8 nodes, and how many chunks should you divide it into?

Exercise 1.

1. Partial: one number (the sum of quantity × price over Roble Alto Winery's lines). Combination: addition. Associative and commutative; the ideal case.
2. Partial: per market, the pair (sum_amounts, order_count). Combination: add component by component; the average is computed only at the end, sum / count. Combining averages directly would be wrong unless every chunk had the same number of orders.
3. Partial: the set of customer ids who bought crianza-wine in that chunk. Combination: set union, and at the end its size. It is combinable but the partial can be large (hundreds of thousands of ids); if that is a problem, one HyperLogLog per chunk (a few KB) combines with a union and gives the cardinality with a 1–2% error.
4. Partial: quantities per product (a Counter), not "the chunk's top 10": the eleventh in one chunk may be the first overall. Combination: add the Counters and pick the 10 at the end. If the number of products were enormous, you could keep the top-K per chunk with a large K as an approximation, accepting some error.
5. Not combinable: the median of medians is not the median. Alternatives: sort globally (a shuffle by amount ranges: expensive but exact), or a t-digest/approximate percentile per chunk, which does combine and gives the median with a bounded error.

Exercise 2.

With a direct write, w1 leaves output/2026-09-14-lleida.json holding half a JSON document. The coordinator requeues and w1' opens the file again in "w" mode, which truncates it, so in this particular case the final result is correct; but between the death and the re-execution (seconds, or minutes in a cluster) the file exists and is corrupt: any reader (the pipeline in 05-05, an hdfs dfs -cat) would fail with a JSONDecodeError, or worse, would read a partial result if the format were CSV. If the worker died after writing but before sending end, the re-execution would also overwrite, with no harm done. The real problem appears if the re-execution does not truncate ("a" mode) or if the output is a system with no truncation. In production you detect it through failing readers or files of unexpected size; os.replace prevents it because the file with the final name either appears all at once and complete or does not appear: it is the atomic output of section 8 and the one MapReduce implements with _temporary directories (05-02).

For PostgreSQL: a primary key (day, market, producer) and a write with INSERT ... ON CONFLICT (day, market, producer) DO UPDATE SET amount = EXCLUDED.amount, all inside a transaction that begins with DELETE FROM market_daily_sales WHERE day = %s AND market = %s and ends with COMMIT: the task replaces its whole partition atomically, and running it twice leaves exactly the same rows. This is the "reprocess a day without duplicating" of 05-05.

Exercise 3.

Amdahl: S(8) = 1 / ((1 − p) + p / 8) = 6.84 / 1.18 = 5.80. Solving, (1 − p) + p / 8 = 1 / 5.80 = 0.1724 → 1 − 0.875 p = 0.1724 → p = 0.946. Sequential fraction 1 − p ≈ 5.4%. With 64 workers: S(64) = 1 / (0.054 + 0.946 / 64) = 1 / 0.0688 = 14.5: in practice, less, because the laptop has 8 cores and there is one disk. Gustafson with s = 0.054 and N = 64: 64 − 0.054 × 63 = 60.6: processing 64 times more data with 64 nodes in almost the same time.

With 1,000 chunks and 8 nodes, each chunk is 130 KB (about 7 ms of processing) plus 1.5 s of scheduling cost: 1,000 tasks × 1.507 s / 8 nodes ≈ 188 s. Worse than the 6.84 s of a single process. With 8 chunks: 8 × (0.86 s + 1.5 s) / 8 ≈ 2.4 s. With 16 or 24 chunks (2–3 per node) dynamic balancing helps with stragglers without blowing up the fixed cost: about 2.5–3 s. The rule: the work per task must be at least an order of magnitude greater than the cost of scheduling it.

Distributing a computation is more than running it on several machines: it means taking the code to where the data is, decomposing the work into tasks whose partial results can be combined, and accepting that the combining (the shuffle) is the expensive part. We have separated data parallelism, which is the concern of this module, from task parallelism, which is the concern of pipelines; and we have gone through the patterns used to organise the split: scatter/gather for anything associative, divide and conquer for combining in a tree, the work queue for balancing and tolerating failures on demand, BSP with its supersteps and barriers for iterative work and graphs, the dataflow DAG of operators that Spark and Flink implement, and actors for per-entity state. The three modes (batch, streaming, interactive) share patterns and problems: the skew that makes Montblanc Dairy lengthen the whole job, the stragglers that speculative execution sidesteps, and the fault tolerance that only works if tasks are deterministic and idempotent with atomic output, the same rule that governed the consumers in 02-05. Amdahl reminded us that a fixed problem has a ceiling, and Gustafson that clusters exist for problems that grow. In simulations/ we have measured a speedup of 5.8 with 8 processes, watched a worker die and come back without altering the result, and broken Montblanc Dairy's hot key into subkeys.

We did all of this by hand, with multiprocessing and a forty-line coordinator that splits work, detects deaths and requeues. A distributed computing framework is exactly that, but for thousands of nodes, with data locality, a distributed shuffle, speculative execution and atomic output solved once and for all for every job. The first to achieve it, and the one that fixed the vocabulary we still use, was MapReduce, and with it Hadoop, which is the next lesson: how the same sales-by-producer computation is expressed as map, shuffle & sort and reduce, and how YARN spreads the tasks across the cluster.