--- title: 'Parallel and Distributed Dataframes' source: 'https://academia.sh/en/courses/data-analysis-with-python/parallel-and-distributed-dataframes' course: 'Data Analysis with Python' language: en updated: '2026-10-07T04:31:11+00:00' license: 'CC BY-SA 4.0' objectives: - 'Counts the items of a distributed computation separately: the work units of the largest partition, the rows shuffled and the coordination per partition' - 'Compares three keys to show that the partitioning key determines both the partition balance and the number of rows shuffled' - 'Computes the dataset size at which distribution turns into gain over a single machine and shows how the threshold shifts with the key' - 'Measures how a query that can be partially aggregated and a query that needs integrity separate in the shuffle item' --- # Parallel and Distributed Dataframes When the 3,199-row dataset of the measurement network is split into partitions, the cost of distribution itself appears and stays larger than the work of a single machine: a five-partition distribution by the region key costs 11,651 units and a four-partition distribution by the subscriber key costs 8,921 units, while a single machine spends 3,199 units. The cost has three parts — the work units of the largest partition, the rows shuffled and the coordination per partition — and the point at which it turns into gain moves with the partitioning key: 10,833 rows with a balanced key, 20,119 rows with a skewed key. A query that is not aligned never sees this threshold; in the period partitioning 66.7% of the rows do not stay in place and the shuffle item alone rises to 6.7 times the work of a single machine. The previous lesson cut the values read by 6.8 times, but the whole computation was done in a single place. The limit of that place is unavoidable: even if the values read fall to one sixth, when the dataset grows a thousandfold the values a single machine would read still do not fit. This lesson distributes the work over several partitions and asks a single question — what does distribution itself add to the computation. The answer is a list of items, and the name of the list is **the cost of the transition**. Splitting the dataset into partitions, shuffling rows between partitions and coordinating the partitions are jobs of their own, separate from the work itself. The measure of the lesson counts these items in the same unit as the work of a single machine and puts two numbers side by side: the **gain**, that is, how many times cheaper distribution brought the work, and the **hidden default**, that is, the balance and the shuffle that are decided silently when the partitioning key is chosen. - **SP16.** The dataset is fictional and is the dataset of the previous lessons: 3,199 rows, 1,329 subscribers, three periods. It is produced with the same seed and the output is identical from run to run. **It is a model.** - **SP17.** The work unit is touching a row once; work on n rows is n units on a single machine. - **SP18.** Shuffling a row between partitions costs 10 units. This ratio is an assumption of the model; when it changes all the thresholds shift, and the direction of the shift is read in the table. **It is a model.** - **SP19.** Setting up a partition and collecting its result requires 2,000 units of coordination; p partitions add p times 2,000 units. - **SP20.** Partitions run concurrently; the critical path of the distributed job is the work units of **the largest partition**. The total cost is the sum of the critical path and the shuffle and coordination items. ## Three Partitionings of the Same Dataset ```python # partition.py — the same fictional dataset is split into partitions with three separate keys. MODEL: # in place of a real distribution, the partitioning, shuffle and coordination items are counted. # The unit of measure is touching a row once. SEED = 1246 REGION = ("north", "east", "center", "south", "west") PERIODS = (("2024-01", 1329), ("2024-02", 1035), ("2024-03", 835)) DENSITY = (0, 1, 2, 2, 2, 3, 4, 2, 2, 1) # center-weighted measurement network SHUFFLE = 10 # shuffling one row to another partition costs ten local touches COORDINATION = 2000 # setting up a partition and collecting its result def rng(seed: int): s = seed & 0xFFFFFFFF def next_value() -> float: nonlocal s s = (1103515245 * s + 12345) & 0xFFFFFFFF return s / 4294967296 return next_value def records() -> list[tuple]: r, out = rng(SEED), [] for period, size in PERIODS: for i in range(size): out.append((10000 + i, f"S-{4000 + i}", period, REGION[DENSITY[i % 10]], round(5 + r() * 32.0, 1))) return out def partition_by(rows: list, key) -> dict: parts = {} for x in rows: parts.setdefault(key(x), []).append(x) return parts r0 = records() KEYS = (("region", lambda x: x[3]), ("subscriber mod 4", lambda x: x[0] % 4), ("period", lambda x: x[2])) G = (18, 12, 30, 14, 12) fmt = lambda h: "".join(str(v).rjust(G[j]) if j else str(v).ljust(G[0]) for j, v in enumerate(h)) print(f"dataset {len(r0)} rows, {len({x[0] for x in r0})} subscribers; shuffle {SHUFFLE} " f"units/row, coordination {COORDINATION} units/partition") print(fmt(["partition key", "partitions", "partition sizes", "largest", "skew"])) for name, f in KEYS: p = partition_by(r0, f) size = sorted((len(v) for v in p.values()), reverse=True) print(fmt([name, len(p), str(size), size[0], f"{size[0] / (len(r0) / len(p)):.2f}x"])) ``` ``` dataset 3199 rows, 1329 subscribers; shuffle 10 units/row, coordination 2000 units/partition partition key partitions partition sizes largest skew region 5 [1601, 639, 321, 319, 319] 1601 2.50x subscriber mod 4 4 [801, 800, 800, 798] 801 1.00x period 3 [1329, 1035, 835] 1329 1.25x ``` All three keys are defensible and the three give a different layout. Partitioning by the remainder of the subscriber number is balanced: four partitions between 798 and 801 rows, skew 1.00. Partitioning by region is skewed: the center partition carries 1,601 rows, 2.50 times the average. The period key stands in between. Skew is written directly onto the critical path. Because partitions run concurrently, the job finishes when the largest partition finishes; a dataset split into five partitions falls not to one fifth but to one half if one of the partitions carries half of the dataset. The partitioning key looks like a detail of arrangement, yet it sets the upper limit of the gain expected from distribution. ## The Items of the Transition ```python # transition.py — two queries are run over three partitionings and the items of the # transition are counted separately: the critical path (work units of the largest partition), # the rows shuffled and the coordination. Total = critical path + shuffled * SHUFFLE + partitions * COORDINATION. def build_index(p: dict) -> dict: return {name: i for i, name in enumerate(sorted(p, key=str))} def is_aligned(p: dict, identity) -> bool: where = {} for name, v in p.items(): for x in v: where.setdefault(identity(x), set()).add(name) return all(len(s) == 1 for s in where.values()) def redistribute(p: dict, identity) -> tuple[dict, int]: """The dataset is re-partitioned by the identity key and the number of partitions is kept. Every row whose target partition differs from its source partition is shuffled.""" ix, n_parts = build_index(p), len(p) new, shuffled = {i: [] for i in range(n_parts)}, 0 for name, v in p.items(): for x in v: h = identity(x) % n_parts new[h].append(x) shuffled += h != ix[name] return new, shuffled cost = lambda p, critical, shuffled: critical + shuffled * SHUFFLE + p * COORDINATION G2 = (18, 12, 10, 15, 14, 13, 9) fmt2 = lambda h: "".join(str(v).rjust(G2[j]) if j else str(v).ljust(G2[0]) for j, v in enumerate(h)) print("Query A: total reading per region (can be partially aggregated)") print(fmt2(["partitioning", "partitions", "shuffled", "critical path", "coordination", "distributed", "ratio"])) for name, f in KEYS: p = partition_by(r0, f) shuffled = sum(len({x[3] for x in v}) for v in p.values()) # partial results critical = max(len(v) for v in p.values()) m = cost(len(p), critical, shuffled) print(fmt2([name, len(p), shuffled, critical, len(p) * COORDINATION, m, f"{len(r0) / m:.2f}x"])) print("Query B: number of subscribers whose increase across periods exceeds 5 m3 (needs subscriber integrity)") print(fmt2(["partitioning", "partitions", "shuffled", "critical path", "coordination", "distributed", "ratio"])) for name, f in KEYS: p = partition_by(r0, f) if is_aligned(p, lambda x: x[0]): shuffled, critical = len(p), max(len(v) for v in p.values()) else: new, shuffled = redistribute(p, lambda x: x[0]) shuffled, critical = shuffled + len(p), max(len(v) for v in new.values()) m = cost(len(p), critical, shuffled) print(fmt2([name, len(p), shuffled, critical, len(p) * COORDINATION, m, f"{len(r0) / m:.2f}x"])) ``` ``` Query A: total reading per region (can be partially aggregated) partitioning partitions shuffled critical path coordination distributed ratio region 5 5 1601 10000 11651 0.27x subscriber mod 4 4 12 801 8000 8921 0.36x period 3 15 1329 6000 7479 0.43x Query B: number of subscribers whose increase across periods exceeds 5 m3 (needs subscriber integrity) partitioning partitions shuffled critical path coordination distributed ratio region 5 5 1601 10000 11651 0.27x subscriber mod 4 4 4 801 8000 8841 0.36x period 3 2136 1067 6000 28427 0.11x ``` Both tables run on the same dataset with the same number of partitions, and in both the distributed cost stays above the 3,199 units of the single machine. At this scale distribution does not pay; the job of the table is to show where the loss comes from. Query A can be partially aggregated: each partition computes its own region totals and sends only the partial result. The rows shuffled are between 5 and 15 and are **independent of the dataset size** — they are the product of the number of partitions and the number of groups. Almost all of the cost is coordination: 10,000 of the 11,651 units with the region key, 8,000 of the 8,921 units with the subscriber key. Query B wants all the rows of a subscriber to be in the same partition, and here the partitioning key decides the result. The region and subscriber keys are aligned — every subscriber is in a single region — and the rows shuffled stay as many as the number of partitions. The period key is not aligned: the three periods of the same subscriber are in three separate partitions, the dataset is redistributed and 2,136 rows are shuffled. The cost rises from 7,479 to 28,427 units, four times. **This is the hidden default:** the partitioning key is chosen before the query, and that choice determines the shuffle item of the query. ## The Threshold Where It Turns into Gain ```python # scale.py — the same model is solved at growing dataset sizes. The partition ratios are # taken from the measured dataset; in the query that can be partially aggregated the rows shuffled # are independent of n, in the unaligned query they grow with n. import math RATIO = {name: max(len(v) for v in partition_by(r0, f).values()) / len(r0) for name, f in KEYS} _p = partition_by(r0, dict(KEYS)["period"]) _new, _moved = redistribute(_p, lambda x: x[0]) SHARE = _moved / len(r0) # the share that does not stay in place in the period partitioning B_RATIO = max(len(v) for v in _new.values()) / len(r0) SCENARIOS = (("A / subscriber mod 4", 4, RATIO["subscriber mod 4"], 12, 0.0), ("A / region", 5, RATIO["region"], 5, 0.0), ("B / period", 3, B_RATIO, 3, SHARE)) G3 = (14, 16, 20, 12, 12, 12) fmt3 = lambda h: "".join(str(v).rjust(G3[j]) if j else str(v).ljust(G3[0]) for j, v in enumerate(h)) def distributed(n, p, ratio, fixed, share): shuffled = fixed + share * n return round(ratio * n + shuffled * SHUFFLE + p * COORDINATION) print(fmt3(["dataset n", "single machine", "A/subscriber mod 4", "A/region", "B/period", "best ratio"])) for n in (3199, 31990, 319900, 3199000): line = [distributed(n, p, o, s, share) for _, p, o, s, share in SCENARIOS] print(fmt3([n, n, line[0], line[1], line[2], f"{n / min(line):.2f}x"])) print(fmt3(["threshold n", "-", *[ (math.ceil((s * SHUFFLE + p * COORDINATION) / (1 - o)) if share * SHUFFLE + o < 1 else "never") for _, p, o, s, share in SCENARIOS], "-"])) print(f"in the period partitioning {100 * SHARE:.1f} percent of the rows do not stay in place; " f"the shuffle item alone is {SHARE * SHUFFLE:.1f} times the work of a single machine") ``` ``` dataset n single machine A/subscriber mod 4 A/region B/period best ratio 3199 3199 8921 11651 28427 0.36x 31990 31990 16130 26060 230000 1.98x 319900 319900 88220 170150 2245730 3.63x 3199000 3199000 809120 1611050 22403030 3.95x threshold n - 10833 20119 never - in the period partitioning 66.7 percent of the rows do not stay in place; the shuffle item alone is 6.7 times the work of a single machine ``` The table shows where distribution turns into gain. Coordination is a fixed item independent of the dataset size; as the dataset grows its share melts away and the critical path becomes dominant. With the balanced key the threshold is 10,833 rows: below this size distributing is expensive, above it it becomes cheap. With the skewed key the threshold rises to 20,119 rows — the same model, the same constants, only the partitioning key is different. Skew nearly doubles the threshold. The third column never sees the threshold. In the unaligned query the rows shuffled grow with the dataset size: 66.7% of the rows do not stay in place and the shuffle item alone rises to 6.7 times the work of a single machine. However large n grows the ratio stays constant and the distributed cost stays above the single machine. Scaling up does not save this query; the only thing that saves it is partitioning the dataset by the right key. The gain side is limited as well. At 3,199,000 rows the best option costs 809,120 units and gains 3.95 times over the single machine — a little under the four times expected from a job split into four partitions. The difference comes entirely from the shuffle and coordination items, and these items are not zeroed at any dataset size. **Increasing the number of partitions does not increase the gain as many times as the number of partitions; the coordination item grows with the number of partitions.** ## The Limit of the Partition Count ```python # partition_count.py — the same job is distributed with different partition counts. The critical path # shrinks with the partition count, the coordination grows with it; the best partition count # is where the two balance. A balanced key is assumed. N = 3199000 G4 = (12, 15, 12, 15, 14, 10) fmt4 = lambda h: "".join(str(v).rjust(G4[j]) if j else str(v).ljust(G4[0]) for j, v in enumerate(h)) total = lambda p: math.ceil(N / p) + (p * 5) * SHUFFLE + p * COORDINATION print(fmt4(["partitions", "critical path", "shuffled", "coordination", "distributed", "ratio"])) for p in (2, 4, 16, 40, 64, 256): print(fmt4([p, math.ceil(N / p), p * 5, p * COORDINATION, total(p), f"{N / total(p):.1f}x"])) best = min(range(1, 1001), key=total) print(f"for n = {N} the best partition count is {best}; cost {total(best)} units, " f"gain {N / total(best):.1f} times") print(f"if {4 * best} is chosen instead of {best} partitions the cost rises to {total(4 * best)} " f"units and the gain falls to {N / total(4 * best):.1f} times") ``` ``` partitions critical path shuffled coordination distributed ratio 2 1599500 10 4000 1603600 2.0x 4 799750 20 8000 807950 4.0x 16 199938 80 32000 232738 13.7x 40 79975 200 80000 161975 19.7x 64 49985 320 128000 181185 17.7x 256 12497 1280 512000 537297 6.0x for n = 3199000 the best partition count is 40; cost 161975 units, gain 19.7 times if 160 is chosen instead of 40 partitions the cost rises to 347994 units and the gain falls to 9.2 times ``` The partition count moves two items in opposite directions. The critical path is divided by the partition count and the coordination is multiplied by it; while the first falls the second rises, and the total takes its smallest value somewhere. For a dataset of 3,199,000 rows that place is 40 partitions: cost 161,975 units, gain 19.7 times. With four partitions the gain is 4.0 times and with 256 partitions it is 6.0 times — past a certain point increasing the partition count lowers the gain. The number itself depends on the constants of the model, and the constants are written in SP18 and SP19; if the shuffle cost or the coordination changes, the best partition count changes too. What does not change is the shape: the gain of distribution is not proportional to the partition count and it has a peak. When four times the best partition count is chosen the cost rises to 347,994 units and the gain falls to 9.2 times: four times as many partitions means twice the cost. The partition count too is not a default but a decision that has to be written down. ## Summary - The cost of a distributed computation is three items: the work units of the largest partition, the rows shuffled times the shuffle cost, and the coordination per partition. The partitioning key determines the partition balance: the subscriber remainder gives 1.00 skew and a critical path of 801 rows, the region key gives 2.50 skew and a critical path of 1,601 rows. - At 3,199 rows distributing does not pay; the best option costs 8,921 units while a single machine spends 3,199 units. - In the query that can be partially aggregated the rows shuffled are independent of the dataset size (5–15 rows); in the query that needs integrity an unaligned key shuffles 2,136 rows and raises the cost fourfold. - The threshold where it turns into gain is 10,833 rows with the balanced key and 20,119 rows with the skewed key; with the unaligned query there is no such threshold. - At 3,199,000 rows a four-partition distribution gains 3.95 times; on the same dataset the best partition count is 40 and the gain saturates at 19.7 times, falling back to 6.0 times at 256 partitions. ## Next Step This lesson moved the work to several partitions, but all the items were still built on **bringing** the data to the analysis environment: rows are read, shuffled, split. If the place where the data sits can also compute, the order can be reversed — the filter and the aggregation are pushed to the source and only the result comes to the analysis environment. The next lesson measures this with a single number: how many rows leave the source for the same question, and which decision is no longer in your hands once the aggregation is done at the source.