Cost and Cluster Sizing

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.

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

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

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.