--- title: 'Cluster Computing Fundamentals' source: 'https://academia.sh/en/courses/distributed-data-processing/cluster-computing-fundamentals' course: 'Distributed Data Processing' language: en updated: '2026-10-21T05:03:00+00:00' license: 'CC BY-SA 4.0' objectives: - 'Defines the driver, executor, worker and task abstractions of a distributed run on a fictional cluster.' - 'Computes the makespan and the total work of a job separately and shows that they are not the same number.' - 'Measures what the partition key does to worker loads and to skew.' --- # 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.