Parallel and Distributed Dataframes

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

# 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

# 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

# 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

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