Distributed Datasets

In the previous three lessons the job was split and run: rows were distributed to workers by a key, the place of the blocks determined where the read would go, and the map was separated from the reduce. The three shared an assumption that was written nowhere: every computation the driver writes runs. The moment a transformation is defined, the cluster executes it, writes its result somewhere and waits for the next one.

This lesson removes that assumption and asks a single question: how much of the computation written is really used in the result. The answer is given with two numbers: the partitions never opened and the rows never read. The structure that makes these two numbers possible is the distributed dataset: it consists of partitions and a plan that has not run yet.

• DC41. The source, the pipeline and the cluster are fictional; no real distributed processing framework or resource manager is called. The seed is 20260218, and the measurement does not depend on the run.
• DC42. Makespan is the model’s own unit, not raw milliseconds: the number of rows read by the busiest worker plus the task overhead.
• DC43. The partition is the unit of data, the worker is the unit of computation. The partition key is period; because the consumption stage drops the first period, 11 partitions remain, each with 200 rows.
• DC44. The partitions are handed to the workers in turn (0, 1, 2, 3, 0, …); three workers hold three partitions each, one holds two.
• DC45. A transformation defines a new dataset and returns no result; an action returns a result to the driver and is the only thing that starts the computation.
• DC46. The units counted are: opened partitions, rows read, intermediate rows written, makespan. There is no shuffle in this lesson, so the total work equals the rows read.
• DC47. The baseline is eager evaluation: every transformation runs the moment it is defined and writes its intermediate result.
• DC48. Resolution: 11 partitions, 200 rows per partition, 2200 in total. No ratio is printed on the presentation; the value itself is written.

Transformation and Action

There are two kinds of calls in a distributed cluster, and the difference between them is the whole of this lesson. A transformation defines a new dataset from an existing one: filtering, mapping, repartitioning. What it returns is not data but a recipe. An action returns a result to the driver: counting, fetching the first rows, summing. Only an action starts the computation.

Lazy evaluation is the consequence of this distinction. The transformations are written into a plan, and no partition is opened until the first action arrives. When the action arrives the plan is known as a whole, and the cluster runs only the partitions that the result requires. Its opposite is eager evaluation: every transformation runs on the line where it is written, writes its intermediate result, and the next transformation reads it again.

The first block builds the fictional source and the first three stages of the pipeline. The 2200 rows coming out of it are the only concrete axis of the course.

# FICTIONAL. The meter readings, the first three stages of the pipeline and the fictional cluster are
# defined in the lesson; no real distributed processing framework or resource manager is called.
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 s1_screen(records):
    REQUIRED = ("subscriber", "period", "region", "index", "unit")
    keep = [k for k in records if all(a in k for a in REQUIRED)]
    return keep, len(records) - len(keep)


def s2_dedupe(records):
    seen, keep = set(), []
    for k in records:
        key = (k["subscriber"], k["period"])
        if key in seen:
            continue
        seen.add(key)
        keep.append(k)
    return keep, len(records) - len(keep)


def s3_consumption(records):
    table = {(k["subscriber"], k["period"]): k for k in records}
    out, dropped = [], 0
    for k in records:
        i = PERIODS.index(k["period"])
        if i == 0:
            continue
        prev = table.get((k["subscriber"], PERIODS[i - 1]))
        if prev is None:
            dropped += 1
            continue
        try:
            f = int(k["index"]) - int(prev["index"])
        except (TypeError, ValueError):
            dropped += 1
            continue
        out.append(dict(k, m3=f))
    return out, dropped


WORKERS = 4
TASK_OVERHEAD = 50       # the fixed cost of handing work to each worker , a choice


def rows():
    v, _ = s1_screen(generate_source())
    v, _ = s2_dedupe(v)
    v, _ = s3_consumption(v)
    return v

The Partition Is the Unit of Data, the Worker Is the Unit of Computation

The number of partitions and the number of workers are not the same thing, and the gain this lesson measures arises precisely from this distinction. The partition is a divided piece of the data and can be read on its own. The worker is the computing unit that reads those pieces, and it can hold more than one partition. Here 11 partitions are spread over four workers: three workers take three partitions each, one takes two. Because there are more partitions than workers, an action can open only some of the partitions, and the rows of a partition that is not opened are never read.

How a partition is used on the storage side, that is, deciding by the file layout which partition will never be read, was measured in the Data Storage Architectures course and is not repeated here. The elimination here is on the computation side: what eliminates the partition is not the file layout but what the action asks for.

S = rows()
PARTITIONS = {d: [k for k in S if k["period"] == d] for d in PERIODS}
PARTITIONS = {d: v for d, v in PARTITIONS.items() if v}
NAMES = sorted(PARTITIONS)
OWNER = {a: i % WORKERS for i, a in enumerate(NAMES)}
COUNTER = {}


def reset():
    COUNTER.update({"partitions": 0, "read": 0, "intermediate": 0,
                    "worker": {i: 0 for i in range(WORKERS)}})


def read_partition(name, data):
    COUNTER["partitions"] += 1
    COUNTER["read"] += len(data)
    COUNTER["worker"][OWNER[name]] += len(data)
    return data


class Lazy:
    """A transformation is written into the plan , it does not run. The computation starts only with an action."""

    def __init__(self, partitions, plan=()):
        self.partitions, self.plan = partitions, tuple(plan)

    def select(self, f):
        return Lazy(self.partitions, self.plan + (("select", f),))

    def map_rows(self, f):
        return Lazy(self.partitions, self.plan + (("map", f),))

    def _run(self, name):
        out = []
        for k in read_partition(name, self.partitions[name]):
            for kind, f in self.plan:
                if kind == "select" and not f(k):
                    k = None
                    break
                if kind == "map":
                    k = f(k)
            if k is not None:
                out.append(k)
        return out

    def count(self):
        return sum(len(self._run(a)) for a in NAMES)

    def first(self, n):
        c = []
        for a in NAMES:
            c += self._run(a)
            if len(c) >= n:
                break
        return c[:n]


class Eager(Lazy):
    """Every transformation runs the moment it is defined and writes its intermediate result."""

    def _apply(self, f, select):
        new = {}
        for a in NAMES:
            v = [k for k in read_partition(a, self.partitions[a]) if f(k)] if select \
                else [f(k) for k in read_partition(a, self.partitions[a])]
            COUNTER["intermediate"] += len(v)
            new[a] = v
        return Eager(new)

    def select(self, f):
        return self._apply(f, True)

    def map_rows(self, f):
        return self._apply(f, False)

    def _run(self, name):
        return self.partitions[name]


print("partitions:", len(PARTITIONS), " rows per partition:", sorted(set(len(v) for v in PARTITIONS.values())))
print("partitions per worker:", [sum(1 for a in NAMES if OWNER[a] == i) for i in range(WORKERS)])

partitions: 11  rows per partition: [200]
partitions per worker: [3, 3, 3, 2]

Both classes offer the same interface; the difference is only when the computation runs. In the Lazy class select returns a new plan; in the Eager class the same call reads all the partitions.

The Computation Laziness Eliminates

The job measured is an ordinary script written by the driver: the readings of the residential subscribers are filtered, the row is reduced to four fields, and those above 30 m3 are selected. The script also defines a commercial dataset and no action asks for it; this is the measure of the computation held in hand. Four setups run the same script and the action changes: first(5) asks for five rows, first(100) for a hundred rows, count for the whole result.

FIELDS = ("subscriber", "period", "region", "m3")


def script(K):
    """The job the driver writes: the readings of residential subscribers above 30 m3."""
    residential = K.select(lambda k: k["type"] == "residential")
    K.select(lambda k: k["type"] == "commercial")          # no action asks for this
    return residential.map_rows(lambda k: {a: k[a] for a in FIELDS}).select(lambda k: k["m3"] > 30)


def measure(mode, action):
    reset()
    y = script(Lazy(PARTITIONS) if mode == "lazy" else Eager(PARTITIONS))
    n = y.count() if action == "count" else len(y.first(int(action)))
    return (COUNTER["partitions"], COUNTER["read"], COUNTER["intermediate"],
            max(COUNTER["worker"].values()) + TASK_OVERHEAD, n)


print(f"{'mode':<9}{'action':>8}{'opened partitions':>19}{'read':>6}"
      f"{'intermediate rows':>19}{'makespan':>10}{'result':>8}")
for mode, action in (("eager", "5"), ("lazy", "5"), ("lazy", "100"),
                     ("eager", "count"), ("lazy", "count")):
    b, o, mid, makespan, n = measure(mode, action)
    print(f"{mode:<9}{action:>8}{b:>19}{o:>6}{mid:>19}{makespan:>10}{n:>8}")

mode       action  opened partitions  read  intermediate rows  makespan  result
eager           5                 44  6556               3727      1838       5
lazy            5                  1   200                  0       250       5
lazy          100                  3   600                  0       250     100
eager       count                 44  6556               3727      1838     449
lazy        count                 11  2200                  0       650     449

The first two rows are the finding of the lesson, and the result column is the same in both. The eager setup opens 44 partitions for a result of five rows, reads 6556 rows and writes 3727 intermediate rows; the makespan is 1838. The lazy setup gives the same five rows by opening a single partition, reading 200 rows and writing no intermediate rows; the makespan is 250. The remaining 10 partitions are never opened and 2000 rows are never read.

The 6556 rows of the eager setup come from two places. The first: the 2200 rows of the source are read twice, because the residential and commercial filters pass separately over all the partitions. The second: the commercial branch reads 2200 rows and writes 1122 intermediate rows and never enters the result. In the lazy setup that branch is nothing but a plan, and a plan does not run.

The third row carries the rule of the course on its own. first(100) opens three partitions and the rows read rise from 200 to 600, so the total work triples, but the makespan stays at 250. The reason is DC44: because of the sequential distribution the first three partitions are on three separate workers and all three run at the same time. The busiest worker still reads 200 rows. A speedup claim that does not state which one was measured falls short here: between the same two setups, the total work is three times and the makespan one time.

The last two rows draw the limit of laziness. Because count asks for the whole result, the lazy setup too opens all 11 partitions and reads 2200 rows. The difference no longer comes from the skipped partitions but from the unused branch and the 3727 intermediate rows not written: 2200 rows against 6556, a makespan of 650 against 1838.

Partition Size Is the Resolution of Laziness

Laziness can skip a partition; it cannot skip a row from inside a partition. So the smallest piece that can be skipped is the partition itself, and the number of partitions is a decision. The same 2200 rows are put into four, eleven and forty-four partitions; the measured job and the action do not change.

SAVED = (PARTITIONS, NAMES, OWNER)


def repartition(parts):
    """The number of partitions is a decision: the smallest unit laziness can skip is one partition."""
    global PARTITIONS, NAMES, OWNER
    PARTITIONS = {i: S[i * len(S) // parts:(i + 1) * len(S) // parts] for i in range(parts)}
    NAMES, OWNER = sorted(PARTITIONS), {i: i % WORKERS for i in range(parts)}


print(f"{'partitions':>12}{'rows/partition':>16}{'first(5) opened':>17}{'first(5) read':>15}"
      f"{'first(5) makespan':>19}{'count makespan':>16}")
for parts in (4, 11, 44):
    repartition(parts)
    b, o, _, makespan, _ = measure("lazy", "5")
    _, _, _, full, _ = measure("lazy", "count")
    print(f"{parts:>12}{len(S) // parts:>16}{b:>17}{o:>15}{makespan:>19}{full:>16}")
PARTITIONS, NAMES, OWNER = SAVED

  partitions  rows/partition  first(5) opened  first(5) read  first(5) makespan  count makespan
           4             550                1            550                600             600
          11             200                1            200                250             650
          44              50                1             50                100             600

The opened partitions are one in all three setups; what changes is how many rows that partition carries. For a result of five rows the rows read fall from 550 to 50 and the makespan from 600 to 100. The gain of laziness is measured by the partition size: in a coarsely divided dataset laziness is written but finds nothing to eliminate.

The last column shows the opposite direction. In the count action, which asks for the whole result, 11 partitions give a makespan of 650, four and forty-four partitions give 600. The reason is that the number of partitions is not divisible by the number of workers: because 11 partitions spread over four workers as [3, 3, 3, 2], the busiest worker reads 600 rows while one worker is left idle at 400 rows. Thinning the partitions also has an invisible cost: in this model TASK_OVERHEAD is added to the makespan only once, whereas forty-four partitions mean handing out work forty-four times.

The Error Laziness Defers

That the plan does not run is a delay as much as a gain. When the driver writes a chain of ten transformations, no row is read, so whether the chain is correct is not tested either. The plan below tries to divide a number by a text, and nothing happens at the place where it is defined.

reset()
broken = Lazy(PARTITIONS).map_rows(lambda k: dict(k, ratio=k["m3"] / k["unit"]))
print("plan built, rows read:", COUNTER["read"])
try:
    broken.first(1)
except TypeError as e:
    print("when the action is called:", type(e).__name__, "-", e)
print("rows read until the error:", COUNTER["read"])

plan built, rows read: 0
when the action is called: TypeError - unsupported operand type(s) for /: 'int' and 'str'
rows read until the error: 200

While the plan is being built the rows read are zero, that is, the error does not show while the plan is written. It shows at the first action, and by then 200 rows have been read. In the eager setup the same error would have fallen on the line where the map_rows call is written. The elimination gain and the error delay come from the same mechanism: the postponement of the computation. The same mechanism has a second cost as well: because a plan is not a result, when two actions are called on the same dataset the chain runs twice; the measurement of this cost belongs to the second topic of the course.

Summary

• A distributed dataset consists of partitions and a plan that has not run yet; a transformation is written into the plan, only an action starts the computation, and the rows read while the plan is being built are zero.
• The partition is the unit of data, the worker is the unit of computation: 11 partitions spread over four workers as [3, 3, 3, 2], and an action can open only some of the partitions.
• In a result of five rows the eager setup reads 6556 rows by opening 44 partitions, while the lazy setup reads 200 rows by opening a single partition; 10 partitions are never opened, 2000 rows are never read.
• first(100) raises the rows read from 200 to 600 but the makespan stays at 250; in the count action, which asks for the whole result, the gain comes from the unused branch and the 3727 intermediate rows not written (2200 rows against 6556, a makespan of 650 against 1838).
• The smallest unit laziness can skip is a partition: the same first(5) reads 550 rows in a four-partition dataset and 50 rows in a forty-four-partition one; on the other hand, when the number of partitions is not divisible by the number of workers, the full-scan makespan rises from 600 to 650.
• The type error in the plan appears only at the first action and after 200 rows have been read; the elimination gain and the error delay come from the same mechanism.

Next Step

This lesson decided which computation will run, but it never touched who will run that computation. The partitions were handed to the workers in turn, the number of workers was taken as four, and TASK_OVERHEAD was quietly added in the makespan calculation. In a real cluster none of these three is in the hands of the driver: how many workers to split the job into, how many of those workers will run at the same time, and when a waiting job takes its turn are decided by the resource manager. The next lesson measures that decision and separates two knobs: increasing the number of tasks and increasing the number of concurrent tasks are not the same thing, because every task has a fixed cost of 50 units and this cost grows together with the number of tasks. The number to be measured is the task count at which the makespan gain loses to this cost.