Distributed File Systems

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.

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

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.

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.

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.