Cluster Computing Fundamentals

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.

# 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.

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.

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.

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.