--- title: 'Distributed File Systems' source: 'https://academia.sh/en/courses/distributed-data-processing/distributed-file-systems' course: 'Distributed Data Processing' language: en updated: '2026-10-21T05:03:00+00:00' license: 'CC BY-SA 4.0' objectives: - 'Splits data into fixed-size blocks and computes the placement of the blocks on the cluster with a fictional placement rule.' - 'Measures data locality as the number of blocks read locally and compares the effect of the replication factor on locality and on stored blocks.' - 'Counts separately the effect of locality-aware assignment and of balanced assignment on the makespan.' --- # Distributed File Systems Splitting data into blocks, placing them on the cluster, and measuring whether the worker that processes a block already holds it: when 2200 rows are split into 22 blocks of 100 rows, with a single copy and assignment by block order only 5 of the 22 blocks are read locally, the makespan is 1850 and the total work is 5600; at a replication factor of 2 the local blocks rise to 17 and the makespan falls to 1150, at 3 they rise to 22 and the makespan falls to 650 but the stored blocks grow from 22 to 66, and a fourth copy adds nothing to reading; locality-aware assignment reads 22 of 22 blocks locally with a single copy, but because the placement is unbalanced the makespan stays at 750 instead of 650. In the previous lesson the rows sat ready in a list and `distribute` handed them out to the workers. The cost of handing out was counted as zero, because the list was already in memory. This assumption is true on a single machine and false on a cluster: the data sits spread over the disks of the machines, the rows a worker will process may be on another machine, and to see those rows they have to be pulled over the network. This lesson splits the data into blocks, places them on the cluster and measures a single question: what is gained if the worker that processes a block already holds it. The answer is called **data locality**, and its cost is paid in two separate places, one in makespan and one in disk. - **DC9.** Block placement and replication are **fictional**; no real distributed file system product, block server or cloud storage service is called. The seed is **20260218**. - **DC10.** The **2200 rows** coming out of the pipeline are split in order into blocks of **100 rows**; there are **22 blocks**. The block is the resolution of the partitioning: a block is not split and goes in its entirety to a single worker. - **DC11.** **The placement rule:** the first copy of a block is chosen from the trace of its key, and the remaining copies fall on the following workers on the ring. The **replication factor** (`copies`) is the number of workers a block sits on. - **DC12.** If the worker that reads a block holds it, the cost per row is **1**; if it does not, it is **`NETWORK_COST`, that is, 3**. Every makespan and total work measured in this lesson is sensitive to this choice. - **DC13.** The **durability** side of replication, that is, the data surviving when a machine goes down, was measured in the Relational Database Administration course of the Databases curriculum and in the System Design and Distributed Systems curriculum. **The procedure is not repeated here**; this lesson counts replication only through its effect on **reading**. - **DC14.** File layout, the columnar format and partitioning were measured in the Data Storage Architectures course and are **not repeated**. The block here is not a file format but a unit of placement. - **DC15.** The baseline is again **a single worker**: all blocks sit on the same worker, locality is complete, makespan **2250**, total work **2200**. ## Block and Placement What sets a distributed file system apart from a single-machine file system is that it does not store a file as a whole. The file is cut into fixed-size **blocks**, and the blocks are distributed to the machines in the cluster. This has two consequences. First, a file that does not fit on a single machine can be stored. Second, different pieces of the same file can be read **at the same time**, because they sit on different disks. A block is a unit of splitting, not a partition. The block boundary does not look at the key of the rows; it looks at which row of the file it is. So the rows of the same subscriber can be spread over two blocks, and a single block can hold all five regions. The placement rule does not look at the data either: it produces a trace from the order of the block and chooses a worker with that trace. The copies fall on the following workers on the ring, so the loss of one worker does not take away all the copies of a single block. ```python # FICTIONAL. The source, pipeline and distributed file system are defined in the lesson; # no real file system product or block server 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, NETWORK_COST, TASK_OVERHEAD = 4, 3, 50 def rows(): v, _ = s1_screen(generate_source()) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return v def trace_id(m): h = 2166136261 for c in str(m): h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF return h # ---- FICTIONAL distributed file system: block , placement , read BLOCK_SIZE = 100 def split_blocks(records, size=BLOCK_SIZE): """Rows are cut into fixed-size blocks in order.""" return [records[i:i + size] for i in range(0, len(records), size)] def placement(block_no, copies, workers=WORKERS): """The first copy follows the block's trace , the rest go to the next worker on the ring.""" first = trace_id(f"blok{block_no}") % workers return [(first + j) % workers for j in range(copies)] def read_blocks(blocks, copies, locality, workers=WORKERS): """One worker reads each block. A worker that does not hold the block pulls the rows over the network.""" work = {i: 0 for i in range(workers)} local = 0 for b, block in enumerate(blocks): holders = placement(b, copies, workers) reader = holders[0] if locality else b % workers if reader in holders: local += 1 work[reader] += len(block) else: work[reader] += len(block) * NETWORK_COST return {"local": local, "remote": len(blocks) - local, "stored": len(blocks) * copies, "duration": max(work.values()) + TASK_OVERHEAD, "total_work": sum(work.values())} ``` ## Locality Is a Coincidence In the first setup the driver gives the blocks to the workers **in order**: block zero to worker zero, block one to worker one, and so on. This assignment equalizes the load, because about the same number of blocks falls on every worker. It does not take the placement into account at all; that the block a worker reads is held by that worker is a coincidence. The block below counts, across replication factors, in how many blocks that coincidence happens. ```python BLOCKS = split_blocks(rows()) single = read_blocks(BLOCKS, 1, False, 1) print("blocks:", len(BLOCKS), " each", BLOCK_SIZE, "rows") print("single-worker baseline -> makespan:", single["duration"], " total work:", single["total_work"]) print() print(f"{'copies':>8}{'local blocks':>14}{'remote blocks':>15}{'stored':>8}{'makespan':>10}{'total work':>12}") for r in (1, 2, 3, 4): o = read_blocks(BLOCKS, r, False) print(f"{r:>8}{o['local']:>14}{o['remote']:>15}{o['stored']:>8}" f"{o['duration']:>10}{o['total_work']:>12}") ``` ``` blocks: 22 each 100 rows single-worker baseline -> makespan: 2250 total work: 2200 copies local blocks remote blocks stored makespan total work 1 5 17 22 1850 5600 2 17 5 44 1150 3200 3 22 0 66 650 2200 4 22 0 88 650 2200 ``` The first row shows the hidden bill of distributing. With a single copy only **five** of the twenty-two blocks are read locally; the remaining seventeen blocks cross the network and each of their rows is counted three times. The total work becomes **5600** instead of 2200, and the makespan **1850**. The number five is close to the expected value of a coincidence: with four workers the probability that the reader and the holder overlap is one in four, and one quarter of twenty-two blocks is 5.5. In the previous lesson the same four workers had brought the makespan down to 611, but that measurement had counted reading as free. Once reading is counted, the four-worker setup falls only from 2250 to 1850 compared with the single-worker baseline; the gain is less than one fifth, not one quarter. The second and third rows give what replication adds to reading. When every block has two copies, the locally read blocks rise from 5 to **17** and the makespan falls to **1150**; with three copies all twenty-two blocks are read locally, the total work returns to **2200** and the makespan becomes **650**. The network traffic disappears completely. These numbers depend on the choice `NETWORK_COST = 3`: if the cost of moving were 1, all four rows would give the same total work and the contribution of replication to reading would not be visible. The fourth row is the limit of replication. With four copies the stored blocks rise to **88**, that is, the data grows fourfold, but the local blocks and the makespan stay **exactly the same** as with three copies. After the third copy there is nothing that is paid for in reading; the justification of the fourth copy lies elsewhere, and that justification is durability. Durability is not measured in this course; it was measured in the prerequisite courses. ## Locality or Balance There is an alternative to buying locality with replication: the driver gives the blocks not in order but to **the worker that holds them**. This assignment provides full locality even with a single copy. In return it hands the balance of the load over to the placement rule, and that rule does not try to equalize the load. The last block compares the two assignments at two replication factors. ```python print(f"{'assignment':<16}{'copies':>8}{'local':>7}{'stored':>8}{'makespan':>10}{'total work':>12}") for name, y, r in (("block order", False, 1), ("block order", False, 3), ("locality-aware", True, 1), ("locality-aware", True, 3)): o = read_blocks(BLOCKS, r, y) print(f"{name:<16}{r:>8}{o['local']:>7}{o['stored']:>8}" f"{o['duration']:>10}{o['total_work']:>12}") ``` ``` assignment copies local stored makespan total work block order 1 5 22 1850 5600 block order 3 22 66 650 2200 locality-aware 1 22 22 750 2200 locality-aware 3 22 66 750 2200 ``` The third row catches full locality with a single copy: twenty-two of twenty-two blocks are read locally, the total work falls to **2200** and the stored blocks stay at **22**. Assignment by block order obtains the same result only with three copies, that is, by storing **66 blocks**. Locality-aware assignment saves three times the disk. On the other hand its makespan is **750**, **100 units** more than the three-copy case of assignment by block order. The reason does not show in the table but lies in the placement rule: the blocks are not evenly distributed to the workers, and the worker that holds the most blocks reads seven blocks while the one that holds the fewest reads four. Locality-aware assignment cannot choose the reader, because if it did, locality would break. The fourth row writes this once more: raising the replication to three does not change the makespan at all in locality-aware assignment, because the worker that holds the first copy has not changed. The choice between the two rows cannot be reduced to a single number. For a hundred units of makespan, forty-four more blocks are stored, or in place of forty-four blocks a hundred units of makespan are paid. Which one is cheaper depends on the disk of the cluster and on how often its job runs, and this decision can be made only when both numbers are written. ## Block Size Is the Resolution of the Partitioning The reason the locality-aware assignment stays at 750 does not show in the table, because the table does not write how the blocks fall on the workers. The last block first prints that distribution, then changes the block size and measures how much of the balance comes from the size. ```python S = rows() holder_count = {i: 0 for i in range(WORKERS)} for b in range(len(BLOCKS)): holder_count[placement(b, 1)[0]] += 1 print("blocks holding the first copy:", [holder_count[i] for i in range(WORKERS)]) print() print(f"{'block size':>12}{'blocks':>8}{'local':>7}{'makespan':>10}{'total work':>12}") for size in (25, 50, 100, 220, 550): chunks = split_blocks(S, size) o = read_blocks(chunks, 1, True) print(f"{size:>12}{len(chunks):>8}{o['local']:>7}{o['duration']:>10}{o['total_work']:>12}") ``` ``` blocks holding the first copy: [7, 4, 4, 7] block size blocks local makespan total work 25 88 88 625 2200 50 44 44 650 2200 100 22 22 750 2200 220 10 10 710 2200 550 4 4 600 2200 ``` The first line gives the missing justification. Twenty-two blocks fell on four workers as **7, 4, 4, 7**; the worker that holds the most blocks reads seven blocks, that is, 700 rows, and with 700 plus the fixed cost of 50 the makespan is **750**. The placement rule does not try to equalize the load, it looks at the trace of the block; the imbalance is not a flaw but the natural result of the rule. The sweep shows how this imbalance relates to the block size. With blocks of twenty-five rows eighty-eight blocks are formed and the makespan falls to **625**, with fifty rows it is **650**: as the block gets smaller, the share of each worker approaches the average, because the weight of a single block falling in the wrong place decreases. The local blocks equal the number of blocks in every row and the total work is fixed at **2200**; the block size does not touch the network traffic, only the balance. The last two rows prevent drawing a one-directional rule. Ten blocks of two hundred and twenty rows bring the makespan down to **710**, and four blocks of five hundred and fifty rows to **600**, which is better than both the hundred-row setup and the twenty-five-row one. The reason is this: the four blocks happened to fall on four distinct workers, and the distribution came out as 1, 1, 1, 1. A small block improves the balance **in expectation**; it does not guarantee it in a single setup. In this model shrinking the block also has no visible cost, because the fixed task cost is counted once per worker, not per block; in a real cluster every block gives birth to a task, and that cost comes back with small blocks. ## Summary - A distributed file system cuts data into fixed-size **blocks** and spreads them over the cluster. 2200 rows become **22 blocks** of 100 rows; a block is not split and goes in its entirety to a single worker. - **Data locality** is the worker that reads a block already holding it. If it is local, the cost per row is 1; if not, it is **`NETWORK_COST`, that is, 3**; all the numbers measured are sensitive to this choice. - With assignment by block order and a single copy, only **5** of the 22 blocks are read locally: makespan **1850**, total work **5600**. At a replication factor of 2 the local blocks are **17** and the makespan **1150**, at 3 they are **22** and **650**. - The contribution of replication to reading ends at the third copy: with four copies the stored blocks rise to **88**, the local blocks and the makespan do not change. The justification for the fourth copy is durability, and it was measured in the prerequisite courses. - Locality-aware assignment gives full locality with a single copy (**22** local blocks, **22** stored blocks), but because the placement is unbalanced the makespan stays at **750** instead of **650**. Locality and balance are not bought at the same time; the assignment cannot be chosen without writing both. - Block size is the resolution of the balance: the 22 blocks fall on the workers as **7, 4, 4, 7**, and with blocks of 25 rows the makespan falls to **625**. But the rule is not one-directional; four blocks of 550 rows happen to fall in a balanced way and give **600**. A small block improves the balance in expectation, it does not guarantee it. ## Next Step In these two lessons every worker read its own share and processed its own share; no row passed from one worker to another. This is possible only when the computation is done row by row. When a result such as the total per region is wanted, the rows of the same region are spread over different workers and they have to come together before they are summed. The next lesson builds a model that splits the computation into two steps and asks a single question: which step stays local, which one touches the network, and how much does the cost of producing the same result change with this distinction.