Distributed Analysis Engines

The previous lesson wrote the three functions and counted the records moved in the shuffle, but it did not ask who runs the functions. When a data frame chain is written on a real cluster, the person who writes the chain orders the operations; the engine splits that ordering into stages, defines the partitions for each stage, distributes the tasks, retries a failed task, and collects the results.

None of this work processes a row and none of it is written in the chain. The person who writes the chain thinks they have written a query; the engine derives a task schedule from that query, and the length of the schedule comes not from the length of the query but from the points at which the query has to redistribute the data. This is the measure of the lesson: the coordination overhead item — countable items that are added per job and do not grow with job size. The second number next to it is how many rows fall to one coordination overhead item. Distributed systems theory, partitioning, and coordination were established in the System Design and Distributed Systems and the DevOps and Platform Engineering curricula and are not repeated here; the only thing written is the item that is counted.

• AS13. No distributed processing engine is installed and no product name is written. The chain, the stage, and the item are modeled with the standard library. It is a model.
• AS14. A transformation is narrow if it ends within a partition (filter, select, derive), or wide if it needs a shuffle (group, sort, join). The stage boundary is drawn at a wide transformation.
• AS15. Coordination overhead items are of four kinds: partition definition, task dispatch, task result, and stage collection. The first three are counted as many as the number of partitions per stage, the last one once per stage.
• AS16. The probability that a task fails is an assumption and is tested with two values: 0.02 and 0.15. At most three tries are made; each retry adds two items. The generator is written in-house and the seed is 20260218.
• AS17. The number of partitions is derived from the job size: part size 400,000 rows, at least four partitions. Both are decisions.
• AS18. The acceptability criterion is a decision: at least 10,000 rows must fall to one coordination overhead item.

How Many Stages the Chain Splits Into

# chain.py — a data frame chain is split into stages. MODEL: no distributed engine is
# installed; the operations of the chain are marked narrow (ends within a partition) and
# wide (needs a shuffle) and the stage boundary is drawn at a wide transformation.
CHAIN = [("filter: period 1-3", "narrow"), ("derive: tariff tier", "narrow"),
         ("group: subscriber", "wide"), ("derive: subscriber average", "narrow"),
         ("filter: average > 10 m3", "narrow"), ("group: region", "wide"),
         ("sort: average", "wide"), ("first 3", "narrow")]
PARTITIONS = 8


def to_stages(chain):
    stages, current = [], []
    for label, kind in chain:
        current.append((label, kind))
        if kind == "wide":
            stages.append(current)
            current = []
    if current:
        stages.append(current)
    return stages


A = to_stages(CHAIN)
print(f"{len(CHAIN)} operations, {sum(1 for _, t in CHAIN if t == 'wide')} wide, "
      f"{len(A)} stages, {PARTITIONS} partitions")
print(f"{'stage':<7}{'operations':>12}{'boundary':>10}  {'content'}")
for i, a in enumerate(A, 1):
    print(f"{i:<7}{len(a):>12}{a[-1][1]:>10}  {' -> '.join(label for label, _ in a)}")

# Coordination overhead item: in each stage, for each partition, a partition definition,
# a task dispatch, and a task result; for each stage, one stage collection item.
ITEMS = (("partition definition", "partition"), ("task dispatch", "partition"),
         ("task result", "partition"), ("stage collection", "stage"))
print(f"{'item kind':<22}{'scale':>10}{'stages':>7}{'partitions':>11}{'items':>7}")
total = 0
for label, scale in ITEMS:
    n = len(A) * PARTITIONS if scale == "partition" else len(A)
    total += n
    print(f"{label:<22}{scale:>10}{len(A):>7}{PARTITIONS if scale == 'partition' else '-':>11}{n:>7}")
print(f"{'total':<22}{'':>10}{'':>7}{'':>11}{total:>7}")

8 operations, 3 wide, 4 stages, 8 partitions
stage    operations  boundary  content
1                 3      wide  filter: period 1-3 -> derive: tariff tier -> group: subscriber
2                 3      wide  derive: subscriber average -> filter: average > 10 m3 -> group: region
3                 1      wide  sort: average
4                 1    narrow  first 3
item kind                  scale stages partitions  items
partition definition   partition      4          8     32
task dispatch          partition      4          8     32
task result            partition      4          8     32
stage collection           stage      4          -      4
total                                                 100

The person who wrote the chain wrote eight operations; the engine built four stages. The number of stages is not written in the chain; it derives from the three wide transformations and is invisible when the chain is read. That the three operations of the second stage sit together is also an engine decision: the two narrow transformations work on the output of the shuffle before them, so no separate stage is needed.

Whether a transformation counts as narrow or wide is not a given fact either; it is the result of the partitioning decision. The third operation of the chain is grouping by the subscriber key; if the dataset is partitioned by the subscriber key, all the rows of a subscriber are already in the same partition and the grouping needs no shuffle, that is, it is narrow. The same transformation is wide in a dataset partitioned by the period key. The four stages in the table rest on the assumption that the dataset is not partitioned by the subscriber key; when the partitioning key changes, the number of stages falls to three and the coordination overhead items fall by the same ratio. The number of stages is the number that the chain and the partitioning give together, not the number of the chain alone.

The lower table gives the measure of the lesson. Four stages and eight partitions amount to 100 coordination overhead items across four item kinds. Not one of these hundred items reads a row, adds a value, or builds a group. All of them are around the job, and their existence is written neither in the chain nor in the query. The item count is entirely independent of job size: the same chain asks for a hundred items at three thousand rows and at three billion rows.

The four item kinds also need to be told apart, because three scale with the number of partitions and one with the number of stages. The partition definition is the determination of which rows a partition is responsible for in a stage. The task dispatch is the handing of that definition to an executor. The task result is the executor’s reporting that it has finished and the recording of its output. These three are counted separately for each partition and multiplied by the number of partitions. The stage collection is seeing that all the partitions of the stage have finished and opening the next stage; there is one per stage and it is unaffected by the number of partitions. Ninety-six of the hundred items come from the first three kinds, four from the last.

Retry Is Not the Dominant Item

# retry.py — task failure and the retry item. The generator is written in-house and the seed
# is visible; the output is the same from run to run. MODEL: the probability that a task
# fails is an assumption and is tested with two values.
SEED, M32, MAX_TRIES = 20260218, 0xFFFFFFFF, 3


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


def item_count(stage, partition):
    return stage * partition * 3 + stage


def retries(stage, partition, q, seed):
    r, retried = rng(seed), 0
    for _ in range(stage * partition):
        for _ in range(MAX_TRIES - 1):
            if r() < q:
                retried += 1
            else:
                break
    return retried


print(f"at most {MAX_TRIES} tries; each retry adds 2 items "
      f"(re-dispatch and re-result)")
print(f"{'partitions':<12}{'tasks':>7}{'base items':>12}{'q=0.02 extra':>14}"
      f"{'q=0.02 rise':>13}{'q=0.15 extra':>14}{'q=0.15 rise':>13}")
for b in (4, 8, 16, 64, 256):
    g, base = len(A) * b, item_count(len(A), b)
    e1 = 2 * retries(len(A), b, 0.02, SEED + 91000 + b)
    e2 = 2 * retries(len(A), b, 0.15, SEED + 91000 + b)
    print(f"{b:<12}{g:>7}{base:>12}{e1:>14}{(base + e1) / base:>12.2f}x"
          f"{e2:>14}{(base + e2) / base:>12.2f}x")

at most 3 tries; each retry adds 2 items (re-dispatch and re-result)
partitions    tasks  base items  q=0.02 extra  q=0.02 rise  q=0.15 extra  q=0.15 rise
4                16          52             0        1.00x            12        1.23x
8                32         100             0        1.00x             8        1.08x
16               64         196             4        1.02x            24        1.12x
64              256         772             6        1.01x            82        1.11x
256            1024        3076            40        1.01x           336        1.11x

Retry is the most visible item of distributed processing, and the table shows it as the smallest item. At a 2% failure rate the item count rises by at most 2%; even at a 15% failure rate the rise stays between 8 and 23%. That the 1.23 times in the first row is higher than the others is not a trend but the fluctuation of a small count of sixteen tasks — the same ratio settles at 1.11 times at 1,024 tasks.

The dominant item is in the third column of the table. When the number of partitions rises from four to 256, the base item count grows from 52 to 3,076, nearly sixty times, and this growth comes not from failure but from the partitioning decision. Raising the number of partitions buys parallelism and, in the same motion, grows the coordination overhead item linearly. Retry is the noise of this table; the number that decides is the number of partitions.

It also needs to be written what the item is not. A coordination overhead item is not a unit of work; a code change that speeds up the computation does not lower the item count, because the items are around the computation, not inside it. Even if the work of a partition becomes ten times faster, that partition still counts one partition definition, one task dispatch, and one task result. There are three decisions that lower the item count and all three are taken in the chain itself: reducing the number of stages, reducing the number of partitions, and aligning the grouping key with the partitioning key.

Rows per Item

# scale.py — the item count is independent of job size, the row count is not.
# The measure: how many rows fall to one coordination overhead item. TARGET is a decision.
import math
TARGET, PART_SIZE, MIN_PARTITIONS = 10000, 400000, 4
CHAIN_B = [("filter: period 1-3", "narrow"), ("derive: tariff tier", "narrow"),
           ("group: region", "wide"), ("first 3", "narrow")]
B = to_stages(CHAIN_B)


def partition_count(n):
    return max(MIN_PARTITIONS, math.ceil(n / PART_SIZE))


def items(n, stage):
    b = partition_count(n)
    return item_count(stage, b) + 2 * retries(stage, b, 0.02,
                                              SEED + 91000 + b)


print(f"chain A {len(A)} stages, chain B {len(B)} stages; part size "
      f"{PART_SIZE} rows, at least {MIN_PARTITIONS} partitions, target {TARGET} rows/item")
print(f"{'job size':<14}{'partitions':>12}{'A items':>9}{'A rows/item':>15}"
      f"{'B items':>9}{'B rows/item':>15}")
for n in (3199, 31990, 319900, 3199000, 31990000, 319900000):
    ka, kb = items(n, len(A)), items(n, len(B))
    print(f"{n:<14}{partition_count(n):>12}{ka:>9}{n / ka:>15.1f}{kb:>9}"
          f"{n / kb:>15.1f}")
threshold = {}
for label, stage in (("A", len(A)), ("B", len(B))):
    n = 1000
    while n / items(n, stage) < TARGET and n < 10 ** 9:
        n += 1000
    threshold[label] = n
print(f"smallest job size that reaches the target: chain A {threshold['A']} rows, "
      f"chain B {threshold['B']} rows; halving the stages shrank the threshold "
      f"{threshold['A'] / threshold['B']:.1f} times")

chain A 4 stages, chain B 2 stages; part size 400000 rows, at least 4 partitions, target 10000 rows/item
job size        partitions  A items    A rows/item  B items    B rows/item
3199                     4       52           61.5       26          123.0
31990                    4       52          615.2       26         1230.4
319900                   4       52         6151.9       26        12303.8
3199000                  8      100        31990.0       50        63980.0
31990000                80      976        32776.6      490        65285.7
319900000              800     9730        32877.7     4866        65741.9
smallest job size that reaches the target: chain A 520000 rows, chain B 260000 rows; halving the stages shrank the threshold 2.0 times

The table shows when coordination becomes invisible. In the 3,199 rows of the fictional dataset, 61.5 rows fall to one coordination overhead item: the job is much smaller than the work that sets it up. In a dataset a thousand times larger 31,990 rows fall and the ratio saturates — the last two rows give 32,777 and 32,878, because the number of partitions also grows with the job size and the ceiling comes from the share of the part size that falls to each stage.

The saturation number is not made up; it derives from two decisions. The number of partitions is the job size divided by the part size, and the item count is stages times partitions times three; when the two cancel each other, the rows falling to one item settle at the part size divided by three times the number of stages. At four stages this is 400,000 over 12, that is, 33,333, and it appears in the table as 32,878 — the difference comes from the stage collection items and the retries. Without enlarging the part size or reducing the number of stages, this ceiling cannot be exceeded; beyond a certain point, enlarging the job size does not move the ratio at all.

When the acceptability criterion is written as 10,000 rows per item, the four-stage chain meets it at 520,000 rows and the two-stage chain at 260,000 rows. Halving the number of stages halves the threshold exactly. The number of stages looks like an engine detail, yet it is a decision given in the way the chain is written: every wide transformation draws a stage boundary and that boundary produces items for all the partitions.

It also needs to be written that the two chains do not ask the same question. Chain A first groups by the subscriber key, chain B directly by the region key; the difference measured in the previous lesson comes from here — the east average moves 0.140 m³ between 24.814 m³ and 24.674 m³. Reducing stages is not free; what it reduces is sometimes the question itself.

This lesson must not be confused with the threshold in the Scale and Performance topic of the Data Analysis with Python course. There the same question was asked in terms of the unit of work, and the size at which distribution turned into a gain was measured as 10,833 rows on a balanced key; what was measured was performance on the analyst’s own machine. Here what is measured is not the unit but the item, and the question asked is not when distribution becomes cheap but when coordination stays invisible. The two thresholds belong to different questions and cannot be written in place of each other. Distributing a job has to clear both thresholds: a job that does not clear the gain threshold slows down when distributed, and a job that does not clear the item threshold runs but is spent more on its surroundings than on the job itself.

Summary

• The three wide transformations of an eight-operation chain draw four stage boundaries; the number of stages is not written in the chain and is an engine decision.
• Four stages and eight partitions amount to 100 coordination overhead items across four item kinds; none of the items processes a row and their number is independent of job size.
• Retry is not the dominant item: at a 2% failure rate the rise is at most 2%, at a 15% failure rate it is between 8 and 23%.
• The dominant item is partitioning: when the partitions rise from four to 256, the base item count rises from 52 to 3,076.
• The rows that fall to one coordination overhead item are 61.5 in a 3,199-row job and 31,990 in a 3,199,000-row job, and the ratio saturates at 32,878.
• A target of 10,000 rows per item is met at 520,000 rows in the four-stage chain and at 260,000 rows in the two-stage chain; halving the stages halves the threshold but also changes the question asked.

Next Step

This lesson counted the item around the computation, but it never opened the data the computation reads. When a stage starts, the partitions read their shares from somewhere, and the layout of that place determines how many values will be read. The next lesson takes up the columnar file layout and ties the measure to the values read: how many values the row group headers and column boundaries let you skip, how many values the header itself makes you read, how many values the same query reads in three layouts, and in which direction the row group size moves this number.