--- title: 'Shuffle Cost' source: 'https://academia.sh/en/courses/distributed-data-processing/shuffle-cost' course: 'Distributed Data Processing' language: en updated: '2026-10-21T05:03:00+00:00' license: 'CC BY-SA 4.0' objectives: - 'Defines the shuffle as shuffled rows and measures the shuffled and non-shuffled setups of the same job separately as makespan and total work.' - 'Counts, over three keys, which rows the partition key leaves in place.' - 'Sweeps the network cost over four values to separate the results that are sensitive to this choice from those that are insensitive.' --- # 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.