--- title: 'Distributed Data Processing' source: 'https://academia.sh/en/courses/distributed-data-processing' language: en updated: '2026-10-21T05:03:00+00:00' license: 'CC BY-SA 4.0' lessons: 11 outcomes: - 'Explains the driver, executor, task and partition abstractions of distributed computing on a fictional cluster' - 'Counts shuffle cost and data skew and shows how the choice of partition key makes adding workers pointless' - 'Justifies a cluster sizing decision by measuring makespan and total work separately' --- # Distributed Data Processing Counting how the work is split when the data does not fit on one machine, and what splitting costs: building the driver, executor and task concepts on a fictional cluster, what block placement and replication add to a read, at which step the map and reduce abstraction touches the network, partitions and lazy evaluation eliminating unnecessary computation, the task a resource manager makes wait while scheduling work, skew of 1.02 on the subscriber key rising to 3.32 on the region key with two of four workers left idle, a shuffle under the same layout moving 1738 rows and raising total work from 2200 to 7414, with the fact side partitioned by region a broadcast join winning up to a 1600-row dimension and losing to the partitioned join at 4000, a makespan of 2250 on one worker rising to 3691 with a second worker and falling only after four, total work rising from 5830 to 8635 while makespan falls from 3691 to 1398, caching an intermediate result lowering two queries from 5500 to 3300, and lowering the checkpoint interval from six to three not reducing recovery at all at the same failure point. ## Distributed Computing Building the basic abstractions for splitting a job across several machines on a fictional cluster and counting the cost of each one: the driver's plan, the executor's share and the definition of a task, with a single-worker baseline giving a makespan of 2250 and a total work of 2200; the case where block placement and replication make a read local, and the number of blocks stored as the replication factor grows; which step of the map and reduce abstraction stays local and which one touches the network; splitting rows into partitions and lazy evaluation eliminating computation that is never used; the number of tasks a resource manager makes wait while balancing the number of workers against concurrent tasks, and the fixed cost of fifty units for handing work to each worker. ### Cluster Computing Fundamentals Source: https://academia.sh/en/courses/distributed-data-processing/cluster-computing-fundamentals Splitting a job across several machines with the driver, executor and task abstractions, and measuring the split with two separate numbers: the 2200 rows coming out of the pipeline give a makespan of 2250 and a total work of 2200 on a single worker, but when split across four workers by the subscriber key the makespan drops to 611, when split by the region key two of the four workers stay idle, the skew is 3.32 instead of 1.02 and the makespan can only fall to 1876; because the total work of the three setups is exactly the same 2200, a speedup claim that does not state which one was measured is incomplete. For sixteen lessons the pipeline was treated as if it ran on a single machine. Tasks were queued, failed and retried, and forty-eight nodes were laid out in a single list; between the stages of the pipeline data passed from one variable to another, and this handoff was treated as if it had no cost. The assumption is true for a single machine: passing a list that sits in the same memory to the next function is nearly free. When the data does not fit on a single machine, the assumption fails. The job has to be split, the pieces run on different machines, and every handoff between them crosses the network. This lesson builds that split and shows one thing: a split job has **no single number**. There are at least two, and they do not have to move in the same direction. - **DC1.** The source, the pipeline and the cluster are **fictional**. No real distributed processing framework, resource manager, distributed file system or cloud service is called. The seed is **20260218**, and the measurement does not depend on the run. - **DC2.** The rule of the course: the number of a distributed job is not a single number. The **makespan** (the work of the busiest worker) and the **total work** are measured separately, and **a speedup claim that does not state which one was measured is incomplete.** - **DC3.** The baseline is **a single worker**: no shuffle, skew 1.00, makespan **2250**, total work **2200**. - **DC4.** **Makespan is the model's own unit**, not raw milliseconds. Processing one row locally counts as one unit; it does not correspond to a wall-clock time measured on another machine. - **DC5.** A **task** is the share given to a worker: the same operation running on a subset of the rows. In the Data Pipelines and Orchestration course a task was a node in the graph, that is, a stage running for a specific period; here the task sits **inside** that node. - **DC6.** Handing work to each worker has a fixed cost: **`TASK_OVERHEAD = 50`**. This is a choice, and every result that is sensitive to it is written together with its value. - **DC7.** Moving a row over the network costs **three times** as much as processing it locally (`NETWORK_COST = 3`). This is also a choice; since there is no shuffle in this lesson, it enters no result. - **DC8.** The units measured are rows, workers, tasks, makespan, total work and skew. Makespan and total work are integers, and no ratio is printed over them. #### Driver, Executor and Task A distributed run has three roles, and none of the three existed in the single-machine pipeline. The **driver** is the single point that holds the plan of the job: it knows which operation will be performed, by which key the data will be split and where the results will be collected. The driver does not process data itself; it hands out work and collects the result. An **executor** is a process that runs on a machine and processes the share given to it. In this model each executor counts as a single **worker**, so the number of workers is the number of executors. The **task** is the third and the most commonly confused concept. In the Data Pipelines and Orchestration course a task was a node in the graph: `screen@2026-07` was a task, and there were forty-eight of them. Here the task has nothing to do with the graph; it is **the share** given to a worker. When the same `screen@2026-07` node is split across four workers, four tasks are born, and all four run the same operation on different subsets of rows. Orchestration puts tasks **in sequence**, a distributed run puts them **side by side**. These three roles are built on the fictional cluster below. The first block rebuilds, exactly as it is, the source defined in the Introduction to Data Engineering course and the first three stages of the pipeline, and adds the partitioning and cost functions on top. ```python # FICTIONAL. Meter readings, pipeline and cluster are defined in the lesson; no real # distributed processing framework, resource manager or file system is called. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records def s1_screen(records): REQUIRED = ("subscriber", "period", "region", "index", "unit") keep = [k for k in records if all(a in k for a in REQUIRED)] return keep, len(records) - len(keep) def s2_dedupe(records): seen, keep = set(), [] for k in records: key = (k["subscriber"], k["period"]) if key in seen: continue seen.add(key) keep.append(k) return keep, len(records) - len(keep) def s3_consumption(records): table = {(k["subscriber"], k["period"]): k for k in records} out, dropped = [], 0 for k in records: i = PERIODS.index(k["period"]) if i == 0: continue prev = table.get((k["subscriber"], PERIODS[i - 1])) if prev is None: dropped += 1 continue try: f = int(k["index"]) - int(prev["index"]) except (TypeError, ValueError): dropped += 1 continue out.append(dict(k, m3=f)) return out, dropped # ---- FICTIONAL cluster: number of workers and two choices WORKERS = 4 NETWORK_COST = 3 # the cost of moving one row over the network , as a multiple of local work TASK_OVERHEAD = 50 # the fixed cost of handing work to each worker def rows(): v, _ = s1_screen(generate_source()) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return v def trace_id(m): h = 2166136261 for c in str(m): h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF return h def distribute(records, key, workers=WORKERS): """Assigns rows to workers by key. The same key always goes to the same worker.""" buckets = {i: [] for i in range(workers)} for k in records: buckets[trace_id(k[key]) % workers].append(k) return buckets def load(buckets): n = [len(v) for v in buckets.values()] return {"total": sum(n), "busiest": max(n), "least": min(n), "skew": round(max(n) / (sum(n) / len(n)), 4)} def cost(buckets, moved=0, workers=WORKERS): """Makespan = the busiest worker's work (parallel). Total work is counted separately.""" n = [len(v) for v in buckets.values()] return {"duration": max(n) + moved * NETWORK_COST // workers + TASK_OVERHEAD, "total_work": sum(n) + moved * NETWORK_COST, "moved": moved} ``` #### The Partition Key Chooses the Worker The first decision the driver makes is which key to split by. `distribute` produces a number from the key value of a row and takes its remainder by the number of workers; the same key value always goes to the same worker. This property comes from a requirement: if two periods of the same subscriber land on different workers, the consumption difference can never be computed. The decision looks free, because whichever key is chosen, the total number of rows stays the same. The first measurement shows why that is misleading. ```python S = rows() print("rows from the pipeline:", len(S), " workers:", WORKERS) print() print(f"{'key':<12}{'worker load':<26}{'busiest':>9}{'least':>7}{'skew':>11}") for a in ("subscriber", "period", "region"): buckets = distribute(S, a, WORKERS) y = load(buckets) print(f"{a:<12}{str([len(v) for v in buckets.values()]):<26}" f"{y['busiest']:>9}{y['least']:>7}{y['skew']:>11.4f}") ``` ``` rows from the pipeline: 2200 workers: 4 key worker load busiest least skew subscriber [539, 550, 561, 550] 561 539 1.0200 period [600, 600, 400, 600] 600 400 1.0909 region [374, 0, 0, 1826] 1826 0 3.3200 ``` The total of the three rows is the same: in every setup 2200 rows were distributed and no row was lost. What changes is how the rows pile up. On the `subscriber` key, because two hundred distinct values fall into four buckets, the loads are squeezed between 539 and 561; **the skew is 1.02**, that is, the busiest worker is two percent above the average. On the `period` key, twelve values are divided into four, and three buckets get three periods each while one gets two: 600, 600, 400, 600. The `region` row is the silent flaw of partitioning. When five regions are spread over four workers, **two workers receive no rows at all** and one worker is left with 1826 rows. The skew is **3.32**. The cluster holds four machines and the bill for all four is running, but two of them sit idle. No error is raised, no row is dropped and the result is correct; the only thing wrong is the partitioning, and it shows only if it is counted. The resolution of the skew depends on the number of workers: with four workers the smallest step is 0.0018. #### Makespan and Total Work When the loads are turned into a number, the real distinction appears. Because the workers run side by side, the job finishes when the slowest worker finishes; this is the **makespan**, and `cost` computes it as the share of the busiest worker plus the fixed task cost of 50 units. The **total work** the cluster spends, on the other hand, is the sum of the shares of all workers, and it is paid no matter who waits. The block below compares the single-worker baseline with three four-worker setups. ```python print(f"{'setup':<27}{'tasks':>6}{'busiest share':>14}{'makespan':>10}{'total work':>12}") for name, key, n in (("single worker", "subscriber", 1), ("four workers / subscriber", "subscriber", 4), ("four workers / period", "period", 4), ("four workers / region", "region", 4)): buckets = distribute(S, key, n) m = cost(buckets, 0, n) print(f"{name:<27}{n:>6}{max(len(v) for v in buckets.values()):>14}" f"{m['duration']:>10}{m['total_work']:>12}") ``` ``` setup tasks busiest share makespan total work single worker 1 2200 2250 2200 four workers / subscriber 4 561 611 2200 four workers / period 4 600 650 2200 four workers / region 4 1826 1876 2200 ``` The last column is **2200** in all four rows. The work the cluster does has not changed at all, because there is no shuffle and every row is processed exactly once. Even so, the makespan varies between 2250 and 611. It is true that a setup "did 2200 units of work", and it **does not tell the four setups apart at all**. The rule of the course is read from this table: a speedup claim that does not state which one was measured is incomplete. The baseline is a single worker: makespan **2250**, total work **2200**. The difference of 50 units between them is the fixed task cost, and it is paid even with a single worker. Four workers on the `subscriber` key bring the makespan down to **611**; this is the cleanest gain that will be seen in this course, and the reason is equally narrow: no data moves at all, and each worker reads its own share and writes its own share. On the `period` key the makespan rises to **650**, because the busiest worker holds 600 rows instead of 561. The `region` row stands apart once more. Although four workers are run, the makespan is only **1876**; the gain over a single worker is less than one sixth, not one quarter. The workers that sit idle do not touch the makespan at all, because the makespan measures the busiest worker. The same total work, the same correct result, a threefold difference in makespan. A warning is needed about the unit of these numbers. **Makespan is the model's own unit.** Processing one row locally is one unit, moving one row over the network is three units, and handing work to a worker is fifty units; the second and the third values are **choices** written as `NETWORK_COST` and `TASK_OVERHEAD`. The number 2250 does not stand for a second, a millisecond or any clock reading. What the model gives is the **comparison** between setups; the makespan of a single setup carries no meaning on its own. #### The Result Stays the Same, the Cost Changes The answer to the question of what happens when the partitioning is wrong is the hardest one to learn in this course: **nothing**. The result does not come out wrong, no row disappears, no error is raised. The driver merges the partial summaries coming from the workers, and the table it holds is identical in all four setups. The last block shows this by computing the region totals for every setup. ```python def region_totals(buckets): """Each worker summarizes its own share , the driver merges the partial summaries.""" partials = [] for share in buckets.values(): t = {} for k in share: t[k["region"]] = t.get(k["region"], 0) + k["m3"] partials.append(t) final = {} for t in partials: for b, v in t.items(): final[b] = final.get(b, 0) + v return final baseline_result = region_totals(distribute(S, "subscriber", 1)) print("baseline result:", baseline_result) print() print(f"{'setup':<27}{'intermediate rows':>18}{'makespan':>10}{'total work':>12}{'result':>18}") for name, a, n in (("single worker", "subscriber", 1), ("four workers / subscriber", "subscriber", 4), ("four workers / period", "period", 4), ("four workers / region", "region", 4)): buckets = distribute(S, a, n) m = cost(buckets, 0, n) partials = sum(len(set(k["region"] for k in share)) for share in buckets.values()) same = "same as baseline" if region_totals(buckets) == baseline_result else "different" print(f"{name:<27}{partials:>18}{m['duration']:>10}{m['total_work']:>12}{same:>18}") ``` ``` baseline result: {'merkez': 14536, 'kuzey': 13338, 'dogu': 9695, 'bati': 9164, 'guney': 11721} setup intermediate rows makespan total work result single worker 5 2250 2200 same as baseline four workers / subscriber 20 611 2200 same as baseline four workers / period 20 650 2200 same as baseline four workers / region 5 1876 2200 same as baseline ``` The last column says the same thing four times. The five region totals do not change whichever partitioning they are computed under, because the summing operation is insensitive to the order in which the rows are added and to the machine they are added on. The consequence is this: a check that looks at the output of the pipeline **cannot tell apart** the setup in which two workers sit idle from the best setup. All the checks built in the Introduction to Data Engineering course stay green. Wrong partitioning is not a correctness flaw; **it shows only if it is counted.** The number before the third column gives the driver's share. With the `subscriber` and `period` partitionings, all five regions are present in the share of every worker, so 20 intermediate rows reach the driver; with the `region` partitioning, since each region sits on a single worker, only 5 intermediate rows arrive. The merge work of the driver is not added to the total work in this model, but in a real cluster it is a cost too, and the most unbalanced partitioning is the cheapest one here. This is the second example of why choosing a partitioning by looking at a single number is not enough. #### Summary - A distributed run has three roles: the **driver** holds and distributes the plan, the **executor** processes the share given to it, and the **task** is that share itself. In this model one executor counts as one **worker**. - The task here is not the same as the task in orchestration: there it was a node in the graph, here it is the share that falls to a worker inside that node. One node gives birth to four tasks on four workers. - The partition key determines the loads: `subscriber` gives a skew of **1.02**, `period` **1.0909** and `region` **3.32**, and with the `region` key **two of the four workers receive no rows at all**. - Makespan and total work are separate numbers. While the total work stays fixed at **2200** in four setups, the makespan is **2250**, **611**, **650** and **1876**; the total work does not tell the setups apart. - The makespan of the single-worker baseline is **2250** and its total work is **2200**; the difference of 50 units is the fixed cost of handing work to each worker. - Makespan is the model's own unit. `NETWORK_COST = 3` and `TASK_OVERHEAD = 50` are choices, and the numbers carry meaning only when compared with one another. #### Next Step In this lesson the rows were ready in a list and `distribute` handed them out to the workers. In reality the rows a worker processes are **read** from somewhere, and that somewhere can be another machine. However balanced the partitioning is, if every worker pulls its share over the network, the gain of distribution stays below the cost of reading. The next lesson splits the data into blocks, places them on the cluster and measures a single question: what is gained if the worker that processes a block already holds it, and how the number of copies of a block that are stored changes that gain. ### Distributed File Systems Source: https://academia.sh/en/courses/distributed-data-processing/distributed-file-systems Splitting data into blocks, placing them on the cluster, and measuring whether the worker that processes a block already holds it: when 2200 rows are split into 22 blocks of 100 rows, with a single copy and assignment by block order only 5 of the 22 blocks are read locally, the makespan is 1850 and the total work is 5600; at a replication factor of 2 the local blocks rise to 17 and the makespan falls to 1150, at 3 they rise to 22 and the makespan falls to 650 but the stored blocks grow from 22 to 66, and a fourth copy adds nothing to reading; locality-aware assignment reads 22 of 22 blocks locally with a single copy, but because the placement is unbalanced the makespan stays at 750 instead of 650. In the previous lesson the rows sat ready in a list and `distribute` handed them out to the workers. The cost of handing out was counted as zero, because the list was already in memory. This assumption is true on a single machine and false on a cluster: the data sits spread over the disks of the machines, the rows a worker will process may be on another machine, and to see those rows they have to be pulled over the network. This lesson splits the data into blocks, places them on the cluster and measures a single question: what is gained if the worker that processes a block already holds it. The answer is called **data locality**, and its cost is paid in two separate places, one in makespan and one in disk. - **DC9.** Block placement and replication are **fictional**; no real distributed file system product, block server or cloud storage service is called. The seed is **20260218**. - **DC10.** The **2200 rows** coming out of the pipeline are split in order into blocks of **100 rows**; there are **22 blocks**. The block is the resolution of the partitioning: a block is not split and goes in its entirety to a single worker. - **DC11.** **The placement rule:** the first copy of a block is chosen from the trace of its key, and the remaining copies fall on the following workers on the ring. The **replication factor** (`copies`) is the number of workers a block sits on. - **DC12.** If the worker that reads a block holds it, the cost per row is **1**; if it does not, it is **`NETWORK_COST`, that is, 3**. Every makespan and total work measured in this lesson is sensitive to this choice. - **DC13.** The **durability** side of replication, that is, the data surviving when a machine goes down, was measured in the Relational Database Administration course of the Databases curriculum and in the System Design and Distributed Systems curriculum. **The procedure is not repeated here**; this lesson counts replication only through its effect on **reading**. - **DC14.** File layout, the columnar format and partitioning were measured in the Data Storage Architectures course and are **not repeated**. The block here is not a file format but a unit of placement. - **DC15.** The baseline is again **a single worker**: all blocks sit on the same worker, locality is complete, makespan **2250**, total work **2200**. #### Block and Placement What sets a distributed file system apart from a single-machine file system is that it does not store a file as a whole. The file is cut into fixed-size **blocks**, and the blocks are distributed to the machines in the cluster. This has two consequences. First, a file that does not fit on a single machine can be stored. Second, different pieces of the same file can be read **at the same time**, because they sit on different disks. A block is a unit of splitting, not a partition. The block boundary does not look at the key of the rows; it looks at which row of the file it is. So the rows of the same subscriber can be spread over two blocks, and a single block can hold all five regions. The placement rule does not look at the data either: it produces a trace from the order of the block and chooses a worker with that trace. The copies fall on the following workers on the ring, so the loss of one worker does not take away all the copies of a single block. ```python # FICTIONAL. The source, pipeline and distributed file system are defined in the lesson; # no real file system product or block server is called. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records def s1_screen(records): REQUIRED = ("subscriber", "period", "region", "index", "unit") keep = [k for k in records if all(a in k for a in REQUIRED)] return keep, len(records) - len(keep) def s2_dedupe(records): seen, keep = set(), [] for k in records: key = (k["subscriber"], k["period"]) if key in seen: continue seen.add(key) keep.append(k) return keep, len(records) - len(keep) def s3_consumption(records): table = {(k["subscriber"], k["period"]): k for k in records} out, dropped = [], 0 for k in records: i = PERIODS.index(k["period"]) if i == 0: continue prev = table.get((k["subscriber"], PERIODS[i - 1])) if prev is None: dropped += 1 continue try: f = int(k["index"]) - int(prev["index"]) except (TypeError, ValueError): dropped += 1 continue out.append(dict(k, m3=f)) return out, dropped WORKERS, NETWORK_COST, TASK_OVERHEAD = 4, 3, 50 def rows(): v, _ = s1_screen(generate_source()) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return v def trace_id(m): h = 2166136261 for c in str(m): h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF return h # ---- FICTIONAL distributed file system: block , placement , read BLOCK_SIZE = 100 def split_blocks(records, size=BLOCK_SIZE): """Rows are cut into fixed-size blocks in order.""" return [records[i:i + size] for i in range(0, len(records), size)] def placement(block_no, copies, workers=WORKERS): """The first copy follows the block's trace , the rest go to the next worker on the ring.""" first = trace_id(f"blok{block_no}") % workers return [(first + j) % workers for j in range(copies)] def read_blocks(blocks, copies, locality, workers=WORKERS): """One worker reads each block. A worker that does not hold the block pulls the rows over the network.""" work = {i: 0 for i in range(workers)} local = 0 for b, block in enumerate(blocks): holders = placement(b, copies, workers) reader = holders[0] if locality else b % workers if reader in holders: local += 1 work[reader] += len(block) else: work[reader] += len(block) * NETWORK_COST return {"local": local, "remote": len(blocks) - local, "stored": len(blocks) * copies, "duration": max(work.values()) + TASK_OVERHEAD, "total_work": sum(work.values())} ``` #### Locality Is a Coincidence In the first setup the driver gives the blocks to the workers **in order**: block zero to worker zero, block one to worker one, and so on. This assignment equalizes the load, because about the same number of blocks falls on every worker. It does not take the placement into account at all; that the block a worker reads is held by that worker is a coincidence. The block below counts, across replication factors, in how many blocks that coincidence happens. ```python BLOCKS = split_blocks(rows()) single = read_blocks(BLOCKS, 1, False, 1) print("blocks:", len(BLOCKS), " each", BLOCK_SIZE, "rows") print("single-worker baseline -> makespan:", single["duration"], " total work:", single["total_work"]) print() print(f"{'copies':>8}{'local blocks':>14}{'remote blocks':>15}{'stored':>8}{'makespan':>10}{'total work':>12}") for r in (1, 2, 3, 4): o = read_blocks(BLOCKS, r, False) print(f"{r:>8}{o['local']:>14}{o['remote']:>15}{o['stored']:>8}" f"{o['duration']:>10}{o['total_work']:>12}") ``` ``` blocks: 22 each 100 rows single-worker baseline -> makespan: 2250 total work: 2200 copies local blocks remote blocks stored makespan total work 1 5 17 22 1850 5600 2 17 5 44 1150 3200 3 22 0 66 650 2200 4 22 0 88 650 2200 ``` The first row shows the hidden bill of distributing. With a single copy only **five** of the twenty-two blocks are read locally; the remaining seventeen blocks cross the network and each of their rows is counted three times. The total work becomes **5600** instead of 2200, and the makespan **1850**. The number five is close to the expected value of a coincidence: with four workers the probability that the reader and the holder overlap is one in four, and one quarter of twenty-two blocks is 5.5. In the previous lesson the same four workers had brought the makespan down to 611, but that measurement had counted reading as free. Once reading is counted, the four-worker setup falls only from 2250 to 1850 compared with the single-worker baseline; the gain is less than one fifth, not one quarter. The second and third rows give what replication adds to reading. When every block has two copies, the locally read blocks rise from 5 to **17** and the makespan falls to **1150**; with three copies all twenty-two blocks are read locally, the total work returns to **2200** and the makespan becomes **650**. The network traffic disappears completely. These numbers depend on the choice `NETWORK_COST = 3`: if the cost of moving were 1, all four rows would give the same total work and the contribution of replication to reading would not be visible. The fourth row is the limit of replication. With four copies the stored blocks rise to **88**, that is, the data grows fourfold, but the local blocks and the makespan stay **exactly the same** as with three copies. After the third copy there is nothing that is paid for in reading; the justification of the fourth copy lies elsewhere, and that justification is durability. Durability is not measured in this course; it was measured in the prerequisite courses. #### Locality or Balance There is an alternative to buying locality with replication: the driver gives the blocks not in order but to **the worker that holds them**. This assignment provides full locality even with a single copy. In return it hands the balance of the load over to the placement rule, and that rule does not try to equalize the load. The last block compares the two assignments at two replication factors. ```python print(f"{'assignment':<16}{'copies':>8}{'local':>7}{'stored':>8}{'makespan':>10}{'total work':>12}") for name, y, r in (("block order", False, 1), ("block order", False, 3), ("locality-aware", True, 1), ("locality-aware", True, 3)): o = read_blocks(BLOCKS, r, y) print(f"{name:<16}{r:>8}{o['local']:>7}{o['stored']:>8}" f"{o['duration']:>10}{o['total_work']:>12}") ``` ``` assignment copies local stored makespan total work block order 1 5 22 1850 5600 block order 3 22 66 650 2200 locality-aware 1 22 22 750 2200 locality-aware 3 22 66 750 2200 ``` The third row catches full locality with a single copy: twenty-two of twenty-two blocks are read locally, the total work falls to **2200** and the stored blocks stay at **22**. Assignment by block order obtains the same result only with three copies, that is, by storing **66 blocks**. Locality-aware assignment saves three times the disk. On the other hand its makespan is **750**, **100 units** more than the three-copy case of assignment by block order. The reason does not show in the table but lies in the placement rule: the blocks are not evenly distributed to the workers, and the worker that holds the most blocks reads seven blocks while the one that holds the fewest reads four. Locality-aware assignment cannot choose the reader, because if it did, locality would break. The fourth row writes this once more: raising the replication to three does not change the makespan at all in locality-aware assignment, because the worker that holds the first copy has not changed. The choice between the two rows cannot be reduced to a single number. For a hundred units of makespan, forty-four more blocks are stored, or in place of forty-four blocks a hundred units of makespan are paid. Which one is cheaper depends on the disk of the cluster and on how often its job runs, and this decision can be made only when both numbers are written. #### Block Size Is the Resolution of the Partitioning The reason the locality-aware assignment stays at 750 does not show in the table, because the table does not write how the blocks fall on the workers. The last block first prints that distribution, then changes the block size and measures how much of the balance comes from the size. ```python S = rows() holder_count = {i: 0 for i in range(WORKERS)} for b in range(len(BLOCKS)): holder_count[placement(b, 1)[0]] += 1 print("blocks holding the first copy:", [holder_count[i] for i in range(WORKERS)]) print() print(f"{'block size':>12}{'blocks':>8}{'local':>7}{'makespan':>10}{'total work':>12}") for size in (25, 50, 100, 220, 550): chunks = split_blocks(S, size) o = read_blocks(chunks, 1, True) print(f"{size:>12}{len(chunks):>8}{o['local']:>7}{o['duration']:>10}{o['total_work']:>12}") ``` ``` blocks holding the first copy: [7, 4, 4, 7] block size blocks local makespan total work 25 88 88 625 2200 50 44 44 650 2200 100 22 22 750 2200 220 10 10 710 2200 550 4 4 600 2200 ``` The first line gives the missing justification. Twenty-two blocks fell on four workers as **7, 4, 4, 7**; the worker that holds the most blocks reads seven blocks, that is, 700 rows, and with 700 plus the fixed cost of 50 the makespan is **750**. The placement rule does not try to equalize the load, it looks at the trace of the block; the imbalance is not a flaw but the natural result of the rule. The sweep shows how this imbalance relates to the block size. With blocks of twenty-five rows eighty-eight blocks are formed and the makespan falls to **625**, with fifty rows it is **650**: as the block gets smaller, the share of each worker approaches the average, because the weight of a single block falling in the wrong place decreases. The local blocks equal the number of blocks in every row and the total work is fixed at **2200**; the block size does not touch the network traffic, only the balance. The last two rows prevent drawing a one-directional rule. Ten blocks of two hundred and twenty rows bring the makespan down to **710**, and four blocks of five hundred and fifty rows to **600**, which is better than both the hundred-row setup and the twenty-five-row one. The reason is this: the four blocks happened to fall on four distinct workers, and the distribution came out as 1, 1, 1, 1. A small block improves the balance **in expectation**; it does not guarantee it in a single setup. In this model shrinking the block also has no visible cost, because the fixed task cost is counted once per worker, not per block; in a real cluster every block gives birth to a task, and that cost comes back with small blocks. #### Summary - A distributed file system cuts data into fixed-size **blocks** and spreads them over the cluster. 2200 rows become **22 blocks** of 100 rows; a block is not split and goes in its entirety to a single worker. - **Data locality** is the worker that reads a block already holding it. If it is local, the cost per row is 1; if not, it is **`NETWORK_COST`, that is, 3**; all the numbers measured are sensitive to this choice. - With assignment by block order and a single copy, only **5** of the 22 blocks are read locally: makespan **1850**, total work **5600**. At a replication factor of 2 the local blocks are **17** and the makespan **1150**, at 3 they are **22** and **650**. - The contribution of replication to reading ends at the third copy: with four copies the stored blocks rise to **88**, the local blocks and the makespan do not change. The justification for the fourth copy is durability, and it was measured in the prerequisite courses. - Locality-aware assignment gives full locality with a single copy (**22** local blocks, **22** stored blocks), but because the placement is unbalanced the makespan stays at **750** instead of **650**. Locality and balance are not bought at the same time; the assignment cannot be chosen without writing both. - Block size is the resolution of the balance: the 22 blocks fall on the workers as **7, 4, 4, 7**, and with blocks of 25 rows the makespan falls to **625**. But the rule is not one-directional; four blocks of 550 rows happen to fall in a balanced way and give **600**. A small block improves the balance in expectation, it does not guarantee it. #### Next Step In these two lessons every worker read its own share and processed its own share; no row passed from one worker to another. This is possible only when the computation is done row by row. When a result such as the total per region is wanted, the rows of the same region are spread over different workers and they have to come together before they are summed. The next lesson builds a model that splits the computation into two steps and asks a single question: which step stays local, which one touches the network, and how much does the cost of producing the same result change with this distinction. ### MapReduce Model Source: https://academia.sh/en/courses/distributed-data-processing/mapreduce-model Splitting a computation into a step that works independently per row and a step that sums per key, and measuring that only one of the two steps touches the network: the map processes 2200 rows locally and moves 0 rows, the makespan is 611 and the total work stays at 2200, while the reduce step that produces the same result moves 1738 rows from the subscriber layout to the region key and raises the total work to 7414 and the makespan to 1914; when every worker reduces its own share before moving it, the moved rows are only 15 of 20 intermediate rows, the total work falls to 2245 and the makespan to 622, and the five region totals presented by the three setups remain exactly the same. In the previous two lessons every worker read its own share and processed its own share; no row passed from one worker to another. This was possible because the computations belonged to a special class: each row could be processed on its own. Computing the consumption difference needs two periods of a subscriber, but the same subscriber is already on a single worker; screening and deduplication also look at the row itself. When a result such as the total per region is wanted, this ends. The rows of a region are spread over two hundred subscribers, in a cluster partitioned by subscriber they have fallen on all four of the four workers, and they have to come together before they are summed. This lesson builds a model that splits the computation into two steps and asks a single thing: which step stays local, and which one touches the network. - **DC16.** The cluster, the pipeline and the two-step model are **fictional**; no real distributed processing framework, stream engine or cloud service is called. The seed is **20260218**. - **DC17.** **MapReduce** is the name of a model and does not denote a product. The map step applies a function to each row independently and produces **(key, value)** pairs, in this lesson one pair per row; the reduce step folds all the values of the same key into a single result. - **DC18.** The map step has **no dependency between rows**; so it can process the data where it is and does not touch the network. - **DC19.** The reduce step requires **all** the values of a key to be gathered on a single worker. If the data is not in that layout, a **shuffle** occurs, and every row that changes workers crosses the network. - **DC20.** The cost of a shuffled row is **`NETWORK_COST`, that is, 3**; this is a choice, and every makespan and total work measured in this lesson is sensitive to it. - **DC21.** The baseline is again **a single worker**: no shuffle, makespan **2250**, total work **2200**. - **DC22.** The rows enter the lesson in the layout they arrived from the previous lessons, that is, **partitioned by the `subscriber` key**. The size of the shuffle depends on this starting layout. - **DC23.** Makespan is the model's own unit and is not raw milliseconds; the numbers carry meaning only when compared with one another. #### The Two-Step Model The distinction the model draws comes from the computation itself. The question of whether one needs to look at other rows to do one row of a computation has two answers, and these two answers mean two steps. **Map** is the part that says no: it takes a key and a value out of the row and never looks at its neighbor. **Reduce** is the part that says yes: it wants all the values that belong to a key and cannot produce a single correct number without having them. The value of this distinction is that it says **in advance** which part of the computation costs network. No matter how many workers are added to the map step, no network traffic arises. The traffic of the reduce step, on the other hand, depends on the choice of the key and on the layout the data enters in. The block below builds the model with three functions: `map_rows` produces pairs, `local_reduce` folds a worker's own share, and `reduce_pairs` sums the result after the shuffle. ```python # FICTIONAL. The source, pipeline and cluster are defined in the lesson; no real distributed # processing framework, stream engine or cloud service is called. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records def s1_screen(records): REQUIRED = ("subscriber", "period", "region", "index", "unit") keep = [k for k in records if all(a in k for a in REQUIRED)] return keep, len(records) - len(keep) def s2_dedupe(records): seen, keep = set(), [] for k in records: key = (k["subscriber"], k["period"]) if key in seen: continue seen.add(key) keep.append(k) return keep, len(records) - len(keep) def s3_consumption(records): table = {(k["subscriber"], k["period"]): k for k in records} out, dropped = [], 0 for k in records: i = PERIODS.index(k["period"]) if i == 0: continue prev = table.get((k["subscriber"], PERIODS[i - 1])) if prev is None: dropped += 1 continue try: f = int(k["index"]) - int(prev["index"]) except (TypeError, ValueError): dropped += 1 continue out.append(dict(k, m3=f)) return out, dropped WORKERS, NETWORK_COST, TASK_OVERHEAD = 4, 3, 50 def rows(): v, _ = s1_screen(generate_source()) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return v def trace_id(m): h = 2166136261 for c in str(m): h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF return h def distribute(records, key, workers=WORKERS): buckets = {i: [] for i in range(workers)} for k in records: buckets[trace_id(k[key]) % workers].append(k) return buckets def shuffle(buckets, new_key, workers=WORKERS): """Repartition. A row that changes workers crosses the network.""" new = {i: [] for i in range(workers)} moved = 0 for old, records in buckets.items(): for k in records: target = trace_id(k[new_key]) % workers if target != old: moved += 1 new[target].append(k) return new, moved def cost(buckets, moved=0, workers=WORKERS): """Makespan = the busiest worker's work (parallel). Total work is counted separately.""" n = [len(v) for v in buckets.values()] return {"duration": max(n) + moved * NETWORK_COST // workers + TASK_OVERHEAD, "total_work": sum(n) + moved * NETWORK_COST, "moved": moved} # ---- FICTIONAL MapReduce: two steps def map_rows(share, key_fn, value_fn): """Produces one (key , value) pair per row. Local , never looks at a neighbor.""" return [{"key": key_fn(k), "value": value_fn(k)} for k in share] def local_reduce(pairs): """Sums the pairs in a worker's own share per key.""" acc = {} for c in pairs: acc[c["key"]] = acc.get(c["key"], 0) + c["value"] return [{"key": a, "value": v} for a, v in acc.items()] def reduce_pairs(buckets): """After the shuffle each worker sums its own keys , the driver merges.""" final = {} for share in buckets.values(): for c in local_reduce(share): final[c["key"]] = final.get(c["key"], 0) + c["value"] return final ``` #### The Map Stays Local The first measurement counts the map step alone. The rows enter the lesson partitioned by the `subscriber` key, and the map takes the region out of every row as the key and the consumption as the value. No worker looks at the row of another worker, so `shuffle` is never called and the moved rows are zero. ```python S = rows() K = distribute(S, "subscriber", WORKERS) MAPPED = {i: map_rows(share, lambda k: k["region"], lambda k: k["m3"]) for i, share in K.items()} print("single-worker baseline ->", cost(distribute(S, "subscriber", 1), 0, 1)) print("mapped pairs :", sum(len(v) for v in MAPPED.values())) print("map step ->", cost(K, 0)) ``` ``` single-worker baseline -> {'duration': 2250, 'total_work': 2200, 'moved': 0} mapped pairs : 2200 map step -> {'duration': 611, 'total_work': 2200, 'moved': 0} ``` Two thousand two hundred pairs come out of two thousand two hundred rows; the map does not change the number of rows, it changes the shape of the row. The total work is **2200**, as in the single-worker baseline, and four workers bring the makespan down from 2250 to **611**. This is the cleanest gain that will be seen in this course, and the reason is the definition of the map: because there is no dependency between rows, the computation is correct whatever the partition key is, and no row changes place. The practical conclusion from this is: how much of a computation fits into the map step tells how much of it will parallelize for free. Filtering, extracting a field, converting a unit and classifying per row belong to this step. Whatever does not belong is left to the second step. #### The Reduce Touches the Network The second step wants all the values of the same key. The region key is spread over all four of the four workers, so the worker that will compute the `merkez` (central) total holds only a quarter of the `merkez` pairs. The shuffle arises for this reason. The block below compares two reduce setups and checks the result of both against the single-worker baseline. ```python PRE_REDUCED = {i: local_reduce(c) for i, c in MAPPED.items()} BASELINE = reduce_pairs({0: map_rows(S, lambda k: k["region"], lambda k: k["m3"])}) print("intermediate rows after pre-reduction:", sum(len(v) for v in PRE_REDUCED.values())) print() print(f"{'setup':<28}{'moved':>7}{'makespan':>10}{'total work':>12}{'result':>17}") for name, source in (("reduce, direct", MAPPED), ("reduce, with pre-reduction", PRE_REDUCED)): y, moved = shuffle(source, "key", WORKERS) m = cost(K, moved) same = "same as baseline" if reduce_pairs(y) == BASELINE else "different" print(f"{name:<28}{moved:>7}{m['duration']:>10}{m['total_work']:>12}{same:>17}") print() print("baseline result:", BASELINE) ``` ``` intermediate rows after pre-reduction: 20 setup moved makespan total work result reduce, direct 1738 1914 7414 same as baseline reduce, with pre-reduction 15 622 2245 same as baseline baseline result: {'merkez': 14536, 'kuzey': 13338, 'dogu': 9695, 'bati': 9164, 'guney': 11721} ``` The first row writes the difference between the two steps. The same two thousand two hundred pairs move **0** rows in the map step and **1738** rows in the reduce step; 79% of the two thousand two hundred pairs change workers. The total work rises from **2200** to **7414**, and the makespan from 611 to **1914**. The 1738 rows that cross the network cost three units per row, that is, 5214 units, and this is the whole of the increase in total work. The number here does not come from the difficulty of the computation. Taking a region total is no heavier an operation than taking a consumption difference; what is heavy is that the data is partitioned by **the wrong key**. In the previous lesson it was measured that partitioning by the `region` key raised the skew to 3.32. There is no free key: partitioning by `subscriber` balances the map and forces the reduce into a shuffle, while partitioning by `region` zeroes the shuffle and leaves two workers idle. #### Pre-Reduction Produces the Same Result Cheaply The second row gives the real gain of the model. If every worker sums the pairs in its own share **before moving them**, the two thousand two hundred pairs coming out of the four workers fall to **20 intermediate rows**: each of the four workers produces one partial total for at most five regions. The shuffle now moves these twenty rows, and **15** of them change workers. The makespan becomes **622** and the total work **2245**; compared with the direct shuffle, the total work has fallen from **7414**, and a gap of nearly threefold has closed. The last column says the same thing in both setups. The five region totals are **exactly the same** as the single-worker baseline both in the direct shuffle and in the pre-reduction. This is the most unsettling observation of the course: the setup of 7414 units and the setup of 2245 units give the same output, and no check that looks at the output can tell them apart. What is wrong is not the computation but the cost; and the cost shows only if it is counted. Where the number twenty comes from also gives the limit of the gain. Because each of the four workers sees at most five distinct regions, pre-reduction could squeeze 2200 pairs into 20 rows; as the **number of distinct values** of the key grows, the squeeze decreases. The last block measures this over four reduce keys. ```python KEYS = (("region", lambda k: k["region"]), ("period", lambda k: k["period"]), ("region+period", lambda k: k["region"] + "|" + k["period"]), ("subscriber", lambda k: k["subscriber"])) print(f"{'reduce key':<14}{'distinct':>10}{'intermediate rows':>19}" f"{'direct':>8}{'pre-reduced':>13}{'total work':>12}") for name, f in KEYS: mapped = {i: map_rows(share, f, lambda k: k["m3"]) for i, share in K.items()} combined = {i: local_reduce(c) for i, c in mapped.items()} _, moved_direct = shuffle(mapped, "key", WORKERS) _, moved_pre = shuffle(combined, "key", WORKERS) print(f"{name:<14}{len(set(f(k) for k in S)):>10}{sum(len(v) for v in combined.values()):>19}" f"{moved_direct:>8}{moved_pre:>13}{cost(K, moved_pre)['total_work']:>12}") ``` ``` reduce key distinct intermediate rows direct pre-reduced total work region 5 20 1738 15 2245 period 11 44 1651 33 2299 region+period 55 220 1653 165 2695 subscriber 200 200 0 0 2200 ``` The fourth and the third column are two writings of the same computation, and the ratio between them closes together with the key. On the `region` key, which has five distinct values, the moved rows fall from 1738 to **15**; on the `region+period` key, which has fifty-five distinct values, they fall from 1653 to **165**, and the total work becomes **2695** instead of 2245. The squeeze ratio is determined by the number of intermediate rows, and that is the number of distinct values of the key multiplied by four workers. If the number of distinct values approaches the number of rows, pre-reduction cannot squeeze anything and pays the same cost as the direct shuffle. The last row is a separate case. When the reduce key is the same as the partition key, that is, when reducing to the `subscriber` key, the moved rows are **0** even in the direct writing: all the rows of every subscriber are already on a single worker. The shuffle is not a necessity but the cost of the **mismatch** between the reduce key and the layout the data enters in. The total work of this row is **2200**, the same as the map step itself. Pre-reduction has a condition, and not every computation meets it. The **order and grouping of the operation must be changeable**: summing is like this, because the sum of partial sums equals the sum of the whole. The smallest and the largest value are also like this. The median is not; the median of a worker's share cannot be computed and summed with the others, because the median wants to see all the values. A distinct count is not either: when the distinct subscriber counts in the shares of two workers are summed, the common subscribers are counted twice. For these computations either all the rows are moved or an approximate result is accepted. The model does not make every computation cheaper; it **says** which ones it makes cheaper. #### Summary - **MapReduce** splits a computation into two steps: the **map** is applied to each row independently and produces a (key, value) pair, the **reduce** folds all the values of a key into a single result. - The map step is local. 2200 rows produce 2200 pairs, the moved rows are **0**, the total work stays at **2200** and four workers bring the makespan from 2250 down to **611**. - The reduce step touches the network. In passing from the `subscriber` layout to the `region` key, **1738** rows change workers; the total work rises to **7414** and the makespan to **1914**. The whole of the increase is the three-unit cost of the moved rows. - When every worker reduces its own share before moving it, 2200 pairs fall to **20 intermediate rows**, **15** of them are moved, and the same result is produced for a total work of **2245** and a makespan of **622**. - The five region totals presented by the three setups are exactly the same. A check that looks at the output **cannot tell apart** the setup of 7414 units and the one of 2245; the difference is only in the cost. - Pre-reduction is valid only for operations whose order and grouping can be changed. Sum, minimum and maximum are suitable; median and distinct count are not. #### Next Step In three lessons the job was split, placed and run. In every measurement the whole computation ran: all two thousand two hundred rows were mapped, all were moved, all were reduced. Yet in a case where a report asks for only one region or the last three periods, most of these rows should never have been touched, and which computation is **really needed** was never asked. How the workers are shared across the cluster was not asked either; four workers were assumed ready in every measurement. The next two lessons measure these two questions: how many rows a computation that is not used in the result leaves untouched, and where the fixed cost of increasing the number of workers eats the gain. ### Distributed Datasets Source: https://academia.sh/en/courses/distributed-data-processing/distributed-datasets A distributed dataset consisting of partitions and a plan that has not run yet, and counting the computation that lazy evaluation eliminates: while eleven partitions are spread over four workers as three, three, three and two, the eager setup opens 44 partitions, reads 6556 rows and writes 3727 intermediate rows for a result of five rows, the lazy setup gives the same result by opening a single partition and reading 200 rows, leaving 2000 rows never read; in a result of a hundred rows the rows read triple while the makespan stays at 250 because three partitions run on three separate workers; in the count action that asks for the whole result the lazy setup reads 2200 rows and the difference comes only from the unused branch and from the intermediate rows not written; because building the plan reads no rows, the type error in the plan appears only at the first action and after 200 rows have been read. In the previous three lessons the job was split and run: rows were distributed to workers by a key, the place of the blocks determined where the read would go, and the map was separated from the reduce. The three shared an assumption that was written nowhere: every computation the driver writes runs. The moment a transformation is defined, the cluster executes it, writes its result somewhere and waits for the next one. This lesson removes that assumption and asks a single question: how much of the computation written is **really used** in the result. The answer is given with two numbers: the partitions never opened and the rows never read. The structure that makes these two numbers possible is the **distributed dataset**: it consists of partitions and a **plan that has not run yet**. - **DC41.** The source, the pipeline and the cluster are **fictional**; no real distributed processing framework or resource manager is called. The seed is **20260218**, and the measurement does not depend on the run. - **DC42.** **Makespan is the model's own unit**, not raw milliseconds: the number of rows read by the busiest worker plus the task overhead. - **DC43.** **The partition is the unit of data, the worker is the unit of computation.** The partition key is `period`; because the consumption stage drops the first period, **11 partitions** remain, each with **200 rows**. - **DC44.** The partitions are handed to the workers in turn (0, 1, 2, 3, 0, ...); three workers hold three partitions each, one holds two. - **DC45.** A **transformation** defines a new dataset and returns no result; an **action** returns a result to the driver and is the only thing that starts the computation. - **DC46.** The units counted are: opened partitions, rows read, intermediate rows written, makespan. There is **no shuffle** in this lesson, so the total work equals the rows read. - **DC47.** The baseline is **eager evaluation**: every transformation runs the moment it is defined and writes its intermediate result. - **DC48.** Resolution: **11** partitions, **200** rows per partition, **2200** in total. No ratio is printed on the presentation; the value itself is written. #### Transformation and Action There are two kinds of calls in a distributed cluster, and the difference between them is the whole of this lesson. A **transformation** defines a new dataset from an existing one: filtering, mapping, repartitioning. What it returns is not data but a **recipe**. An **action** returns a result to the driver: counting, fetching the first rows, summing. Only an action starts the computation. **Lazy evaluation** is the consequence of this distinction. The transformations are written into a plan, and no partition is opened until the first action arrives. When the action arrives the plan is known as a whole, and the cluster runs only the partitions that the result requires. Its opposite is **eager evaluation**: every transformation runs on the line where it is written, writes its intermediate result, and the next transformation reads it again. The first block builds the fictional source and the first three stages of the pipeline. The 2200 rows coming out of it are the only concrete axis of the course. ```python # FICTIONAL. The meter readings, the first three stages of the pipeline and the fictional cluster are # defined in the lesson; no real distributed processing framework or resource manager is called. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records def s1_screen(records): REQUIRED = ("subscriber", "period", "region", "index", "unit") keep = [k for k in records if all(a in k for a in REQUIRED)] return keep, len(records) - len(keep) def s2_dedupe(records): seen, keep = set(), [] for k in records: key = (k["subscriber"], k["period"]) if key in seen: continue seen.add(key) keep.append(k) return keep, len(records) - len(keep) def s3_consumption(records): table = {(k["subscriber"], k["period"]): k for k in records} out, dropped = [], 0 for k in records: i = PERIODS.index(k["period"]) if i == 0: continue prev = table.get((k["subscriber"], PERIODS[i - 1])) if prev is None: dropped += 1 continue try: f = int(k["index"]) - int(prev["index"]) except (TypeError, ValueError): dropped += 1 continue out.append(dict(k, m3=f)) return out, dropped WORKERS = 4 TASK_OVERHEAD = 50 # the fixed cost of handing work to each worker , a choice def rows(): v, _ = s1_screen(generate_source()) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return v ``` #### The Partition Is the Unit of Data, the Worker Is the Unit of Computation The number of partitions and the number of workers are not the same thing, and the gain this lesson measures arises precisely from this distinction. The **partition** is a divided piece of the data and can be read on its own. The **worker** is the computing unit that reads those pieces, and it can hold more than one partition. Here 11 partitions are spread over four workers: three workers take three partitions each, one takes two. Because there are more partitions than workers, an action can open **only some of the partitions**, and the rows of a partition that is not opened are never read. How a partition is used on the storage side, that is, deciding by the file layout which partition will never be read, was measured in the Data Storage Architectures course and is not repeated here. The elimination here is on the computation side: what eliminates the partition is not the file layout but **what the action asks for**. ```python S = rows() PARTITIONS = {d: [k for k in S if k["period"] == d] for d in PERIODS} PARTITIONS = {d: v for d, v in PARTITIONS.items() if v} NAMES = sorted(PARTITIONS) OWNER = {a: i % WORKERS for i, a in enumerate(NAMES)} COUNTER = {} def reset(): COUNTER.update({"partitions": 0, "read": 0, "intermediate": 0, "worker": {i: 0 for i in range(WORKERS)}}) def read_partition(name, data): COUNTER["partitions"] += 1 COUNTER["read"] += len(data) COUNTER["worker"][OWNER[name]] += len(data) return data class Lazy: """A transformation is written into the plan , it does not run. The computation starts only with an action.""" def __init__(self, partitions, plan=()): self.partitions, self.plan = partitions, tuple(plan) def select(self, f): return Lazy(self.partitions, self.plan + (("select", f),)) def map_rows(self, f): return Lazy(self.partitions, self.plan + (("map", f),)) def _run(self, name): out = [] for k in read_partition(name, self.partitions[name]): for kind, f in self.plan: if kind == "select" and not f(k): k = None break if kind == "map": k = f(k) if k is not None: out.append(k) return out def count(self): return sum(len(self._run(a)) for a in NAMES) def first(self, n): c = [] for a in NAMES: c += self._run(a) if len(c) >= n: break return c[:n] class Eager(Lazy): """Every transformation runs the moment it is defined and writes its intermediate result.""" def _apply(self, f, select): new = {} for a in NAMES: v = [k for k in read_partition(a, self.partitions[a]) if f(k)] if select \ else [f(k) for k in read_partition(a, self.partitions[a])] COUNTER["intermediate"] += len(v) new[a] = v return Eager(new) def select(self, f): return self._apply(f, True) def map_rows(self, f): return self._apply(f, False) def _run(self, name): return self.partitions[name] print("partitions:", len(PARTITIONS), " rows per partition:", sorted(set(len(v) for v in PARTITIONS.values()))) print("partitions per worker:", [sum(1 for a in NAMES if OWNER[a] == i) for i in range(WORKERS)]) ``` ``` partitions: 11 rows per partition: [200] partitions per worker: [3, 3, 3, 2] ``` Both classes offer the same interface; the difference is only **when** the computation runs. In the `Lazy` class `select` returns a new plan; in the `Eager` class the same call reads all the partitions. #### The Computation Laziness Eliminates The job measured is an ordinary script written by the driver: the readings of the residential subscribers are filtered, the row is reduced to four fields, and those above 30 m3 are selected. The script also defines a `commercial` dataset and **no action asks for it**; this is the measure of the computation held in hand. Four setups run the same script and the action changes: `first(5)` asks for five rows, `first(100)` for a hundred rows, `count` for the whole result. ```python FIELDS = ("subscriber", "period", "region", "m3") def script(K): """The job the driver writes: the readings of residential subscribers above 30 m3.""" residential = K.select(lambda k: k["type"] == "residential") K.select(lambda k: k["type"] == "commercial") # no action asks for this return residential.map_rows(lambda k: {a: k[a] for a in FIELDS}).select(lambda k: k["m3"] > 30) def measure(mode, action): reset() y = script(Lazy(PARTITIONS) if mode == "lazy" else Eager(PARTITIONS)) n = y.count() if action == "count" else len(y.first(int(action))) return (COUNTER["partitions"], COUNTER["read"], COUNTER["intermediate"], max(COUNTER["worker"].values()) + TASK_OVERHEAD, n) print(f"{'mode':<9}{'action':>8}{'opened partitions':>19}{'read':>6}" f"{'intermediate rows':>19}{'makespan':>10}{'result':>8}") for mode, action in (("eager", "5"), ("lazy", "5"), ("lazy", "100"), ("eager", "count"), ("lazy", "count")): b, o, mid, makespan, n = measure(mode, action) print(f"{mode:<9}{action:>8}{b:>19}{o:>6}{mid:>19}{makespan:>10}{n:>8}") ``` ``` mode action opened partitions read intermediate rows makespan result eager 5 44 6556 3727 1838 5 lazy 5 1 200 0 250 5 lazy 100 3 600 0 250 100 eager count 44 6556 3727 1838 449 lazy count 11 2200 0 650 449 ``` The first two rows are the finding of the lesson, and **the result column is the same in both**. The eager setup opens 44 partitions for a result of five rows, reads 6556 rows and writes 3727 intermediate rows; the makespan is 1838. The lazy setup gives the same five rows by opening a **single partition**, reading **200 rows** and writing no intermediate rows; the makespan is 250. The remaining **10 partitions are never opened and 2000 rows are never read.** The 6556 rows of the eager setup come from two places. The first: the 2200 rows of the source are read **twice**, because the `residential` and `commercial` filters pass separately over all the partitions. The second: the `commercial` branch reads 2200 rows and writes 1122 intermediate rows and **never enters the result**. In the lazy setup that branch is nothing but a plan, and a plan does not run. The third row carries the rule of the course on its own. `first(100)` opens three partitions and the rows read rise from 200 to 600, so the **total work triples**, but the makespan **stays at 250**. The reason is DC44: because of the sequential distribution the first three partitions are on three separate workers and all three run at the same time. The busiest worker still reads 200 rows. **A speedup claim that does not state which one was measured falls short here**: between the same two setups, the total work is three times and the makespan one time. The last two rows draw the limit of laziness. Because `count` asks for the whole result, the lazy setup too opens all 11 partitions and reads 2200 rows. The difference no longer comes from the skipped partitions but from **the unused branch and the 3727 intermediate rows not written**: 2200 rows against 6556, a makespan of 650 against 1838. #### Partition Size Is the Resolution of Laziness Laziness can skip a partition; it cannot skip a row from inside a partition. So the smallest piece that can be skipped is the partition itself, and **the number of partitions is a decision**. The same 2200 rows are put into four, eleven and forty-four partitions; the measured job and the action do not change. ```python SAVED = (PARTITIONS, NAMES, OWNER) def repartition(parts): """The number of partitions is a decision: the smallest unit laziness can skip is one partition.""" global PARTITIONS, NAMES, OWNER PARTITIONS = {i: S[i * len(S) // parts:(i + 1) * len(S) // parts] for i in range(parts)} NAMES, OWNER = sorted(PARTITIONS), {i: i % WORKERS for i in range(parts)} print(f"{'partitions':>12}{'rows/partition':>16}{'first(5) opened':>17}{'first(5) read':>15}" f"{'first(5) makespan':>19}{'count makespan':>16}") for parts in (4, 11, 44): repartition(parts) b, o, _, makespan, _ = measure("lazy", "5") _, _, _, full, _ = measure("lazy", "count") print(f"{parts:>12}{len(S) // parts:>16}{b:>17}{o:>15}{makespan:>19}{full:>16}") PARTITIONS, NAMES, OWNER = SAVED ``` ``` partitions rows/partition first(5) opened first(5) read first(5) makespan count makespan 4 550 1 550 600 600 11 200 1 200 250 650 44 50 1 50 100 600 ``` The opened partitions are **one** in all three setups; what changes is how many rows that partition carries. For a result of five rows the rows read fall from 550 to 50 and the makespan from 600 to 100. **The gain of laziness is measured by the partition size**: in a coarsely divided dataset laziness is written but finds nothing to eliminate. The last column shows the opposite direction. In the `count` action, which asks for the whole result, 11 partitions give a makespan of **650**, four and forty-four partitions give **600**. The reason is that the number of partitions is not divisible by the number of workers: because 11 partitions spread over four workers as [3, 3, 3, 2], the busiest worker reads 600 rows while one worker is left idle at 400 rows. Thinning the partitions also has an invisible cost: in this model `TASK_OVERHEAD` is added to the makespan only **once**, whereas forty-four partitions mean handing out work forty-four times. #### The Error Laziness Defers That the plan does not run is a delay as much as a gain. When the driver writes a chain of ten transformations, no row is read, so **whether the chain is correct is not tested either**. The plan below tries to divide a number by a text, and nothing happens at the place where it is defined. ```python reset() broken = Lazy(PARTITIONS).map_rows(lambda k: dict(k, ratio=k["m3"] / k["unit"])) print("plan built, rows read:", COUNTER["read"]) try: broken.first(1) except TypeError as e: print("when the action is called:", type(e).__name__, "-", e) print("rows read until the error:", COUNTER["read"]) ``` ``` plan built, rows read: 0 when the action is called: TypeError - unsupported operand type(s) for /: 'int' and 'str' rows read until the error: 200 ``` While the plan is being built the rows read are **zero**, that is, the error does not show while the plan is written. It shows at the first action, and by then **200 rows** have been read. In the eager setup the same error would have fallen on the line where the `map_rows` call is written. The elimination gain and the error delay come from **the same mechanism**: the postponement of the computation. The same mechanism has a second cost as well: because a plan is not a result, when two actions are called on the same dataset the chain runs twice; the measurement of this cost belongs to the second topic of the course. #### Summary - A distributed dataset consists of **partitions and a plan that has not run yet**; a transformation is written into the plan, only an action starts the computation, and the rows read while the plan is being built are **zero**. - The partition is the unit of data, the worker is the unit of computation: 11 partitions spread over four workers as **[3, 3, 3, 2]**, and an action can open only some of the partitions. - In a result of five rows the eager setup reads **6556 rows by opening 44 partitions**, while the lazy setup reads **200 rows by opening a single partition**; **10 partitions are never opened, 2000 rows are never read**. - `first(100)` raises the rows read from 200 to 600 but the makespan **stays at 250**; in the `count` action, which asks for the whole result, the gain comes from the unused branch and the **3727 intermediate rows** not written (2200 rows against 6556, a makespan of 650 against 1838). - The smallest unit laziness can skip is a partition: the same `first(5)` reads **550** rows in a four-partition dataset and **50** rows in a forty-four-partition one; on the other hand, when the number of partitions is not divisible by the number of workers, the full-scan makespan rises from 600 to **650**. - The type error in the plan appears only at the first action and **after 200 rows have been read**; the elimination gain and the error delay come from **the same mechanism**. #### Next Step This lesson decided which computation will run, but it never touched **who** will run that computation. The partitions were handed to the workers in turn, the number of workers was taken as four, and `TASK_OVERHEAD` was quietly added in the makespan calculation. In a real cluster none of these three is in the hands of the driver: how many workers to split the job into, how many of those workers will run at the same time, and when a waiting job takes its turn are decided by the **resource manager**. The next lesson measures that decision and separates two knobs: increasing the number of tasks and increasing the number of concurrent tasks are not the same thing, because every task has a **fixed cost of 50 units** and this cost grows together with the number of tasks. The number to be measured is the task count at which the makespan gain loses to this cost. ### Resource Managers Source: https://academia.sh/en/courses/distributed-data-processing/resource-managers Counting that the number of tasks and the number of concurrent tasks are two separate knobs in scheduling a job on a cluster, and where the fixed cost of fifty units per task goes: with one task in one slot the makespan is 2250 and the fixed cost 50, while sixteen tasks in sixteen slots bring the makespan down to 215 but raise the fixed cost to 800; increasing concurrency does not change the fixed cost at all, whereas increasing the number of tasks raises it by fifty units per task; splitting the job into more tasks at the same concurrency raises the makespan from 611 to 950; going from eight tasks to sixteen the makespan gain stays at 121 and loses to the 400 units of added fixed cost; and when two jobs run on the same cluster the priority and the shared policy change the order of finishing but do not change the 600 units of fixed cost paid. The previous lesson decided which computation would run and counted what would not, but it never touched **who** would run that computation. The partitions were handed to the workers in turn, the number of workers was taken as four, and `TASK_OVERHEAD` was quietly added to the makespan once. In a real cluster none of these three is in the hands of the driver. This lesson measures the component that stands in between: the **resource manager**. The driver submits a job, and the resource manager decides how many tasks to split the job into, how many tasks will run at the same time, and when a waiting job takes its turn. The number measured is this: when every task has a **fixed cost of 50 units**, how far do increasing the number of tasks and the concurrency bring the makespan down, and **where does the gain lose to this cost.** - **DC49.** The cluster, the job and the resource manager are **fictional**; no real resource manager, cluster manager or cloud service is called. The seed is **20260218**, and the measurement does not depend on the run. - **DC50.** **Makespan is the model's own unit**, not raw milliseconds: it is the work finished by the busiest slot. - **DC51.** **`TASK_OVERHEAD = 50` is a choice**, and the finding of this lesson rests directly on it; every result is written together with its value. The cost is paid **per task**: if a worker is given two tasks, it is paid twice. - **DC52.** A **task** is the share given to a worker. A **slot** is a task place that can run at the same time in the cluster; when a slot frees up it takes the next task. The **number of concurrent tasks** is the number of slots. - **DC53.** The partition key is `subscriber` and there is **no shuffle** in this lesson; the row work is **2200** in every setup. The case in which a wrong key makes adding workers meaningless belongs to the second topic of this course. - **DC54.** The resource manager puts the large task first and places every task in the **emptiest slot**. - **DC55.** The units counted are: tasks, concurrent tasks, makespan, row work, the **total fixed cost** and the **total paid** (row work plus fixed cost). The baseline is **one task in one slot**. - **DC56.** The resource request, the limit and the queueing procedure were measured in the Container Orchestration course, and autoscaling in the Compute and Networking in the Cloud course, and they are **not repeated**. The question here is only what the number of tasks and the number of slots do to the makespan and to the fixed cost. - **DC57.** Resolution: rows **2200**, tasks at most **32**, slots at most **16**. The makespan and the fixed cost are whole numbers, and no ratio is printed. #### What the Resource Manager Decides What the driver submits is a computation plan; it does not say on how many machines it will run. That decision is made by the resource manager, and the decision has three parts. The first is the **number of tasks**: into how many shares the job will be split. The second is the **number of slots**: how many of these shares will run at the same time. The third is the **order**: when this job will start if the cluster is full of other jobs. All three decisions share one cost. Giving work to a task is not free: the plan is sent, a slot is set aside, and an executor is prepared on the worker side. In this model that cost is called `TASK_OVERHEAD` and its value is **50**. The value is a choice, but the moment it is chosen it has a consequence: **increasing the number of tasks raises the fixed cost by 50 per task**, while increasing the number of slots does not change it at all. The two knobs are not the same knob. ```python # FICTIONAL. The source, the first three stages of the pipeline, the cluster and the resource manager are # defined in the lesson; no real resource manager or cluster manager is called. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records def s1_screen(records): REQUIRED = ("subscriber", "period", "region", "index", "unit") keep = [k for k in records if all(a in k for a in REQUIRED)] return keep, len(records) - len(keep) def s2_dedupe(records): seen, keep = set(), [] for k in records: key = (k["subscriber"], k["period"]) if key in seen: continue seen.add(key) keep.append(k) return keep, len(records) - len(keep) def s3_consumption(records): table = {(k["subscriber"], k["period"]): k for k in records} out, dropped = [], 0 for k in records: i = PERIODS.index(k["period"]) if i == 0: continue prev = table.get((k["subscriber"], PERIODS[i - 1])) if prev is None: dropped += 1 continue try: f = int(k["index"]) - int(prev["index"]) except (TypeError, ValueError): dropped += 1 continue out.append(dict(k, m3=f)) return out, dropped WORKERS = 4 NETWORK_COST = 3 # the cost of moving one row over the network , as a multiple of local work TASK_OVERHEAD = 50 # the fixed cost of handing work to a task , a choice def rows(): v, _ = s1_screen(generate_source()) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return v def trace_id(m): h = 2166136261 for c in str(m): h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF return h def distribute(records, key, workers=WORKERS): buckets = {i: [] for i in range(workers)} for k in records: buckets[trace_id(k[key]) % workers].append(k) return buckets def cost(buckets, moved=0, workers=WORKERS): """Makespan = the busiest worker's work (parallel). Total work is counted separately.""" n = [len(v) for v in buckets.values()] return {"duration": max(n) + moved * NETWORK_COST // workers + TASK_OVERHEAD, "total_work": sum(n) + moved * NETWORK_COST, "moved": moved} ``` #### Task, Slot and Fixed Cost The scheduler carries the core of a real resource manager: the tasks are put in order from large to small, every task is placed in the emptiest slot, and every placement pays its own fixed cost. The makespan is the work finished by the busiest slot. ```python S = rows() def shares(records, tasks): """The job is split into 'tasks' shares; the same key always goes to the same share.""" return sorted((len(v) for v in distribute(records, "subscriber", tasks).values()), reverse=True) def schedule(shares, concurrent, fixed=TASK_OVERHEAD): """Resource manager: at most 'concurrent' tasks run at the same time. A slot that frees up takes the next task , and every task pays its own fixed cost.""" slots = [0] * concurrent for share in shares: i = slots.index(min(slots)) slots[i] += share + fixed return {"tasks": len(shares), "duration": max(slots), "row_work": sum(shares), "fixed": len(shares) * fixed, "paid": sum(shares) + len(shares) * fixed} print("cost (4 workers , no shuffle):", cost(distribute(S, "subscriber", 4))) print("schedule (4 tasks , 4 slots) :", schedule(shares(S, 4), 4)) print("baseline (1 task , 1 slot) :", schedule(shares(S, 1), 1)) ``` ``` cost (4 workers , no shuffle): {'duration': 611, 'total_work': 2200, 'moved': 0} schedule (4 tasks , 4 slots) : {'tasks': 4, 'duration': 611, 'row_work': 2200, 'fixed': 200, 'paid': 2400} baseline (1 task , 1 slot) : {'tasks': 1, 'duration': 2250, 'row_work': 2200, 'fixed': 50, 'paid': 2250} ``` The first two rows give the same makespan: **611**. This is a check: the course's `cost` function and the scheduler of this lesson produce the same result for four tasks in four slots. The difference is in the third number. `cost` adds the fixed cost to the makespan **once**, since exactly one task falls on each worker; `schedule` counts it as a separate item and writes the total fixed cost of four tasks as **200**. The baseline is the last row: one task, one slot, makespan **2250**, fixed cost **50**. #### The Two Knobs Are Not the Same Knob The matrix below puts the number of tasks in the rows and the number of slots in the columns. Every cell is the makespan; the left column is the total fixed cost paid by each row and is **independent of the number of slots**. ```python CONC = (1, 2, 4, 8, 16) print(f"{'tasks':>6}{'fixed cost':>12}", "".join(f"{'conc=' + str(e):>9}" for e in CONC)) for g in (4, 8, 16, 32): p = shares(S, g) print(f"{g:>6}{g * TASK_OVERHEAD:>12}", "".join(f"{schedule(p, e)['duration']:>9}" for e in CONC)) ``` ``` tasks fixed cost conc=1 conc=2 conc=4 conc=8 conc=16 4 200 2400 1200 611 611 611 8 400 2600 1300 650 336 336 16 800 3000 1500 750 386 215 32 1600 3800 1900 950 475 243 ``` Reading along a row shows the slot knob: the job of sixteen tasks finishes with a makespan of **3000** in one slot and **215** in sixteen slots, and the fixed cost paid is **800** at both ends. Increasing the concurrency does not touch the fixed cost; it only shortens the time in which it is paid. Reading along a column shows the second knob, and it works in the opposite direction. In four slots, if the job is split into four tasks the makespan is **611**, into eight **650**, into sixteen **750**, into thirty-two **950**. The job did not change; the row work is still 2200. The only thing that rises is the fixed cost: from 200 to 1600. **Every task beyond the number of slots pays a 50 and buys no parallelism**, because there is no free place to run. The two knobs are tied to each other, and the most expensive corner of the table says so. To be able to use sixteen slots, the job must be split into at least sixteen tasks; that is, the precondition of dropping to a makespan of **215** is a fixed cost of **800** units. Splitting into thirty-two tasks raises the makespan in the same sixteen slots from 215 to **243** and the fixed cost to **1600**. Saturation is in the same place too: the job of four tasks finishes with a makespan of **611** in eight or sixteen slots as well, and the extra slots sit idle. #### Where the Makespan Gain Loses The diagonal of the matrix represents the most frequently made decision: one slot for every task, that is, choosing how many workers to split the job into. To see where the gain loses, two columns are put side by side: the makespan gained and the fixed cost added. ```python print(f"{'tasks=conc':>12}{'skew':>6}{'makespan':>10}{'makespan gain':>15}" f"{'added fixed':>13}{'paid':>6}") previous = None for g in (1, 2, 4, 8, 16): p, z = shares(S, g), schedule(shares(S, g), g) c = round(max(p) / (sum(p) / len(p)), 4) gain = "-" if previous is None else str(previous[0] - z["duration"]) added = "-" if previous is None else str(z["fixed"] - previous[1]) print(f"{g:>12}{c:>6}{z['duration']:>10}{gain:>15}{added:>13}{z['paid']:>6}") previous = (z["duration"], z["fixed"]) ``` ``` tasks=conc skew makespan makespan gain added fixed paid 1 1.0 2250 - - 2250 2 1.0 1150 1100 50 2300 4 1.02 611 539 100 2400 8 1.04 336 275 200 2600 16 1.2 215 121 400 3000 ``` The first three steps are beyond dispute: going from one task to two lowers the makespan by **1100** units and adds **50** units of fixed cost. From two to four the gain is 539, the cost 100. From four to eight the gain is 275, the cost 200. While the gain is halved at every step, the cost is doubled at every step; that is, the two numbers are running toward each other. They meet in the step from eight to sixteen, and the order reverses: the makespan gain is **121**, the added fixed cost **400**. **The place where the makespan gain loses is between eight and sixteen tasks.** This does not mean the makespan does not fall; it really drops from 336 to 215. What it means is this: while the total paid rises from 2600 to 3000, the row work does not change at all, that is, all of the added 400 units is not work but **the cost of handing out work**. Someone who looks at this row without being told which one was measured sees a pure speedup. The skew column carries a second warning. At sixteen tasks even the `subscriber` key produces a skew of **1.2**: the heaviest share is 165 rows, the lightest 110. As the share gets smaller its equality deteriorates and the fixed cost grows; the two effects work in the same direction. #### Two Jobs on the Same Cluster The real work of the resource manager is not to speed up a single job but to **share the cluster**. Two jobs arrive at a cluster of eight slots: A is a job of eight tasks that processes all the rows, B is a job of four tasks that processes only the residential subscribers. In the priority policy A takes the whole cluster and B waits for its turn; in the shared policy the cluster is divided in two. ```python RESIDENTIAL = [k for k in S if k["type"] == "residential"] A, B = shares(S, 8), shares(RESIDENTIAL, 4) print("job A rows:", sum(A), " tasks:", len(A), " job B rows:", sum(B), " tasks:", len(B)) for policy in ("priority", "shared"): if policy == "priority": # first A takes the whole cluster a = schedule(A, 8)["duration"] b = a + schedule(B, 8)["duration"] else: # the cluster is divided in two a, b = schedule(A, 4)["duration"], schedule(B, 4)["duration"] print(f"{policy:<12} A finishes {a:>5} B finishes {b:>5} both {max(a, b):>5}" f" fixed cost {(len(A) + len(B)) * TASK_OVERHEAD:>5}") ``` ``` job A rows: 2200 tasks: 8 job B rows: 1078 tasks: 4 priority A finishes 336 B finishes 683 both 683 fixed cost 600 shared A finishes 650 B finishes 347 both 650 fixed cost 600 ``` The policy changes the order of finishing from top to bottom. In the priority setup A finishes at **336** and B waits until **683**; in the shared setup A is delayed to **650** while B finishes at **347**. The moment when both are finished drops from 683 to 650; that is, the shared policy keeps the cluster a little fuller. There is no superior policy; what is chosen is which job is delayed. The last column is the summing sentence of this lesson: **the fixed cost is 600 in both policies.** The scheduling policy determines who finishes when, and it **does not determine the fixed cost paid**. That cost is a function of the number of tasks alone, and twelve tasks pay 50 twelve times. What can be gained by changing the resource manager is the order, not the work. #### Summary - The resource manager decides three things: the **number of tasks**, the **number of slots** and the **order**; the first two are separate knobs and they behave differently toward the fixed cost. - Increasing the number of slots **does not change** the fixed cost at all: the job of sixteen tasks finishes with a makespan of 3000 in one slot and 215 in sixteen slots, and the fixed cost paid is **800** at both ends. - Increasing the number of tasks raises the fixed cost by **50** per task: in four slots the makespan rises from 611 to **950** while the row work stays at 2200, because a task beyond the number of slots buys no parallelism. - When every task is given a slot, the makespan gain is halved at every step and the added fixed cost is doubled at every step; **between eight and sixteen tasks the gain stays at 121 and loses to the 400 added units**, and the total paid rises from 2600 to 3000. - At sixteen tasks even the `subscriber` key produces a skew of **1.2**; the inequality of the shares and the growth of the fixed cost work in the same direction. - When two jobs run on the same cluster, the policy changes the order of finishing (A 650 and B 347 instead of A 336 and B 683) but **does not change the fixed cost paid**: **600** in both setups. #### Next Step In five lessons the job was split. A cluster was built and separated by the names driver, executor and task; the blocks were placed and which worker the read fell on was counted; the map was separated from the reduce; which part of the computation would never run was measured; and who decides into how many tasks the job is split was written down. In all these lessons one number stayed the same and was never questioned: **the row work is 2200**. We split the job, distributed it, scheduled it, but never touched the job itself. Yet the slowness of a distributed job most often does not come from the number of rows. Moving from one partition key to another carries the rows over the network, and in this model that movement is counted at **three times** the cost of a local operation; there are two ways to join two relations, and the cheaper one changes with size; when a key does not divide the shares equally, some workers sit idle. The next topic counts these three separately, starting with this question: in the same job, with the same number of workers, only because the partition key changed, how many rows pass through the network and to where does the total work rise from 2200. ## Performance Counting where a distributed job slows down and when attempts to speed it up make it worse: shuffling in the same layout moving 1738 rows and raising the makespan from 611 to 1914 and the total work from 2200 to 7414, the broadcast join winning up to a size of 1600 rows while the fact side is partitioned by region and the crossover point lying between 1600 and 4000, the skew of 1.02 on the subscriber key rising to 3.32 on the region key with two of four workers left idle so that adding workers becomes meaningless, the total of two queries falling from 5500 to 3300 when the intermediate result is cached, lowering the checkpoint interval from six to three leaving the rows recomputed at the same failure point at 400, the makespan of 2250 on a single worker rising to 3691 with a second worker, and the total work rising from 5830 to 8635 while the makespan falls, which makes it mandatory to write which number was measured. ### Shuffle Cost Source: https://academia.sh/en/courses/distributed-data-processing/shuffle-cost Counting for the first time where a distributed job slows down: when the same 2200 rows laid out by the subscriber key are repartitioned by the subscriber key, the shuffled rows stay at 0, the makespan is 611 and the total work 2200; when the same rows are shuffled to the region key, 1738 rows cross the network, the makespan rises to 1914 and the total work to 7414, and all three setups produce exactly the same result. When the network cost is swept over 1, 3, 6 and 12, the shuffled rows do not change at all and the 611 and 2200 of the setup without a shuffle stay fixed, while the makespan of the setup with a shuffle moves from 1045 to 5825 and its total work from 3938 to 23056; the ordering of the three keys comes out the same at all four values, so the ordering is insensitive to the choice and the magnitude is entirely of the choice. The cost of a shuffle comes not from the computation itself but from moving the data from its place. Over five lessons the work was split. The driver divided the job into tasks, the workers received their partitions, the resource manager put the tasks in a queue, and the distributed setup produced exactly the same result as the pipeline on a single machine. It was measured that the partitioning **works**; **where it slows down** was never asked. This lesson asks that question and looks for the answer in a single place: what happens when a row passes from one worker to another. In a distributed job the computation itself is the same everywhere: summing is summing, comparing is comparing. The only thing that differs is **where the data sits**, and moving data from its place is more expensive than processing it in place. What this lesson measures is that difference. - **PF1.** The source, the consumption rows and the cluster are **fictional** and are generated within the lesson. No distributed processing framework, resource manager or distributed file system is called. The seed is 20260218. - **PF2.** **Makespan** is the model's **own unit**, not raw milliseconds. Makespan is **the work of the busiest worker**, because the workers run in parallel and the job ends when the last one ends. **Total work** is the sum of the work of all the workers. The two are measured separately and can move in different directions. - **PF3.** **Shuffle** is repartitioning the rows by a new key. A row that changes workers crosses the network and is counted as a **shuffled row**; a row that keeps its worker never touches the network. - **PF4.** The cost of a row crossing the network is `NETWORK_COST` times that of a local operation, its default value is **3**, and **this is a choice**. This lesson sweeps it over 1, 3, 6 and 12. - **PF5.** The fixed cost of handing work to each worker is `TASK_OVERHEAD` = **50**, and it is also a choice. When two setups with the same number of workers are compared, this constant is added to both equally, so it does not enter the difference. - **PF6.** Partitioning is done by the hash of the key: **the same key value always goes to the same worker**. Reduce requires this; if the rows of a key are not collected on a single worker, the total of that key cannot be computed. - **PF7.** A **partition** here is the set of rows given to a worker. The partitioning procedure was measured in the Relational Database Administration course of the Databases curriculum and in the Data Storage Architectures course of this curriculum; **it is not repeated**. **Sharding** is a separate concept and is **not confused** with partitioning. - **PF8.** The baseline is **a single worker**: no shuffle, makespan **2250**, total work **2200**. - **PF9.** All the setups in this lesson produce **exactly the same result**; the only thing measured is cost. #### The Two Numbers of a Shuffle The block below splits the 2200 consumption rows coming out of the pipeline across four workers by the `subscriber` key, then repartitions the same rows by three different keys and counts, each time, how many rows change workers. ```python # FICTIONAL. Meter readings of the measurement network, the five-stage pipeline and the cluster # are generated within the lesson; no distributed processing framework, resource manager or # file system is called. Seed 20260218. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records def rows(): """Consumption rows coming out of the pipeline. On a sound source the screen and dedupe stages drop no rows; consumption is found by subtracting the previous period's index.""" raw, output = generate_source(), [] table = {(k["subscriber"], k["period"]): k for k in raw} for k in raw: i = PERIODS.index(k["period"]) prev = table.get((k["subscriber"], PERIODS[i - 1])) if i else None if prev is not None: output.append(dict(k, m3=k["index"] - prev["index"])) return output WORKERS = 4 NETWORK_COST = 3 # the cost of moving one row over the network , as a multiple of local work TASK_OVERHEAD = 50 # the fixed cost of handing work to each worker def trace_id(m): h = 2166136261 for c in str(m): h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF return h def distribute(records, key, workers=WORKERS): """Assigns rows to workers by key. The same key always goes to the same worker.""" buckets = {i: [] for i in range(workers)} for k in records: buckets[trace_id(k[key]) % workers].append(k) return buckets def shuffle(buckets, new_key, workers=WORKERS): """Repartition. A row that changes workers crosses the network.""" new = {i: [] for i in range(workers)} moved = 0 for old, records in buckets.items(): for k in records: target = trace_id(k[new_key]) % workers if target != old: moved += 1 new[target].append(k) return new, moved def cost(buckets, moved=0, workers=WORKERS): """Makespan = the busiest worker's work (parallel). Total work is counted separately.""" n = [len(v) for v in buckets.values()] return {"duration": max(n) + moved * NETWORK_COST // workers + TASK_OVERHEAD, "total_work": sum(n) + moved * NETWORK_COST, "moved": moved} ALL = rows() SINGLE = cost(distribute(ALL, "subscriber", 1), 0, 1) BUCKETS = distribute(ALL, "subscriber") print(f"rows from the pipeline {len(ALL)}, workers {WORKERS}, network cost {NETWORK_COST}," f" task overhead {TASK_OVERHEAD}") print(f"single-worker baseline: makespan {SINGLE['duration']}, total work {SINGLE['total_work']}, shuffled 0") print(f"starting layout subscriber, worker load {[len(v) for v in BUCKETS.values()]}") print("\nnew key shuffled in place makespan total work new worker load") for target in ("subscriber", "period", "region"): new, t = shuffle(BUCKETS, target) m = cost(BUCKETS, t) print(f"{target:<12} {t:>9} {len(ALL) - t:>8} {m['duration']:>8} {m['total_work']:>10}" f" {[len(v) for v in new.values()]}") ``` ``` rows from the pipeline 2200, workers 4, network cost 3, task overhead 50 single-worker baseline: makespan 2250, total work 2200, shuffled 0 starting layout subscriber, worker load [539, 550, 561, 550] new key shuffled in place makespan total work new worker load subscriber 0 2200 611 2200 [539, 550, 561, 550] period 1651 549 1849 7153 [600, 600, 400, 600] region 1738 462 1914 7414 [374, 0, 0, 1826] ``` The first row draws a limit: a job distributed without a shuffle is **possible**. When the rows laid out by `subscriber` are repartitioned by the `subscriber` key again, **no row changes workers**. Shuffled rows 0, makespan 611, total work 2200. The total work is exactly the same as the single-worker baseline: four workers shared the same 2200 units of work, nothing missing and nothing extra. The makespan has fallen from 2250 to 611, and the whole of the fall comes from sharing the work. The third row shows the entirely different cost that the same data, at the same number of workers, pays to produce the same result. Moving to the `region` key sends **1738** of the 2200 rows across the network; only 462 rows stay in place. The makespan rises from 611 to **1914**, the total work from 2200 to **7414**. The computation has not changed: the same totals, the same regions, the same output. The only thing that changed is **where the rows sit**, and that movement has written a line item larger than the job itself. 5214 units of the total work, more than two thirds of it, are pure movement cost in which no computation is done. What the makespan consists of must be written out plainly here. The 1914 in the `region` row is the sum of three items: the 561 rows of the busiest worker, the 1303 units that come from multiplying the 1738 shuffled rows by the network cost and dividing by the number of workers, and the task overhead of 50 units. Dividing the transfer cost by the number of workers is a modeling choice and assumes that the network traffic is **shared equally** among the workers. In a real cluster a single worker may take on most of the transfer, and the makespan comes out higher than this. The model therefore shows the cost of a shuffle as **less** than it is; the difference here is a lower bound. The middle row shows that the number of shuffled rows changes from key to key: moving to the `period` key shuffles 1651 rows, moving to `region` 1738. The difference is small but its reason is clear. A row stays in place when its old key and its new key **land on the same worker**. With four workers the expected share for this is one in four, that is, 550 rows. The 549 rows that stay in place under `period` give exactly this share; the 462 rows that stay under `region` are fewer than that, because the regions are not spread equally across the workers. A row that the shuffle does not move is not a rescued row, it is a row that **happened to land in the same place**. The last column opens the second face of the shuffle and sets up the subject of the next two lessons. Under the `subscriber` key the load of the four workers is between 539 and 561; under `period` between 400 and 600; under `region` between **374 and 1826**, and **two workers are completely idle**. A shuffle does not only move data, it also **produces a new imbalance**; because the makespan depends on the busiest worker, this imbalance goes directly onto the bill. The unwritten column of the table is correctness. The region totals that the three setups produce are **exactly the same**; whichever key the partitioning is made by, every row enters the computation exactly once. This is the third claim of this course: **the result is the same in every setup, only the cost changes.** A test that checks the output passes all three setups and says **nothing** about the 5214-unit movement cost of the `region` partitioning. A wrong partitioning is not an error but a price, and a price becomes visible only if it is counted. #### Network Cost Is a Choice `NETWORK_COST = 3` sits inside all of the numbers above. This number is not a measurement; it is a stand-in given in order to reduce to a single unit how many times more it costs to send a row across the network than to process a local row. In a real cluster this ratio varies with the speed of the network, the size of the row and the cost of serialization. So nothing counts as measured until it is written how much the result depends on it. The block below repeats the same three shuffles with four different network costs. ```python # This block continues the definitions of rows(), trace_id(), distribute(), shuffle(), ALL, # BUCKETS, WORKERS and TASK_OVERHEAD from the first block. The only thing that changes is NETWORK_COST. def cost_with_network(buckets, moved, network, workers=WORKERS): """The same computation as cost() , only the network cost is passed in from outside.""" n = [len(v) for v in buckets.values()] return {"duration": max(n) + moved * network // workers + TASK_OVERHEAD, "total_work": sum(n) + moved * network} TARGETS = [(h, shuffle(BUCKETS, h)[1]) for h in ("subscriber", "period", "region")] print("independent of the network cost: shuffled rows " + ", ".join(f"{h} {t}" for h, t in TARGETS)) print("\nnetwork cost" + "".join(f"{h + ' makespan/total work':>32}" for h, _ in TARGETS) + " region/subscriber makespan") for network in (1, 3, 6, 12): m = [cost_with_network(BUCKETS, t, network) for _, t in TARGETS] print(f"{network:>12}" + "".join(f"{str(x['duration']) + '/' + str(x['total_work']):>32}" for x in m) + f"{round(m[2]['duration'] / m[0]['duration'], 2):>29}") ``` ``` independent of the network cost: shuffled rows subscriber 0, period 1651, region 1738 network cost subscriber makespan/total work period makespan/total work region makespan/total work region/subscriber makespan 1 611/2200 1023/3851 1045/3938 1.71 3 611/2200 1849/7153 1914/7414 3.13 6 611/2200 3087/12106 3218/12628 5.27 12 611/2200 5564/22012 5825/23056 9.53 ``` The first line of the output, the one before the header, is the precondition of the sweep: **the shuffled rows are independent of the network cost.** 0, 1651 and 1738 are the same at all four values, because whether a row changes workers depends on the hash of the key, not on the speed of the network. The network cost determines only **how many units that number writes**. Measuring and pricing the measurement are separated in this table. The second insensitivity is in the `subscriber` column: **611 and 2200 do not change in any of the four rows.** A setup that shuffles no rows never pays the network cost, so it is completely indifferent to that choice. This gives a decision rule: if the cost of the network is uncertain, **the plan that does not touch the uncertainty** is the safe one. The value of the setup without a shuffle lies less in its being fast than in its estimate **not depending on the choice**. The third column gives the sensitive one. The makespan of the `region` shuffle moves from 1045 to **5825**, and its total work from 3938 to **23056**. The last column reduces this to a single number: the setup with a shuffle is **1.71** times as expensive as the setup without one at a network cost of 1, **3.13** times at 3, **5.27** times at 6 and **9.53** times at 12. Doubling the network cost does not double the difference, because the makespan contains a floor that is independent of the cost (the 561 rows of the busiest worker and the task overhead of 50 units); only the transfer share grows. The last thing the sweep **does not change** is the ordering. At all four network costs `subscriber` is the cheapest, `period` is in the middle and `region` is the most expensive. The reason is that the makespan rises monotonically with the shuffled rows: as long as `NETWORK_COST` is positive, the plan that moves more is more expensive and the ordering does not change. So the question "which key shuffles fewer rows" is **insensitive** to the choice, while "how expensive will it be" is **entirely sensitive** to it. The next lesson shows a decision in which this distinction changes the outcome: there one of two plans pays shuffled rows and the other does not, and which one wins depends directly on the value of `NETWORK_COST`. #### Summary - A shuffle is repartitioning rows by a new key; its measure is the **shuffled rows**. Going from the `subscriber` layout to `subscriber` shuffles 0 rows, going to `period` 1651, going to `region` 1738 rows. - For the same 2200 rows, the same four workers and the same result, the makespan rises from **611** to **1914** and the total work from **2200** to **7414**. 5214 units of the total work are pure movement cost in which no computation is done; the computation is the same in all three setups. - A distributed job without a shuffle is possible, and its total work stays exactly the same as the single-worker baseline (2200); the makespan falls from 2250 to 611 through sharing alone. - A shuffle produces a second cost as well: the new worker load is 539–561 under `subscriber` and 374–1826 under `region`, and two workers are left idle. Because the makespan depends on the busiest worker, this imbalance goes directly onto the bill. - When the network cost is swept over 1, 3, 6 and 12, the shuffled rows and the 611/2200 of the setup without a shuffle never change; the setup with a shuffle moves from 1045/3938 to 5825/23056 and the difference rises from 1.71 times to 9.53 times. The ordering is the same at all four values: the magnitude is of the choice, the ordering is not. #### Next Step All the measurements in this lesson were made on **a single data set**: 2200 rows were partitioned, repartitioned, and the shuffled rows were counted. Yet most distributed jobs bring together not one data set but **two**: a large fact side and a small dimension side. Both sides must be partitioned by the same key, otherwise the rows that are to match sit on different workers. There are two separate ways to ensure this and their costs are of entirely different kinds: **copying** the small side **to every worker**, or **shuffling both sides**. The next lesson compares these two strategies as the dimension side grows, and looks, by counting, for the **crossover point** at which the winner changes. ### Joins at Scale Source: https://academia.sh/en/courses/distributed-data-processing/joins-at-scale Measuring a job that brings two sets face to face with two separate strategies and counting the point where the winner changes: while the fact side is partitioned by the region key, the broadcast join copies the dimension side to every worker and writes a makespan of 1926 at a dimension of 50 rows, 2276 at 400 and 3476 at 1600; the partitioned join, however, has to move 1738 rows of the fact side no matter how small the dimension gets, so it cannot drop below 3179. Because the makespan of the broadcast join grows by one unit per dimension row and that of the partitioned join by nine sixteenths of a unit, the two lines cross: the makespans are equal at 2978 and 2979, the partitioned join pulls ahead at 2980, and at 4000 the gap is 5876 against 5429. The same crossover point comes out as 2980 when total work is measured. The decision is sensitive to the network cost: at a cost of 1 the crossover point drops to 535, and at 6 and 12 the partitioned join cannot pull ahead at any size. Both strategies match the same rows at every size. All the measurements of the previous lesson were made on **a single data set**: 2200 rows were partitioned, repartitioned, and the shuffled rows were counted. Most distributed jobs, however, bring together not one set but **two**: a large fact side and a dimension side attached to it. For a join to work, the two rows that are to match must sit on **the same worker**, otherwise they cannot meet. There are two ways to ensure this, and their costs are of different kinds. The small side can be **copied to every worker**; then no row is shuffled, but the small side is multiplied by the number of workers. Or **both sides** are shuffled by the join key; then there are no copies, but the rows of both sides cross the network. This lesson compares the two strategies as the dimension side grows and looks for the point where the winner changes. - **PF10.** The source, the consumption rows, the dimension side and the cluster are **fictional** and are generated within the lesson. No distributed processing framework, query engine or file system is called. The seed is 20260218. - **PF11.** The fact side is the **2200** consumption rows coming out of the pipeline, and it sits partitioned by the `region` key, in the most unbalanced layout of the previous lesson. The join key is `subscriber`. The distinction between the **fact and dimension relations** was established in the Relational Database Administration course of the Databases curriculum; it is **not repeated**. - **PF12.** The dimension side is a **subscriber index**: keys starting from `A000`. Its size is swept; an index of 50 rows represents a subset of the subscribers, and an index of 4000 rows a register far wider than the pipeline sees. - **PF13.** **Broadcast join**: the dimension side is **copied to every worker**. Shuffled rows 0, copies `dimension x workers`. The cost of a copy is **one unit** and is not multiplied by the network cost; this is a model choice and it directly determines the result of the lesson. - **PF14.** **Partitioned join**: both sides are **shuffled** by the join key. Copies 0, and each shuffled row writes a cost of `NETWORK_COST` times. - **PF15.** The dimension side enters the job **on a single worker, unpartitioned**. In the partitioned strategy, therefore, its rows that change workers are added to the shuffled rows. - **PF16.** Makespan is the work of the busiest worker, **total work** is counted separately, and makespan is the model's **own unit**. `TASK_OVERHEAD` = 50 is added equally to both strategies at the same number of workers, so it does not enter the comparison. - **PF17.** The baseline is **a single worker**: makespan **2250**, total work **2200**. - **PF18.** The two strategies **match the same rows** at every dimension size; the only thing measured is cost. #### Two Strategies, Two Kinds of Cost The block below partitions the fact side by `region`, generates the dimension side at six different sizes and writes the cost of the two strategies side by side at each size. The matched rows column is the same for both strategies and is printed once. ```python # FICTIONAL. Meter readings, consumption rows and the cluster are generated within the lesson; # no distributed processing framework or file system is called. Seed 20260218. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] WORKERS, NETWORK_COST, TASK_OVERHEAD = 4, 3, 50 SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def rows(): """Consumption rows coming out of the pipeline (the fact side). On a sound source the screen and dedupe stages drop no rows; consumption is found from the period difference.""" raw = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) raw.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) table, output = {(k["subscriber"], k["period"]): k for k in raw}, [] for k in raw: i = PERIODS.index(k["period"]) prev = table.get((k["subscriber"], PERIODS[i - 1])) if i else None if prev is not None: output.append(dict(k, m3=k["index"] - prev["index"])) return output def trace_id(m): h = 2166136261 for c in str(m): h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF return h def distribute(records, key, workers=WORKERS): """Assigns rows to workers by key. The same key always goes to the same worker.""" buckets = {i: [] for i in range(workers)} for k in records: buckets[trace_id(k[key]) % workers].append(k) return buckets def shuffle(buckets, new_key, workers=WORKERS): """Repartition. A row that changes workers crosses the network.""" new, moved = {i: [] for i in range(workers)}, 0 for old, records in buckets.items(): for k in records: target = trace_id(k[new_key]) % workers if target != old: moved += 1 new[target].append(k) return new, moved def broadcast_join(buckets, dimension, workers=WORKERS): """The dimension is copied to every worker. No shuffle , but there are copies.""" copies = len(dimension) * workers n = [len(v) for v in buckets.values()] return {"moved": 0, "copies": copies, "duration": max(n) + copies // workers + TASK_OVERHEAD, "total_work": sum(n) + copies} def partitioned_join(buckets, dimension, key="subscriber", workers=WORKERS): """Both sides are shuffled by key.""" _, moved = shuffle(buckets, key, workers) dimension_moved = sum(1 for a in dimension if trace_id(a) % workers != 0) n = [len(v) for v in buckets.values()] t = moved + dimension_moved return {"moved": t, "copies": 0, "duration": max(n) + t * NETWORK_COST // workers + TASK_OVERHEAD, "total_work": sum(n) + t * NETWORK_COST} def generate_dimension(n): """Subscriber index: n keys. Sits on a single worker , unpartitioned.""" return [f"A{i:03d}" for i in range(n)] ALL = rows() BUCKETS = distribute(ALL, "region") print(f"fact side {len(ALL)} rows , partitioned by the region key ," f" worker load {[len(v) for v in BUCKETS.values()]}") print(f"single-worker baseline: makespan {len(ALL) + TASK_OVERHEAD} , total work {len(ALL)}") print(f"join key subscriber , workers {WORKERS} , network cost {NETWORK_COST}") print("\n size matched copies bc.dur bc.total moved pt.dur pt.total winner") for n in (50, 200, 400, 800, 1600, 4000): d = generate_dimension(n) b, p = broadcast_join(BUCKETS, d), partitioned_join(BUCKETS, d) m = sum(1 for k in ALL if k["subscriber"] in set(d)) print(f"{n:>5} {m:>8} {b['copies']:>6} {b['duration']:>7} {b['total_work']:>9}" f" {p['moved']:>9} {p['duration']:>7} {p['total_work']:>9}" f" {'broadcast' if b['duration'] < p['duration'] else 'partitioned'}") ``` ``` fact side 2200 rows , partitioned by the region key , worker load [374, 0, 0, 1826] single-worker baseline: makespan 2250 , total work 2200 join key subscriber , workers 4 , network cost 3 size matched copies bc.dur bc.total moved pt.dur pt.total winner 50 539 200 1926 2400 1775 3207 7525 broadcast 200 2189 800 2076 3000 1888 3292 7864 broadcast 400 2200 1600 2276 3800 2038 3404 8314 broadcast 800 2200 3200 2676 5400 2338 3629 9214 broadcast 1600 2200 6400 3476 8600 2938 4079 11014 broadcast 4000 2200 16000 5876 18200 4738 5429 16414 partitioned ``` The matched rows column is the precondition of the table. At the same dimension size the two strategies **match the same rows**: 539 in the index of 50 rows, 2189 in the one of 200, and 2200 after 400. The column grows with the size because the index covers more and more subscribers; after 400 rows it is fixed because all the subscribers are covered. What is measured is not correctness, because correctness is the same in both strategies. The first row gives the most important property of the partitioned strategy: **it has a floor.** Even if the dimension is only 50 rows, the partitioned join has to move **1738 rows** of the fact side, because the fact side is partitioned by `region` while the join is made over `subscriber`. The partitioned makespan therefore has a floor: **3179** even if the dimension fell to zero, and **3207** at 50 rows, the smallest size in the table. The broadcast strategy has no floor; as the dimension shrinks its cost approaches zero, and at 50 rows the makespan is only **1926**. The broadcast join winning at a small dimension is not a tendency but a structural result. The growth rate of the two columns decides the matter. In the broadcast strategy each row of the dimension is copied as many times as there are workers, and because the copies are divided among the workers, it adds **one unit per row** to the makespan: 1876 plus the dimension rows. In the partitioned strategy three quarters of the dimension changes workers, each shuffled row is multiplied by the network cost and divided among the workers, and what is added to the makespan is **nine sixteenths of a unit per row**. The broadcast line is steeper but starts from zero; the partitioned line grows more slowly but from a height of 1303 units. The two lines are bound to cross, and the place where they cross is the number this lesson is looking for. The total work column tells the same story in another unit. At a dimension of 1600 rows the broadcast join does 8600 units of work and the partitioned join 11014; at 4000 the order reverses (18200 against 16414). The rule of the course says that makespan and total work can go in different directions; **in this comparison they do not**, because the makespan of each strategy is its total work divided by the number of workers, and the divisor is the same in both. The meaning of the rule here is this: that the two agree **must also be shown**, otherwise it is not known which one was measured. #### The Turning Point Six sizes say that the turn lies between 1600 and 4000, but they do not say where. The block below sweeps the dimension size one by one. To make the sweep fast, the shuffled rows of the dimension side are kept as a prefix sum; the first line verifies that this shortcut gives the same result as the functions in the first block. ```python # This block continues the definitions of trace_id(), distribute(), shuffle(), broadcast_join(), # partitioned_join(), generate_dimension(), ALL and BUCKETS from the first block. PREFIX = [0] # PREFIX[n]: how many of the first n dimension rows change workers for a in generate_dimension(4000): PREFIX.append(PREFIX[-1] + (1 if trace_id(a) % WORKERS else 0)) _, FACT_MOVED = shuffle(BUCKETS, "subscriber") LOAD = [len(v) for v in BUCKETS.values()] def measure(n, network=NETWORK_COST): """The same computation as the two functions in the first block; the only differences are that the network cost is passed in from outside and the dimension side is sped up with a prefix sum. Returns: broadcast makespan , partitioned makespan , broadcast total work , partitioned total work.""" t = FACT_MOVED + PREFIX[n] return (max(LOAD) + n + TASK_OVERHEAD, max(LOAD) + t * network // WORKERS + TASK_OVERHEAD, sum(LOAD) + n * WORKERS, sum(LOAD) + t * network) d = generate_dimension(4000) print(f"check , dimension 4000: with the functions {broadcast_join(BUCKETS, d)['duration']}" f"/{partitioned_join(BUCKETS, d)['duration']} , with the prefix sum" f" {measure(4000)[0]}/{measure(4000)[1]}") print("\nboth sides of the turning point at network cost 3") for n in (2978, 2979, 2980, 2981): b, p, _, _ = measure(n) print(f"dimension {n} broadcast {b} partitioned {p} " f"{'broadcast' if b < p else 'partitioned' if p < b else 'equal'}") first_total = next(n for n in range(1, 4001) if measure(n)[3] < measure(n)[2]) print(f"\nfirst dimension size at which the partitioned join pulls ahead on total work: {first_total}") print("\nnetwork cost first dimension size at which the partitioned join pulls ahead on makespan (up to 4000)") for network in (1, 3, 6, 12): first = next((n for n in range(1, 4001) if measure(n, network)[1] < measure(n, network)[0]), None) print(f"{network:>12} {first if first else 'none'}") ``` ``` check , dimension 4000: with the functions 5876/5429 , with the prefix sum 5876/5429 both sides of the turning point at network cost 3 dimension 2978 broadcast 4854 partitioned 4854 equal dimension 2979 broadcast 4855 partitioned 4855 equal dimension 2980 broadcast 4856 partitioned 4855 partitioned dimension 2981 broadcast 4857 partitioned 4856 partitioned first dimension size at which the partitioned join pulls ahead on total work: 2980 network cost first dimension size at which the partitioned join pulls ahead on makespan (up to 4000) 1 535 3 2980 6 none 12 none ``` The crossover point is **2980**. At 2978 and 2979 the two strategies are **equal**, and at 2980 the partitioned join pulls ahead by one unit. That the crossing is not sharp comes from the integer division of the model: the broadcast makespan grows by a whole unit per row and the partitioned makespan by nine sixteenths of a unit, so the two lines meet at the same integer for a few rows. The practical reading of the decision is not "the strategy changes as the dimension approaches three thousand" but **"around three thousand the difference between the two strategies becomes unmeasurable"**; the only places where the difference is meaningful are the extremes. At 50 rows the broadcast join is ahead by **1281** units, at 4000 rows the partitioned join by **447**. The total work row confirms that the turn is at the same place: **2980**. This is no coincidence. The total work of the broadcast strategy grows by four units per dimension row and its makespan by one; the total work of the partitioned strategy grows by 2.25 units per row and its makespan by one quarter of that. The ratio of the two measures is the number of workers in both, so their orderings turn at the same point. This is not an exception to the rule of the course but **proof of why the rule had to be written**: that they agree is known only when both are measured. The last table shows what the decision hangs on, and it is the harshest finding of this lesson. When the network cost is 1, the crossover point drops to **535**; a decision made for an index of 2980 rows **reverses**, with the same data and the same number of workers, only because one constant changed. When the network cost is 6 or 12, the partitioned join cannot pull ahead at **any** of the swept dimension sizes. The reason can be read off: the partitioned makespan grows by three sixteenths of the network cost per dimension row, the broadcast makespan by one unit; as long as the network cost stays at five, the partitioned join grows more slowly, and when it rises to six it grows faster and can never catch up again. So the sentence "at a large dimension the partitioned join wins" is true for a network cost of 3 and **false** for 6. The rule that follows is this: a join strategy cannot be chosen by looking at the data. Both sides match the same rows, give the same result, and the correctness check passes both. The choice is made by looking at the size of the dimension side, at the key by which the fact side is partitioned, and at the network cost; **a decision given without writing down one of the three inputs counts as unmeasured.** #### Summary - The broadcast join copies the dimension side to every worker (shuffled 0), and the partitioned join shuffles both sides (copies 0). The costs are of different kinds: one is copies, the other shuffled rows. - The partitioned strategy has a floor: because the fact side is partitioned by `region` while the join is made over `subscriber`, 1738 rows are shuffled under every condition and the makespan does not fall below **3179**; at a dimension of 50 rows it is **3207**. The broadcast join has no floor; at a dimension of 50 rows the makespan is **1926**. - The broadcast makespan grows by one unit per dimension row, the partitioned makespan by nine sixteenths of a unit. The two lines cross: the makespans are equal at 2978 and 2979, and the partitioned join pulls ahead at **2980**. At 4000 the gap is 5876 against **5429**. - When total work is measured, the crossover point again comes out as **2980**, because the ratio of the two measures is the number of workers in both strategies. That they agree is known only when both are measured. - The decision hangs on the network cost: at a cost of 1 the crossover point is **535**, at 3 it is **2980**, and at 6 and 12 the partitioned join cannot pull ahead at any size. Because both strategies match the same rows at every size, the correctness check shows none of this difference. #### Next Step In all the tables of this lesson a number stood there silently: the worker load of the fact side had been printed as `[374, 0, 0, 1826]`. **Two of the four workers had received no rows at all**, and all the makespan calculations had been made over the 1826 rows of the busiest worker. The floor of 1876 units of the broadcast strategy and the floor of 3207 of the partitioned strategy were both fed by this one number. For two lessons the source of the slowness was sought in **movement**; yet in this layout half of the workers were not working at all, and that is a loss independent of movement. The next lesson looks at the partitioning itself: it measures how unbalanced the spread of a key across the workers is as **skew**, partitions the same job by the right and the wrong key, and counts what **adding workers** does for unbalanced partitioning. ### Data Skew Source: https://academia.sh/en/courses/distributed-data-processing/data-skew Measuring an unbalanced partitioning and counting why adding workers is not a fix: when the same 2200 rows are partitioned by the subscriber key, the load of the four workers stays between 539 and 561 and the skew comes out as 1.02; when they are partitioned by the region key, the load becomes 374, 0, 0 and 1826, the skew rises to 3.32 and two workers receive no rows at all. When the number of workers is increased with the wrong key, the skew rises from 1.66 to 3.32, 6.64 and 6.88 and the number of idle workers from 0 to 13; the makespan, which is 2250 on a single worker, rises to 3691 with two workers and drops below the baseline only at sixteen workers, while the total work is carried from 2200 to 8635. With the right key, no rows are shuffled at any number of workers, the total work stays fixed at 2200 and the makespan falls to 215. All three partitionings produce exactly the same region totals. In all the tables of the previous lesson a number stood there silently: the worker load of the fact side, `[374, 0, 0, 1826]`. **Two of the four workers had received no rows at all**, and all the makespan calculations had been made over the 1826 rows of the busiest worker. For two lessons the source of the slowness was sought in **movement**; yet in this layout half of the workers were not working at all, and that is a loss independent of movement. This lesson looks at the partitioning itself. What is measured is how unevenly the rows are spread across the workers, where this imbalance comes from, and what **adding workers** does for an unbalanced partitioning. The last is the clearest example of the second claim of this course. - **PF19.** The source, the consumption rows and the cluster are **fictional** and are generated within the lesson. No distributed processing framework, resource manager or file system is called. The seed is 20260218. - **PF20.** **Data skew** is the ratio of the busiest worker to the average per worker. 1.00 is perfect balance; as it grows, it means that one worker is doing more work than the others. - **PF21.** Makespan is **the work of the busiest worker**, so the skew is written directly into the makespan. Makespan is the model's **own unit**. **Total work** is counted separately. - **PF22.** Partitioning is done by the hash of the key, and **the same key value always goes to the same worker**. This has two consequences: the rows of one key value **cannot be split**, and two separate values **can collide on the same worker**. - **PF23.** The reduce is done over the `subscriber` key in all the tables; the only thing that changes is **by which key the rows are partitioned**. `subscriber` is therefore the right partition key and `region` the wrong one. - **PF24.** `NETWORK_COST` = 3 and `TASK_OVERHEAD` = 50 are **choices**. The skew numbers of this lesson are independent of both; the makespan and total work numbers are not. - **PF25.** The baseline is **a single worker**: no shuffle, skew 1.00, makespan **2250**, total work **2200**. - **PF26.** All the partitionings produce **exactly the same result**; the only thing measured is cost. - **PF27.** The resolution of the cluster: 2200 rows, at most 16 workers. With four workers the step of the skew is 0.0018; the differences read at this scale are far above this step. #### Skew Is the Measure of Partitioning The block below splits the same 2200 rows across four workers by three different keys, computes the skew of each partitioning, and compares the region totals that the three produce. ```python # FICTIONAL. Meter readings, consumption rows and the cluster are generated within the lesson; # no distributed processing framework or resource manager is called. Seed 20260218. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] WORKERS, NETWORK_COST, TASK_OVERHEAD = 4, 3, 50 SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def rows(): """Consumption rows coming out of the pipeline. On a sound source the screen and dedupe stages drop no rows; consumption is found by subtracting the previous period's index.""" raw = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) raw.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) table, output = {(k["subscriber"], k["period"]): k for k in raw}, [] for k in raw: i = PERIODS.index(k["period"]) prev = table.get((k["subscriber"], PERIODS[i - 1])) if i else None if prev is not None: output.append(dict(k, m3=k["index"] - prev["index"])) return output def trace_id(m): h = 2166136261 for c in str(m): h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF return h def distribute(records, key, workers=WORKERS): """Assigns rows to workers by key. The same key always goes to the same worker.""" buckets = {i: [] for i in range(workers)} for k in records: buckets[trace_id(k[key]) % workers].append(k) return buckets def load(buckets): """Skew: the ratio of the busiest worker to the average.""" n = [len(v) for v in buckets.values()] return {"total": sum(n), "busiest": max(n), "least": min(n), "skew": round(max(n) / (sum(n) / len(n)), 4)} def shuffle(buckets, new_key, workers=WORKERS): """Repartition. A row that changes workers crosses the network.""" new, moved = {i: [] for i in range(workers)}, 0 for old, records in buckets.items(): for k in records: target = trace_id(k[new_key]) % workers if target != old: moved += 1 new[target].append(k) return new, moved def cost(buckets, moved=0, workers=WORKERS): """Makespan = the busiest worker's work (parallel). Total work is counted separately.""" n = [len(v) for v in buckets.values()] return {"duration": max(n) + moved * NETWORK_COST // workers + TASK_OVERHEAD, "total_work": sum(n) + moved * NETWORK_COST, "moved": moved} def region_total(buckets): """Each worker first computes the region totals of its own share , then they are merged.""" final = {} for records in buckets.values(): partial = {} for k in records: partial[k["region"]] = partial.get(k["region"], 0) + k["m3"] for b, v in partial.items(): final[b] = final.get(b, 0) + v return tuple(sorted(final.items())) ALL = rows() print(f"rows from the pipeline {len(ALL)} , workers {WORKERS} , network cost {NETWORK_COST}") print("rows per region: " + " , ".join(f"{b} {sum(1 for k in ALL if k['region'] == b)}" for b in REGION)) print("which worker each region lands on: " + " , ".join(f"{b} {trace_id(b) % WORKERS}" for b in REGION)) print("\nkey distinct worker load busiest least skew") for a in ("subscriber", "region", "period"): k, y = distribute(ALL, a), load(distribute(ALL, a)) print(f"{a:<10} {len(set(x[a] for x in ALL)):>12} {str([len(v) for v in k.values()]):<24}" f"{y['busiest']:>6} {y['least']:>6} {y['skew']:>10}") RESULT = {a: region_total(distribute(ALL, a)) for a in ("subscriber", "region", "period")} print(f"\ndo the three partitionings produce a single result: {len(set(RESULT.values())) == 1}") print("region total: " + " , ".join(f"{b} {v}" for b, v in RESULT["subscriber"])) ``` ``` rows from the pipeline 2200 , workers 4 , network cost 3 rows per region: dogu 374 , bati 341 , kuzey 506 , guney 440 , merkez 539 which worker each region lands on: dogu 0 , bati 3 , kuzey 3 , guney 3 , merkez 3 key distinct worker load busiest least skew subscriber 200 [539, 550, 561, 550] 561 539 1.02 region 5 [374, 0, 0, 1826] 1826 0 3.32 period 11 [600, 600, 400, 600] 600 400 1.0909 do the three partitionings produce a single result: True region total: bati 9164 , dogu 9695 , guney 11721 , kuzey 13338 , merkez 14536 ``` The last two lines are the precondition of the whole lesson. The three partitionings produce **exactly the same** region totals; whichever key the partitioning is made by, every row enters the computation exactly once. This is the third claim of this course, and its form here is its harshest one: **no test that checks the result can see the imbalance in the `region` partitioning.** Skew is not a correctness problem but a cost that becomes visible only when it is counted. The first table shows where the skew comes from. The `subscriber` key has 200 distinct values, they spread across the four workers between 539 and 561, and the skew is **1.02**. The `period` key has 11 distinct values (the first period of the pipeline has dropped out because consumption cannot be computed for it), and because 11 values are divided among four workers as 3, 3, 2, 3, the load is 600, 600, 400, 600 and the skew is **1.0909**. The `region` key, on the other hand, has only **5** distinct values, and the skew rises to **3.32**. The reason for the third row is written in the second line. **All four** of the five regions fall on worker 3: 341 plus 506 plus 440 plus 539 comes to exactly 1826. The one remaining region, `dogu` (east), sits alone on worker 0, and two workers receive no rows at all. Two separate mechanisms are stacked here. The first is **the scarcity of keys**: five distinct values can fill at most five workers. The second is **hash collision**: four of the five values have fallen into the same bucket. This has a consequence that must be told apart. The regions are not very unbalanced within themselves; the smallest carries 341 rows and the largest 539, and the ratio between them is below two. So the data is not skewed by itself. What produces the skew is **the joint behavior of the key and the number of workers**. The sentence "the data was skewed" is therefore incomplete; skew is not a property of the data but of **the partitioning**, and the partitioning is a decision. #### Adding Workers Is Not a Fix The first fix that comes to mind for an unbalanced partitioning is more workers. The block below makes the same sweep with two keys: the wrong one (`region`) and the right one (`subscriber`). The reduce is over `subscriber` in both. ```python # This block continues the definitions of rows(), trace_id(), distribute(), load(), shuffle(), # cost() and ALL from the first block. The reduce is done over the subscriber key in both # tables; the only thing that changes is the partition key. def sweep(partition_key): print(f"partition key {partition_key} , reduce subscriber") print("workers skew busiest idle workers moved makespan total work") for workers in (1, 2, 4, 8, 16): k = distribute(ALL, partition_key, workers) _, t = shuffle(k, "subscriber", workers) m, y = cost(k, t, workers), load(k) print(f"{workers:>7} {y['skew']:>10} {y['busiest']:>9}" f" {sum(1 for v in k.values() if not v):>12}" f" {t:>8} {m['duration']:>8} {m['total_work']:>10}") sweep("region") print() sweep("subscriber") ``` ``` partition key region , reduce subscriber workers skew busiest idle workers moved makespan total work 1 1.0 2200 0 0 2250 2200 2 1.66 1826 0 1210 3691 5830 4 3.32 1826 2 1738 3179 7414 8 6.64 1826 6 1958 2610 8074 16 6.88 946 13 2145 1398 8635 partition key subscriber , reduce subscriber workers skew busiest idle workers moved makespan total work 1 1.0 2200 0 0 2250 2200 2 1.0 1100 0 0 1150 2200 4 1.02 561 0 0 611 2200 8 1.04 286 0 0 336 2200 16 1.2 165 0 0 215 2200 ``` The second row of the first table is the **second claim** of this course: where the makespan is **2250** on a single worker, when a second worker is added it **rises** to **3691**. Adding a worker has made the job worse. The reason can be read off: on a single worker there is no shuffle, because all the rows are already in the same place. The moment the second worker appears, the reduce key separates from the partition key, 1210 rows cross the network, and the total work jumps from 2200 to **5830**. Parallelism is not free, and in this layout **its first step is negative**. The makespan begins to fall only after four, but it drops below the single-worker baseline only at **sixteen workers** (1398). It is 3179 at four workers and 2610 at eight, and both are above 2250. So a job partitioned by the `region` key is **still slower** on a cluster of eight workers than on a single worker. Without this table being written, it cannot be known whether the sentence "we enlarged the cluster and it got faster" is true. The busiest worker column gives the mechanism of this, and it is the most striking number of the table: at two, four and eight workers the load of the busiest worker is **fixed at 1826**. While the number of workers quadruples, that worker does not hand over a single row, because the four regions that fall on it stay in the same bucket in all three partitionings. The fall in the makespan (3691, 3179, 2610) comes not from the decrease of the load but only from the transfer share being divided among more workers. The load breaks for the first time at sixteen workers and drops to **946**. The skew column **gets worse** as workers are added: 1.66, 3.32, 6.64, 6.88. The number of idle workers rises from 0 to 2, to 6 and to **13**. In the cluster of sixteen workers only three workers receive rows. The limit of this is written in the key itself: `region` carries five distinct values, so however many workers are added, at most **five** workers can receive rows. Because the rows of one key value cannot be split, the busiest worker cannot drop below the largest region either, that is, below **539** rows. The makespan therefore cannot drop below **589** together with the task overhead, however many workers are added. The sixteenth row of the second table turns this into a comparison: with the `subscriber` key the makespan is already **215**, far below that threshold. **If the partition key is wrong, the number of workers is not a fix**, because a wrong key puts a ceiling on the cluster. The skew rising while the makespan falls looks contradictory at first glance and has to be read separately. Skew is a **ratio**: as the number of workers grows, the average shrinks, while the load of the busiest worker does not shrink at the same rate, so the ratio grows. Makespan is an **absolute number** and falls because the transfer share is spread across the workers. The two do not say the same thing: makespan measures **when the job finishes**, and skew measures **how many workers were set aside for nothing**. The makespan of 1398 at sixteen workers was bought at the price of thirteen workers waiting idle and a total work of **8635** units. When the two tables are put side by side, the rule of the course becomes visible. In the `region` table the makespan falls from 3691 to 1398 while the total work rises from 5830 to **8635**; the two measures go in opposite directions, and a sentence such as "it got four times faster" is **incomplete** unless it is written which one was measured. In the `subscriber` table, on the other hand, the total work stays at **2200** in all five rows, because the shuffled rows are zero at every number of workers. What the right key earns is not only makespan: this is the only layout in which every worker added to the cluster **produces something in return**. #### Summary - Data skew is the ratio of the busiest worker to the average: **1.02** in the `subscriber` partitioning, **1.0909** in the `period` partitioning, **3.32** in the `region` partitioning. In the last layout **two of the four workers are idle** and the load is `[374, 0, 0, 1826]`. - Skew is a property of the partitioning, not of the data. The regions are between 341 and 539 rows, so the data is balanced by itself; what produces the skew is that the key carries only **5** distinct values and that **four of these five values collide on the same worker** (341 + 506 + 440 + 539 = 1826). - Adding workers can make the job worse: on a single worker the makespan is **2250**, on two workers **3691**, because with the second worker 1210 rows cross the network and the total work rises from 2200 to 5830. The makespan drops below the single-worker baseline only at sixteen workers. - As workers are added with the wrong key, the skew rises from 1.66 to **6.88** and the idle workers from 0 to **13**. A key with five distinct values can fill at most five workers, and because the largest region is 539 rows, the makespan cannot drop below **589** at any number of workers; with the `subscriber` key the makespan is already **215** at sixteen workers. - The three partitionings produce exactly the same region totals. No test that checks correctness shows this difference; in the `region` table the makespan falls from 3691 to 1398 while the total work rises from 5830 to **8635**, and in the `subscriber` table the total work stays at 2200 in all five rows. #### Next Step Over three lessons the source of the slowness was gathered under two headings: the **movement** of the rows and the **imbalance** of the workers. In both measures it was assumed that a single computation was done a single time; that two separate queries might need the same intermediate result, or that a worker might fail before the job is done, was never discussed. Yet doing the same computation twice is a common line item in a distributed job, and the failure of a worker raises the question of how much of the work done up to that moment will be asked for again. The next lesson measures the storage of the intermediate result: it counts the difference between computing the same data twice and computing it once and holding it, and writes down the cost of holding itself. ### Caching and Persistence Source: https://academia.sh/en/courses/distributed-data-processing/caching-and-persistence Counting the cost of computing the same intermediate result twice when more than one query reads it, and the cost that comes back when the cache is not held: two queries sharing an intermediate result of 2200 rows total 5500 without a cache and 3300 with one, but with a single query both are 2750 and the cache gains nothing. The intermediate result sits in the memory of four workers as 539, 550, 561 and 550 rows; as workers fail, the total rises from 3300 to 3850, 4411, 4961 and 5500, and a cache that is not held at all pays exactly the cost of the setup without a cache, although the plan was made for 3300. The failure of a single worker moves the total to between 3839 and 3861. The same cache shows a gain of 40% in the mixed total, 25% in total work and 23.91% in makespan, and a claim of speedup is incomplete unless it is written which of the three numbers was measured. The previous three lessons gathered the source of the slowness under two headings: the data **was moving** and the load **was unevenly distributed**. The shuffle carried 1738 rows across the network and raised the total work from 2200 to 7414, the choice of join strategy put the crossover point between 1600 and 4000 rows, and the `region` key left two of the four workers idle and carried the skew to 3.32. In all three, what was measured was a single job running **once**. On a cluster, intermediate results are not read once. One query wants the same consumption table of 2200 rows for the region total, another for the subscriber statement. This lesson asks two questions: how much it costs to do the same computation twice, and, when you decide to do that computation once and store it, **whether what is stored is really there**. - **PF41.** The source, the first three stages of the pipeline and the distributed setup are **fictional** and are generated within the lesson. No distributed processing framework, cluster manager or cache product is called. The seed is 20260218. - **PF42.** The cache hit, the eviction policy and the cache sizing procedure were measured in the In-Memory Stores and Caching Systems course of the Databases curriculum, and the call cache in the Building Applications with Model APIs course of the AI Engineering curriculum; they are **not repeated**. The only measure here is **the reuse of the intermediate result**. - **PF43. Makespan and total work are the model's own unit**, not raw milliseconds. - **PF44.** `two_queries` counts the preparation as **total work** (all the rows) and the query step as a **per-worker share**. This is a mixed measure; the last section of the lesson splits it in two. - **PF45.** The baseline is the setup **without a cache**: the intermediate result is recomputed in every query. `NETWORK_COST = 3` and `TASK_OVERHEAD = 50` are choices. - **PF46.** Holding the intermediate result in memory is not counted as **work**; its cost is **the held rows**, and those rows are lost together with the worker that holds them. #### Computing the Same Thing Twice The first block sets up the source, the three stages of the pipeline up to consumption, and the measure of the two queries. `two_queries` comes from the common definition of the course; `cached` is its generalization and gives the same numbers at the two ends. ```python # FICTIONAL. Meter readings of the measurement network, the first three stages of the pipeline and # the distributed setup are generated within the lesson; no distributed processing framework, # cluster manager or cache product is called. Seed 20260218. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records def s1_screen(records): REQUIRED = ("subscriber", "period", "region", "index", "unit") keep = [k for k in records if all(a in k for a in REQUIRED)] return keep, len(records) - len(keep) def s2_dedupe(records): seen, keep = set(), [] for k in records: key = (k["subscriber"], k["period"]) if key in seen: continue seen.add(key) keep.append(k) return keep, len(records) - len(keep) def s3_consumption(records): table = {(k["subscriber"], k["period"]): k for k in records} out, dropped = [], 0 for k in records: i = PERIODS.index(k["period"]) if i == 0: continue prev = table.get((k["subscriber"], PERIODS[i - 1])) if prev is None: dropped += 1 continue out.append(dict(k, m3=int(k["index"]) - int(prev["index"]))) return out, dropped WORKERS, NETWORK_COST, TASK_OVERHEAD = 4, 3, 50 def rows(): v, _ = s1_screen(generate_source()) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return v def two_queries(records, cache=False, workers=WORKERS): preparation = len(records) if cache: return {"computation": preparation, "query": 2 * (preparation // workers), "total": preparation + 2 * (preparation // workers)} return {"computation": 2 * preparation, "query": 2 * (preparation // workers), "total": 2 * preparation + 2 * (preparation // workers)} # ---- Generalization of the same measure: q queries , the intermediate result is alive as far as the held rows. def cached(n, q, held, workers=WORKERS): """Preparation is counted as total work , the query step as a per-worker share.""" preparation = n + (q - 1) * (n - held) return {"preparation": preparation, "query": q * (n // workers), "total": preparation + q * (n // workers)} INTERMEDIATE = rows() N = len(INTERMEDIATE) print(f"intermediate result {N} rows , workers {WORKERS}") print(f"two queries uncached {two_queries(INTERMEDIATE)['total']} " f"({two_queries(INTERMEDIATE)['computation']} preparation + {two_queries(INTERMEDIATE)['query']} query)" f" cached {two_queries(INTERMEDIATE, True)['total']} " f"({two_queries(INTERMEDIATE, True)['computation']} + {two_queries(INTERMEDIATE, True)['query']})") print(f"generalization {cached(N, 2, 0)['total']} and {cached(N, 2, N)['total']}") print("\n query count uncached cached diff") for q in (1, 2, 3, 4): a, b = cached(N, q, 0)["total"], cached(N, q, N)["total"] print(f"{q:>12} {a:>11} {b:>10} {a - b:>4}") ``` ``` intermediate result 2200 rows , workers 4 two queries uncached 5500 (4400 preparation + 1100 query) cached 3300 (2200 + 1100) generalization 5500 and 3300 query count uncached cached diff 1 2750 2750 0 2 5500 3300 2200 3 8250 3850 4400 4 11000 4400 6600 ``` In the setup without a cache the preparation is done **twice**: half of the 4400 units of computation is regenerating a table that the first query has already produced. The cache deletes this second production, and the total falls from 5500 to 3300. The 2200 gained is **the intermediate result's own row count**, nothing more. The first row shows the limit of this. With a single query the setups without and with a cache pay **the same 2750**; the gain is zero, because there is no computation to reuse. The cache gains 6600 at the fourth query, but at the first it holds 2200 rows in memory and gives nothing in return. **The gain of a cache comes not from the size of the intermediate result but from how many times it is read.** Caching a result that is read once is a decision that has a cost and no gain. #### When the Cache Is Not Held The two numbers so far assumed that the cache **was standing there**. The second block removes that assumption. The intermediate result is distributed to four workers by the `subscriber` key, and each worker holds its own share in its own memory; when a worker fails, only that share is lost. ```python # This block continues the definitions of INTERMEDIATE, N, WORKERS, TASK_OVERHEAD and cached from the first block. def trace_id(m): h = 2166136261 for c in str(m): h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF return h def distribute(records, key, workers=WORKERS): buckets = {i: [] for i in range(workers)} for k in records: buckets[trace_id(k[key]) % workers].append(k) return buckets BUCKETS = distribute(INTERMEDIATE, "subscriber") SHARES = [len(BUCKETS[i]) for i in range(WORKERS)] print(f"worker shares of the intermediate result {SHARES}") print("alive workers held rows total of two queries vs uncached") for alive in (4, 3, 2, 1, 0): t = sum(SHARES[:alive]) tot = cached(N, 2, t)["total"] print(f"{alive:>13} {t:>13} {tot:>20} {tot - 5500:>15}") print("if a single worker fails the total " + " , ".join(str(cached(N, 2, N - p)["total"]) for p in SHARES)) # ---- The same setup , two separate measures: makespan (per worker) and total work. def duration_work(n, q, held, workers=WORKERS): preparation = n + (q - 1) * (n - held) return {"duration": preparation // workers + q * (n // workers) + q * TASK_OVERHEAD, "total_work": preparation + q * n} empty, full = duration_work(N, 2, 0), duration_work(N, 2, N) print("\nmeasure uncached cached gain") for name, a, b in (("mixed total", 5500, 3300), ("total work", empty["total_work"], full["total_work"]), ("makespan", empty["duration"], full["duration"])): print(f"{name:<12} {a:>11} {b:>10} {round((a - b) / a * 100, 2):>5} %") ``` ``` worker shares of the intermediate result [539, 550, 561, 550] alive workers held rows total of two queries vs uncached 4 2200 3300 -2200 3 1650 3850 -1650 2 1089 4411 -1089 1 539 4961 -539 0 0 5500 0 if a single worker fails the total 3839 , 3850 , 3861 , 3850 measure uncached cached gain mixed total 5500 3300 40.0 % total work 8800 6600 25.0 % makespan 2300 1750 23.91 % ``` The last row of the first table is the core of the lesson: a cache that is not held at all pays **exactly 5500**, that is, the cost of the setup without a cache. Not a unit more and not a unit less. But the person who made the plan decided by looking at 3300; when the job is done the result that comes out is the same, the queries are correct, no check fires, and the only difference is **the cost paid**. This is the form that the third claim of the course takes in this lesson: **what goes wrong shows up not in the output but only in the number.** The middle rows show how the gain melts away. As the held rows decrease, the total climbs linearly, and the gain is always **as many as the held rows**: 2200 if 2200 rows are alive, 1089 if 1089 rows are alive. A cache with three quarters alive gives three quarters of the gain. Even the failure of a single worker moves the total from 3300 to between 3839 and 3861; which worker fails makes a difference of 22 units, because the shares are not equal between 539 and 561. **Persistence** is the decision beneath this table. Holding the intermediate result in memory ties it to the life of the worker; writing it to a durable place leaves the held rows at 2200 when a worker fails. In return a write job is added, and that job is paid **for certain**, regardless of how much of the rows will actually be read again. The gain of the cache is probabilistic, the cost of persistence is certain. #### The Layout of the Cache Up to here the cache has been treated as a **set of rows**. In a distributed cluster it is not so: the intermediate result sits distributed across the workers by a key, and that layout determines whether the query that will read it needs a shuffle. The third block names two queries (one asks for the region total, the other for the subscriber statement) and changes the layout in which the cache is held. ```python # This block continues the definitions of INTERMEDIATE, N, BUCKETS, WORKERS, NETWORK_COST, trace_id # and distribute from the first two blocks. def shuffle(buckets, new_key, workers=WORKERS): new = {i: [] for i in range(workers)} moved = 0 for old, records in buckets.items(): for k in records: target = trace_id(k[new_key]) % workers if target != old: moved += 1 new[target].append(k) return new, moved BY_REGION, sub_to_region = shuffle(BUCKETS, "region") _, region_to_sub = shuffle(BY_REGION, "subscriber") print(f"moved from the subscriber layout to region {sub_to_region} , from the region layout to subscriber {region_to_sub}") print("cache layout held rows shuffled in 4 queries network work") for name, held, moved in (("subscriber", N, 2 * sub_to_region), ("region", N, 2 * region_to_sub), ("two copies", 2 * N, sub_to_region)): print(f"{name:<18} {held:>9} {moved:>21} {moved * NETWORK_COST:>12}") ``` ``` moved from the subscriber layout to region 1738 , from the region layout to subscriber 1738 cache layout held rows shuffled in 4 queries network work subscriber 2200 3476 10428 region 2200 3476 10428 two copies 4400 1738 5214 ``` Both directions move **the same 1738 rows**; in which layout it is stored does nothing other than choose which query will be free. If two of the four queries ask for the region and two for the subscriber, the cache with a single layout moves 3476 rows and, at `NETWORK_COST = 3`, produces 10428 units of network work. Holding two copies brings this number down to 5214, because the second copy is built once and is never shuffled afterward. The cost is visible in memory: the held rows rise from 2200 to **4400**, and according to the table of the previous section, every worker that fails now takes a share of both copies with it. The cache here is not a set of rows but **a set of rows plus a layout**; storing a second layout is holding a second cache. #### Which Number Fell The last table of the second block writes a single cache decision with three measures, and all three are correct. The mixed total shows a gain of 40%, because it counts the preparation with all the rows and the query with a per-worker share. Total work shows 25%: in the figure that falls from 8800 to 6600, the query step is also counted with all the rows, and the cache does not touch it. Makespan shows 23.91%; the preparation per worker falls from 1100 to 550, but every query still pays its own `TASK_OVERHEAD = 50`. The same change has three separate speedup numbers, and the largest gap among them is 16%. For a setup, the sentence "the cache sped the job up by forty percent" is not a verifiable claim unless it is written which number fell. **Makespan and total work are measured separately**; in this lesson the two went in the same direction, but that is not a rule, it is the result of this setup. #### Summary - The two queries that share an intermediate result of 2200 rows total **5500** without a cache and **3300** with one; the 2200 gained is the intermediate result's own row count. - With a single query both setups pay **2750**. The gain of a cache comes not from the size of the intermediate result but from how many times it is read; caching a result that is read once is a decision that has a cost and no gain. - The intermediate result sits in four workers as 539, 550, 561 and 550 rows. As workers fail, the total rises from 3300 to 3850, 4411, 4961 and **5500**; a cache that is not held at all pays exactly the cost of the setup without a cache. - The gain is always as many as the held rows, and the failure of a single worker moves the total to between 3839 and 3861. A durable write leaves the held rows at 2200, and in return the write cost is paid for certain. - A cache is not a set of rows but a set of rows plus a layout: both directions move the same **1738** rows, with four queries a single layout produces 3476 rows and 10428 units of network work, and two copies bring this down to 5214 while raising the held rows from 2200 to 4400. - The same cache shows a gain of **40%** in the mixed total, **25%** in total work and **23.91%** in makespan; a claim of speedup that does not say which was measured is incomplete. #### Next Step The last table of this lesson counted the failure of a worker only as **the loss of the cache**: what disappeared was a stored intermediate result, and it could be recomputed. In a job that flows continuously, however, what sits in the worker's memory is not an intermediate result but **the state of the computation up to that moment**, and when it is to be recomputed, it is not clear where to start. The next lesson counts **how much a failed worker asks to have back**: how many rows are recomputed in a job that never saves the state, how far raising the checkpoint frequency lowers this number, and what it costs to prevent the same row from being counted twice. ### State in Stream Engines Source: https://academia.sh/en/courses/distributed-data-processing/state-in-stream-engines Counting how much a failed worker asks to have back in a job that flows continuously, and whether raising the checkpoint frequency really lowers that number: in a streaming job of 2200 rows, when a failure occurs in period 2026-09, 1600 rows are recomputed without a checkpoint, 400 at interval 6, again 400 at interval 3, and 0 at interval 1. Halving the interval does not halve the recovery, because the last point before the failure has not moved; interval 4 gives 0 on the same failure. When all eleven failure points are scanned, interval 6 makes 545.5 rows on average be recomputed and interval 3 makes 218.2, but the certain cost rises from 400 to 800; by the worst case interval 4 and 3 tie, by the average interval 4 wins. When the failure probability drops from 1 to 0.5, the best choice turns back to holding no checkpoint. With partitioned state a single failed worker asks for 408 rows instead of 1600 and 102 instead of 400, and exactly-once requires holding exactly that many rows of state. The previous lesson counted a failed worker only as **the loss of the cache**: what disappeared was a stored intermediate result, it could be recomputed, and the total of two queries rose from 3300 to 5500. That is what is right in a batch job, because the input is still standing where it was. In a job that flows continuously the situation is different. What sits in the worker's memory is not an intermediate result but **the state** of the computation up to that moment: which subscriber has been reached up to which period, where which partial total has stopped. When a failure occurs, this state is lost and the job has to start again **from somewhere**. This lesson counts a single question: how much the failed worker asks to have back, and whether the cost paid to shrink that number really pays for itself. - **PF47.** The source, the pipeline and the distributed setup are **fictional** and are generated within the lesson. No stream engine, cluster manager or state store is called. The seed is 20260218. - **PF48.** The procedures for the checkpoint, resilience to duplicates and exactly-once semantics were measured in the Data Ingestion topic of the Data Pipelines and Orchestration course of the Data Engineering and MLOps curriculum and in the Caching, Queues and Asynchronous Processing course of the Backend Development curriculum; they are **not repeated**. The numbers there are **input** here: 2200 rows at least once with 19 duplicates, 2162 rows at most once. The only measure here is **the cost of recovery**. - **PF49.** The streaming job advances period by period and **the failure occurs in period `2026-09`**; the job continues from the last checkpoint before the failure. In the code the `checkpoints` field carries the number of checkpoints. - **PF50.** A checkpoint makes the current state durable; the state is **one entry per subscriber**, that is, `STATE_ROWS = 200` rows. This is a choice, and all the totals below are sensitive to it. - **PF51.** When averaging, the failure is assumed to occur **with equal probability in any one** of the eleven periods; this too is a choice. The row is the model's own unit, not raw milliseconds. - **PF52.** The baseline is the setup **that holds no checkpoint**: point 0, certain cost 0, and after a failure it starts over from the beginning. #### The Cost of Recovery The first block sets up the source, the three stages of the pipeline up to consumption, and the `recovery` measure. `recovery` counts how many of the rows up to the failure period will be recomputed. ```python # FICTIONAL. Meter readings of the measurement network, the first three stages of the pipeline and # the streaming job are generated within the lesson; no stream engine , cluster manager # or state store is called. Seed 20260218. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records def s1_screen(records): REQUIRED = ("subscriber", "period", "region", "index", "unit") keep = [k for k in records if all(a in k for a in REQUIRED)] return keep, len(records) - len(keep) def s2_dedupe(records): seen, keep = set(), [] for k in records: key = (k["subscriber"], k["period"]) if key in seen: continue seen.add(key) keep.append(k) return keep, len(records) - len(keep) def s3_consumption(records): table = {(k["subscriber"], k["period"]): k for k in records} out, dropped = [], 0 for k in records: i = PERIODS.index(k["period"]) if i == 0: continue prev = table.get((k["subscriber"], PERIODS[i - 1])) if prev is None: dropped += 1 continue out.append(dict(k, m3=int(k["index"]) - int(prev["index"]))) return out, dropped WORKERS, NETWORK_COST, TASK_OVERHEAD, STATE_ROWS = 4, 3, 50, 200 def rows(): v, _ = s1_screen(generate_source()) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return v def recovery(records, checkpoint_interval, failed_period="2026-09", workers=WORKERS): """Without a checkpoint , from the start; with one , from the last point.""" i = PERIODS.index(failed_period) if checkpoint_interval is None: recomputed = sum(1 for k in records if PERIODS.index(k["period"]) <= i) return {"checkpoints": 0, "recomputed": recomputed} last = (i // checkpoint_interval) * checkpoint_interval recomputed = sum(1 for k in records if last < PERIODS.index(k["period"]) <= i) return {"checkpoints": len(PERIODS) // checkpoint_interval, "recomputed": recomputed} STREAM = rows() print(f"stream job {len(STREAM)} rows , failure 2026-09") print("interval checkpoints recomputed rows") for a in (None, 6, 3, 1): s = recovery(STREAM, a) print(f"{str(a):>8} {s['checkpoints']:>11} {s['recomputed']:>15}") ``` ``` stream job 2200 rows , failure 2026-09 interval checkpoints recomputed rows None 0 1600 6 2 400 3 4 400 1 12 0 ``` Three of the four rows are what one would expect. The job that holds no checkpoint recomputes **all 1600 rows** up to the failure period; interval 6 lowers this to 400; interval 1 lowers it to zero because it saves every period. The third row is not what one would expect: **going from interval 6 to interval 3 does not change the recomputed rows at all.** The interval has been cut in two, the number of held points has gone from 2 to 4, and the recovery has stayed at 400. What is paid has doubled; what is received has stayed the same. #### Halving the Interval The reason stands in a single calculation. The failure is `2026-09`, that is, the eighth in the index of twelve. At interval 6 the last point before the failure is the sixth period; at interval 3 it is also the sixth period, because the remainder of dividing eight by three, two, carries the last point back to six. **The interval was halved but the last point before the failure did not move.** The second block scans this behavior across all eleven failure points. ```python # This block continues the definitions of STREAM, PERIODS, STATE_ROWS and recovery from the first block. INTERVALS = (None, 6, 4, 3, 2, 1) print("failure " + " ".join(f"{str(a):>5}" for a in INTERVALS)) for d in PERIODS[1:]: print(f"{d} " + " ".join(f"{recovery(STREAM, a, d)['recomputed']:>5}" for a in INTERVALS)) print("\ninterval checkpoints certain cost worst worst total average average total") for a in INTERVALS: y = [recovery(STREAM, a, d)["recomputed"] for d in PERIODS[1:]] k = recovery(STREAM, a)["checkpoints"] * STATE_ROWS o = sum(y) / len(y) print(f"{str(a):>8} {recovery(STREAM, a)['checkpoints']:>11} {k:>12} {max(y):>5}" f" {max(y) + k:>11} {round(o, 1):>7} {round(o + k, 1):>13}") ``` ``` failure None 6 4 3 2 1 2026-02 200 200 200 200 200 0 2026-03 400 400 400 400 0 0 2026-04 600 600 600 0 200 0 2026-05 800 800 0 200 0 0 2026-06 1000 1000 200 400 200 0 2026-07 1200 0 400 0 0 0 2026-08 1400 200 600 200 200 0 2026-09 1600 400 0 400 0 0 2026-10 1800 600 200 0 200 0 2026-11 2000 800 400 200 0 0 2026-12 2200 1000 600 400 200 0 interval checkpoints certain cost worst worst total average average total None 0 0 2200 2200 1200.0 1200.0 6 2 400 1000 1400 545.5 945.5 4 3 600 600 1200 327.3 927.3 3 4 800 400 1200 218.2 1018.2 2 6 1200 200 1400 109.1 1309.1 1 12 2400 0 2400 0.0 2400.0 ``` The matrix shows by name where the gain comes from: **from the distance of the failure to the last point.** In `2026-07` interval 6 gives zero, because the failure has landed exactly on the point; the same interval gives 1000 in `2026-12`. Interval 4 gives 0 in `2026-09` while interval 3 gives 400 there; the setup that holds points more often is worse in that failure. None of the six columns beats the others in every row. The average column erases this noise and behaves regularly: 1200.0 → 545.5 → 327.3 → 218.2 → 109.1 → 0. **On average, halving the interval roughly halves the recovery; in a single failure it does not.** A setup experiences a single failure, not the average; whoever decides by the average meets the failure that actually happens. The last two columns add the cost. Holding a checkpoint pays a certain cost over `STATE_ROWS = 200`: 400 at interval 6, 800 at interval 3, 2400 at interval 1. By the worst case the cheapest setup is **interval 4 and interval 3 with 1200**, and by the average **interval 4 with 927.3**. The two criteria give two different answers, and the rule "halve the interval" is the best in neither. #### Certain Cost and Probabilistic Gain The totals above carry a silent assumption: that there is a failure in every run. The third block relaxes this assumption and gives the same decision again at four failure probabilities. Then it splits the job across four workers and counts how much the single failed worker asks to have back. ```python # This block continues the definitions of STREAM, PERIODS, WORKERS, STATE_ROWS, INTERVALS and # recovery from the first two blocks. def trace_id(m): h = 2166136261 for c in str(m): h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF return h def distribute(records, key, workers=WORKERS): buckets = {i: [] for i in range(workers)} for k in records: buckets[trace_id(k[key]) % workers].append(k) return buckets print("failure probability " + " ".join(f"{str(a):>6}" for a in INTERVALS) + " best") for p in (1.0, 0.75, 0.5, 0.25): s = [] for a in INTERVALS: y = [recovery(STREAM, a, d)["recomputed"] for d in PERIODS[1:]] s.append(p * sum(y) / len(y) + recovery(STREAM, a)["checkpoints"] * STATE_ROWS) print(f"{p:>19} " + " ".join(f"{round(v, 1):>6}" for v in s) + f" {str(INTERVALS[s.index(min(s))]):>6}") BUCKETS = distribute(STREAM, "subscriber") print("\ninterval whole job from scratch busiest worker with partitioned state exactly-once state") for a in (None, 6, 3, 1): g = recovery(STREAM, a)["recomputed"] per = max(recovery(BUCKETS[i], a)["recomputed"] for i in range(WORKERS)) print(f"{str(a):>8} {g:>22} {per:>37} {g:>18}") ``` ``` failure probability None 6 4 3 2 1 best 1.0 1200.0 945.5 927.3 1018.2 1309.1 2400.0 4 0.75 900.0 809.1 845.5 963.6 1281.8 2400.0 6 0.5 600.0 672.7 763.6 909.1 1254.5 2400.0 None 0.25 300.0 536.4 681.8 854.5 1227.3 2400.0 None interval whole job from scratch busiest worker with partitioned state exactly-once state None 1600 408 1600 6 400 102 400 3 400 102 400 1 0 0 0 ``` The first table shows that the interval decision depends **not on the workload but on the failure frequency**. If there is a failure in every run the best choice is interval 4; at a probability of three quarters it moves to interval 6; when the probability is halved the best choice is **to hold no checkpoint at all**. The reason is one sentence: **the cost of a checkpoint is certain, its gain is probabilistic.** The certain cost is paid in every run; the gain materializes only when a failure occurs and only as much as the distance of the failure to the last point. #### How Much the Failed Worker Asks to Have Back The second table answers the opening question of the lesson. If the job is split across four workers by the `subscriber` key and the state is held at each worker **for its own partition**, the single failed worker asks to have back not the whole job but only its own share: **408** rows instead of 1600, **102** instead of 400. Partitioned state lowers the recovery to a quarter, but only on the condition that the state is held partitioned. If the state is gathered in one place, or if the engine restarts the whole job on a failure, the number asked back is again 1600 and 400. **Partitioning makes recovery cheaper only if the state is partitioned too.** The last column is the cost of **exactly-once** semantics, and its number is surprisingly familiar. The rows recomputed during recovery are written to the output **a second time**; the only way to prevent this is to remember which rows have been written since the last point. That set is exactly the set of recomputed rows: 1600 when there is no interval, 400 at intervals 6 and 3, 0 at interval 1. **Exactly-once is holding as much state as the rows the recovery recomputes**; making the interval denser makes recovery cheaper while it also shrinks this state, and the two gains come from the same number. This set is itself a state, and it is lost together with the failed worker. If the list of written rows stands only in memory, the recovery loses that too, and which row went to the output becomes unknown; exactly-once is achieved only if the list is taken into the checkpoint. This carries the certain cost beyond `STATE_ROWS = 200`: the point now holds not only the partial totals but also the identities of the rows written since the last point. At interval 6 that is 400 identities, and at interval 3 also 400. What happens if this is not held was measured in the Data Pipelines and Orchestration course: the at-least-once setup lets 2200 rows through, 19 duplicates occur and the reported deviation **stays empty**. The third claim of the course holds here too: as long as exactly-once is achieved, all four of the four intervals produce **the same result**. There is no way to tell from the result which interval was chosen; the difference is only in the cost paid. #### Summary - In the streaming job of 2200 rows, on a failure in `2026-09` the recomputed rows are **1600** with no checkpoint, **400** at interval 6, **again 400** at interval 3, and **0** at interval 1. - Halving the interval does not halve the recovery: the last point before the failure has not moved. On the same failure interval 4 gives **0** and interval 3 gives **400**; the setup that holds points more often can be worse. - The average over the eleven failure points is regular (1200.0 → 545.5 → 327.3 → 218.2 → 109.1 → 0), but a setup experiences a single failure, not the average. By the worst case the cheapest is interval 4 and 3 with 1200, by the average interval 4 with 927.3. - The cost of a checkpoint is certain, its gain is probabilistic: at failure probability 1 the best interval is 4, at 0.75 it is 6, and at 0.5 and below it is **to hold no checkpoint**. - With partitioned state the single failed worker asks for **408** rows instead of 1600 and **102** instead of 400; partitioning makes recovery cheaper only if the state is partitioned too. - Exactly-once is holding as much state as the rows the recovery recomputes: 1600, 400, 400 and 0. As long as this is achieved, all four intervals produce the same result and the difference is only in the cost. #### Next Step Five lessons wrote five separate names of the cost: the shuffled row, the copy, the skew, the held row and the recomputed row. Beneath all of them stood the same two numbers, **how long a job took** and **how much work it did in total**, and in every lesson these two were written separately. One question remains: with how many workers a cluster should be set up. The next lesson sweeps the number of workers from one to sixteen and shows that the total work rises while the makespan falls; it counts that this decision cannot be made without writing which one is paid, and closes the course. ### Cost and Cluster Sizing Source: https://academia.sh/en/courses/distributed-data-processing/cost-and-cluster-sizing Counting that the decision of how many workers a cluster is set up with is made between makespan and total work, and that the two numbers go in opposite directions: the same job of 2200 rows gives a makespan of 2250, 3691, 3179, 2610 and 1398 and a total work of 2200, 5830, 7414, 8074 and 8635 on one, two, four, eight and sixteen workers; as the makespan falls the total work rises, and the sizing decision cannot be made without writing which one is paid. The second worker raising the makespan from 2250 to 3691 comes from the shuffle coming into being, and even with a good key of skew 1.00 a rise from 2250 to 2800 is seen. The product of workers and makespan climbs one way from 2250 to 22368, when the fixed cost is counted per worker the gain of sixteen workers falls from 852 to 102, and when the network cost rises to 12 the setup with the shortest makespan returns to a single worker. The lesson closes the course and gathers the numbers of eleven lessons in one table. The previous lesson wrote the fifth name of the cost: the recomputed row. Before it had come the shuffled row, the copy, the skew and the held row. Beneath the five names there stood always the same two numbers: **how long** a job took and **how much work** it did in total. What remains is the question the course is asked most often: with how many workers should this cluster be set up. The question looks innocent and has no answer, because it is incomplete. Growing a cluster changes two numbers at the same time, and they go in **opposite directions**. This lesson sweeps the number of workers from one to sixteen, writes both at every step, and counts which measure, which choice and which partition key the decision hangs on. - **PF53.** The source, the pipeline and the distributed setup are **fictional** and are generated within the lesson. No distributed processing framework, cluster manager or cloud service is called. The seed is 20260218. - **PF54.** The procedure of cost accounting was measured in the Cloud Governance, Security and Cost course of the Cloud Computing curriculum and in the Building Applications with Model APIs course of the AI Engineering curriculum; it is **not repeated**. Here the cost is written only as **makespan and total work**; no currency or price is used. - **PF55. Makespan is the work of the busiest worker** and is the model's own unit, not raw milliseconds. **Total work** is the sum of the work done by all the workers. - **PF56.** `NETWORK_COST = 3` and `TASK_OVERHEAD = 50` are choices; the result of this lesson is **sensitive to both**, and below both are handled together with their values. - **PF57.** `cost` counts the fixed cost **once per job**; the Resource Managers lesson counts the same cost **per worker**. The two ways of counting are written here side by side as two columns. - **PF58.** The baseline is the single worker: no shuffle, skew 1.00, makespan **2250**, total work **2200**. #### The Worker Sweep The first block sets up the source, the three stages of the pipeline up to consumption, and the measures of partitioning and cost. The job measured is the job used throughout the course: the rows are partitioned by the `region` key and then shuffled to the `subscriber` key. ```python # FICTIONAL. Meter readings of the measurement network, the first three stages of the pipeline and # the distributed setup are generated within the lesson; no distributed processing framework, # cluster manager or cloud service is called. Seed 20260218. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records def s1_screen(records): REQUIRED = ("subscriber", "period", "region", "index", "unit") keep = [k for k in records if all(a in k for a in REQUIRED)] return keep, len(records) - len(keep) def s2_dedupe(records): seen, keep = set(), [] for k in records: key = (k["subscriber"], k["period"]) if key in seen: continue seen.add(key) keep.append(k) return keep, len(records) - len(keep) def s3_consumption(records): table = {(k["subscriber"], k["period"]): k for k in records} out, dropped = [], 0 for k in records: i = PERIODS.index(k["period"]) if i == 0: continue prev = table.get((k["subscriber"], PERIODS[i - 1])) if prev is None: dropped += 1 continue out.append(dict(k, m3=int(k["index"]) - int(prev["index"]))) return out, dropped WORKERS, NETWORK_COST, TASK_OVERHEAD = 4, 3, 50 def rows(): v, _ = s1_screen(generate_source()) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return v def trace_id(m): h = 2166136261 for c in str(m): h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF return h def distribute(records, key, workers=WORKERS): buckets = {i: [] for i in range(workers)} for k in records: buckets[trace_id(k[key]) % workers].append(k) return buckets def load(buckets): n = [len(v) for v in buckets.values()] return {"total": sum(n), "busiest": max(n), "least": min(n), "skew": round(max(n) / (sum(n) / len(n)), 4)} def shuffle(buckets, new_key, workers=WORKERS): new = {i: [] for i in range(workers)} moved = 0 for old, records in buckets.items(): for k in records: target = trace_id(k[new_key]) % workers if target != old: moved += 1 new[target].append(k) return new, moved def cost(buckets, moved=0, workers=WORKERS, network=NETWORK_COST, overhead=TASK_OVERHEAD): """Makespan = the work of the busiest worker (parallel). Total work is counted separately.""" n = [len(v) for v in buckets.values()] return {"duration": max(n) + moved * network // workers + overhead, "total_work": sum(n) + moved * network, "moved": moved} ALL = rows() COUNTS = (1, 2, 4, 8, 16) print(f"job {len(ALL)} rows , partition key region , shuffle to subscriber") print("workers skew moved makespan total work workers x makespan makespan (overhead per worker)") for i in COUNTS: bk = distribute(ALL, "region", i) _, t = shuffle(bk, "subscriber", i) m = cost(bk, t, i) mp = cost(bk, t, i, overhead=i * TASK_OVERHEAD) print(f"{i:>7} {load(bk)['skew']:>4} {t:>5} {m['duration']:>8} {m['total_work']:>10}" f" {i * m['duration']:>18} {mp['duration']:>30}") ``` ``` job 2200 rows , partition key region , shuffle to subscriber workers skew moved makespan total work workers x makespan makespan (overhead per worker) 1 1.0 0 2250 2200 2250 2250 2 1.66 1210 3691 5830 7382 3741 4 3.32 1738 3179 7414 12716 3329 8 6.64 1958 2610 8074 20880 2960 16 6.88 2145 1398 8635 22368 2148 ``` The makespan column does not fall in a straight line: it **rises** from 2250 to 3691, then falls to 3179, 2610 and 1398. It drops below the single worker only at sixteen workers. The total work column, on the other hand, never turns back: from 2200 to 5830, to 7414, to 8074 and to 8635 it **climbs one way**. The same set of rows, the same result, four times the total work. The second row is the clearest example of the second claim of the course. With a single worker there is no shuffle; the moment the second worker is added, 1210 rows have to cross the network, the total work jumps from 2200 to 5830, and because the share of the busiest worker is still large, the makespan also rises. **Parallelism is not free, and its first step is negative.** The last two columns show how the decision turns. The product of workers and makespan, which is the equivalent of renting the cluster for the length of the job, climbs from 2250 to **22368** without ever falling; by this measure the best setup is the single worker. When the fixed cost is counted per worker, the makespan of sixteen workers rises from 1398 to **2148**, and its advantage over the single worker falls from 852 units to **102 units**. A choice as small as `TASK_OVERHEAD = 50` nearly erases the gain of a setup sixteen times as large. #### What the Decision Hangs On The three measures gave three separate answers. The next question is how much these answers depend on the two choices in the model and on the partition key. The second block sweeps the network cost, and then repeats the same sweep with a key that has no skew. ```python # This block continues the definitions of ALL, COUNTS, distribute, load, shuffle and cost from the first block. print("network cost " + " ".join(f"{i:>5}" for i in COUNTS) + " shortest makespan") for network in (1, 3, 6, 12): s = [] for i in COUNTS: bk = distribute(ALL, "region", i) _, t = shuffle(bk, "subscriber", i) s.append(cost(bk, t, i, network=network)["duration"]) print(f"{network:>12} " + " ".join(f"{v:>5}" for v in s) + f" {COUNTS[s.index(min(s))]:>17}") print("\npartition key subscriber , shuffle to period") print("workers skew moved makespan total work") for i in COUNTS: bk = distribute(ALL, "subscriber", i) _, t = shuffle(bk, "period", i) m = cost(bk, t, i) print(f"{i:>7} {load(bk)['skew']:>4} {t:>5} {m['duration']:>8} {m['total_work']:>10}") ``` ``` network cost 1 2 4 8 16 shortest makespan 1 2250 2481 2310 2120 1130 16 3 2250 3691 3179 2610 1398 16 6 2250 5506 4483 3344 1800 16 12 2250 9136 7090 4813 2604 1 partition key subscriber , shuffle to period workers skew moved makespan total work 1 1.0 0 2250 2200 2 1.0 1100 2800 5500 4 1.02 1651 1849 7153 8 1.04 1927 1058 7981 16 1.2 2066 602 8398 ``` The first table shows what it means that `NETWORK_COST` is a choice. At a cost of 1, 3 and 6 sixteen workers give the shortest makespan; at a cost of 12 the winner **returns to the single worker**, because the number 2604 is above 2250. Once the ratio of the network to local processing changes, the case for a cluster sixteen times as large disappears. **The sizing decision is not a decision of the measured job but of the ratio between the network and the computation**, and without that ratio written the result cannot be carried. The second table separates out the share of skew. With the `subscriber` key the skew does not exceed 1.20 up to sixteen workers, and the makespan falls to 602, far below the 1398 of the `region` key. On the other hand **the second worker raising the makespan happens here too**: from 2250 to 2800. The shuffle comes into being even with a good key, and at the first step it is larger than its gain. The second claim of the course is not a consequence of skew; skew only deepens it. #### Summary - The same job of 2200 rows gives a makespan of **2250, 3691, 3179, 2610, 1398** and a total work of **2200, 5830, 7414, 8074, 8635** on one, two, four, eight and sixteen workers; as the makespan falls the total work rises. - The second worker raises the makespan from 2250 to **3691**: the shuffle comes into being, 1210 rows cross the network and the total work jumps to 5830. The makespan falls below the single worker only at sixteen workers. - Three measures choose three separate setups: makespan chooses sixteen workers, total work the single worker, and the product of workers and makespan (from 2250 to **22368**) again the single worker. The decision cannot be made without writing which measure is paid. - When `TASK_OVERHEAD = 50` is counted per worker, the makespan of sixteen workers becomes 2148 and the gain falls from 852 to **102**; when `NETWORK_COST` rises to 12, the setup with the shortest makespan **returns to the single worker**. - With the `subscriber` key, which has no skew, the makespan falls as far as 602, but the second worker raising the makespan from 2250 to **2800** happens there too; the cost of the shuffle at the first step is independent of skew. #### Course Wrap-Up The rows carry no common unit; **by which measure** each row is read stands next to the lesson name. The difference in parentheses is **the distributed setup minus the single-worker baseline**; the direction of the sign is read from the measure name. | Lesson | Single-worker baseline | Distributed setup | Cost | |---|---|---|---| | Cluster Computing Fundamentals — makespan and total work | single worker makespan **2250**, total work **2200** | four workers with `subscriber` makespan **611** (**−1639**), total work **2200** (**0**) | total work is 2200 in all four setups and does not tell them apart | | Distributed File Systems — locally read blocks and makespan | single copy **5/22** local, makespan **1850** | replication 3 gives **22/22** (**+17**), makespan **650** (**−1200**) | the stored blocks rise from 22 to **66**, and a fourth copy adds nothing to locality | | MapReduce Model — shuffled rows and total work | map shuffled **0**, total work **2200** | direct reduction shuffled **1738** (**+1738**), total work **7414** (**+5214**) | pre-reduction gives the same result with 20 intermediate rows and a total work of **2245** | | Distributed Datasets — rows read and partitions opened | eager plan **6556** rows, **44** partitions | lazy plan **200** rows (**−6356**), **1** partition (**−43**) | a type error in the plan shows up only on the first action, after 200 rows have been read | | Resource Managers — makespan and fixed cost paid | single slot makespan **3000**, fixed cost **800** | sixteen slots makespan **215** (**−2785**), fixed cost **800** (**0**) | when every task is given a slot, the gain from eight to sixteen stays at **121** and is beaten by the added **400** | | Shuffle Cost — shuffled rows and makespan | shuffled **0** in the `subscriber` layout, makespan **611** | shuffling to `region` moves **1738** (**+1738**), makespan **1914** (**+1303**) | the total work rises from 2200 to **7414**; 5214 units are pure movement that does no computation | | Joins at Scale — join makespan, size of 4000 rows | broadcast **5876** | partitioned **5429** (**−447**) | the crossover point is **2980**, and at network costs 6 and 12 partitioned cannot get ahead at any size | | Data Skew — skew and makespan | single worker skew **1.00**, makespan **2250** | two workers with `region` skew **1.66** (**+0.66**), makespan **3691** (**+1441**) | at sixteen workers skew **6.88**, **13** idle workers; the makespan cannot fall below **589** at any number | | Caching and Persistence — total of two queries | uncached **5500** | cached **3300** (**−2200**) | if the cache is not held the total is again **5500**, and with a single query the gain is **0** | | State in Stream Engines — recomputed rows | no checkpoint **1600** | interval 3 gives **400** (**−1200**) | the certain cost rises from 0 to **800**, yet interval 6 gives the same **400** | | Cost and Cluster Sizing — makespan and total work | single worker makespan **2250**, total work **2200** | sixteen workers makespan **1398** (**−852**), total work **8635** (**+6435**) | the second worker raises the makespan to **3691**; when the fixed cost is counted per worker the gain falls to **102** | The table is a single rule written eleven times. **The number of a distributed job is not one number: the makespan and the total work are measured separately, and a claim of speedup that does not say which was measured is incomplete.** The places where the two columns go in opposite directions in the same row are the rule itself: in MapReduce Model the makespan falls while the total work rises from 2200 to 7414, and in this lesson the makespan falls by 852 units while the total work climbs by 6435 units. Cluster Computing Fundamentals shows its mirror: the makespan of the four setups varies between 611 and 2250 while the total work is 2200 in all four, so a report that writes only the total work cannot tell any of these setups apart. **The second claim is read from the same table: adding workers makes the job worse.** In Data Skew the second worker raises the makespan from 2250 to 3691 and carries the total work to 5830; in this lesson the same step raises the makespan to 2800 even with the `subscriber` key, whose skew is 1.00. In Resource Managers the 121 units gained in going from eight to sixteen tasks are beaten by the added 400 units of fixed cost. In Distributed File Systems a fourth copy raises the stored block to 88 and adds nothing to locality. In Caching and Persistence, in a job with a single query the cache holds 2200 rows and gains 0, and in State in Stream Engines lowering the interval from six to three raises the certain cost from 400 to 800 and changes the recovery not at all. In six lessons **the added resource was paid for and nothing was received in return.** **The third claim is quieter: the result is the same in every setup.** In MapReduce Model the five region totals presented by the three setups are exactly the same, although one spends 7414 and another 2245 of total work. In Shuffle Cost the output does not change while the makespan rises from 611 to 1914, in Joins at Scale the two join strategies match the same rows at every size, in Data Skew the three partitionings produce the same region totals, in Caching and Persistence a cache that is not held changes not the result but only the cost, and in State in Stream Engines all four intervals give the same result as long as exactly-once is achieved. **No check that tests correctness shows a wrong partitioning**; a wrong partitioning becomes visible only when it is counted, and as long as it is not counted the bill silently grows. In eleven lessons the data was split, moved, balanced, stored and recovered; in twenty-seven lessons it was ingested, transformed, stored and queued. Where this data goes was never asked. The **very model it feeds**, whether **the same model can be reproduced** with the same data, and **how a trained model goes to production** were not measured in any lesson. The next course takes the output of the pipeline as input and counts the reproducibility, the versioning and the deployment of a training run.