--- title: 'Data Skew' source: 'https://academia.sh/en/courses/distributed-data-processing/data-skew' course: 'Distributed Data Processing' language: en updated: '2026-10-21T05:03:00+00:00' license: 'CC BY-SA 4.0' objectives: - 'Defines data skew as the ratio of the busiest worker to the average and computes it for three partition keys.' - 'Counts how increasing the number of workers on a wrong partition key enlarges the skew and the total work.' - 'Shows that the partitioning does not change the result at all and justifies why a wrong partitioning can be seen only by counting the cost.' --- # 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.