--- title: 'Distributed Datasets' source: 'https://academia.sh/en/courses/distributed-data-processing/distributed-datasets' course: 'Distributed Data Processing' language: en updated: '2026-10-21T05:03:00+00:00' license: 'CC BY-SA 4.0' objectives: - 'Separates a transformation from an action and defines a distributed dataset as consisting of partitions and an uncomputed plan.' - 'Counts the computation that lazy evaluation eliminates as opened partitions, rows read and intermediate rows written.' - 'Shows that in the same job the makespan and the total work can move in different directions, and that laziness defers the error to the moment of the action.' --- # Distributed Datasets A distributed dataset consisting of partitions and a plan that has not run yet, and counting the computation that lazy evaluation eliminates: while eleven partitions are spread over four workers as three, three, three and two, the eager setup opens 44 partitions, reads 6556 rows and writes 3727 intermediate rows for a result of five rows, the lazy setup gives the same result by opening a single partition and reading 200 rows, leaving 2000 rows never read; in a result of a hundred rows the rows read triple while the makespan stays at 250 because three partitions run on three separate workers; in the count action that asks for the whole result the lazy setup reads 2200 rows and the difference comes only from the unused branch and from the intermediate rows not written; because building the plan reads no rows, the type error in the plan appears only at the first action and after 200 rows have been read. 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. ```python # 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**. ```python 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. ```python 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. ```python 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. ```python 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.