Building Mini Spark, a Tiny Distributed Computing Engine

I wanted to understand what actually happens inside a distributed data engine, so I built a very small one in Python.

The result is Mini Spark. The name is intentionally ambitious: this first version is really a small MapReduce executor. It splits input into tasks, runs those tasks on several workers, shuffles intermediate results over HTTP, and retries work when a worker disappears. We will use pizza orders as the dataset because distributed systems are complicated enough without a complicated example too.

The complete source code is available in the Mini Spark repository.

A real-world analogy

Imagine you are organising a huge pizza party. Thousands of people have sent you their orders, and you have ended up with a CSV containing a name and a pizza type:

Name	Pizza type
Ada	Margherita
Grace	Mushrooms
Linus	Margherita

You need one total for each pizza type. On your own, you would read the file from top to bottom and keep a running count. With Alice, Bob, and Chris helping, you can split the list into three chunks and give one chunk to each person. They count their chunk, send back the totals, and you add those totals together.

That is the basic shape of the engine we are going to build.

Meet the compute engine

You play the coordinator (often called the driver), while Alice, Bob, and Chris are the workers. The coordinator splits up the job, hands out work, keeps track of progress, and decides what to do when something fails.

The workers will not always behave nicely. One may be slow, one may crash halfway through a task, and another may finish after the coordinator has already given up on it. Dealing with those cases is a large part of distributed execution.

Before getting into failures, it helps to separate the computation we asked for from the work the cluster has to perform.

The logical plan: what needs to happen

At the logical level, the pizza query is just:

1. Read every order.
2. Group the orders by pizza type.
3. Count the orders in each group.

There are no workers or partition sizes in that description. The query stays the same whether it runs on one worker or a thousand.

The physical plan: how it will happen

The physical side is more practical. It has to decide how that query will actually run:

1. Divide the list into smaller pieces.
2. Create tasks that process those pieces.
3. Assign the tasks to available workers.
4. Move partial counts for the same pizza type to the same place.
5. Combine them into the final totals.

This first version does not have a real logical-plan API or planner yet. We build this physical plan directly in CountPizza(). Part two will separate those layers properly, but the distinction is still useful for understanding the execution model here.

Tasks and partitions

Two ideas do most of the work: partitions split the data, and tasks describe what to do with each piece.

Partitions divide the data

A partition is simply a chunk of a dataset. Instead of treating the order list as one giant file, we divide it into ranges:

Pizza orders
├── Partition 1: rows 1–100
├── Partition 2: rows 101–200
├── Partition 3: rows 201–300
└── ...

Alice can work on one partition while Bob and Chris work on others. The partition size is a trade-off: a few large partitions are cheap to coordinate but harder to balance and expensive to retry; lots of tiny partitions spread out nicely but create more scheduling overhead.

Tasks divide the computation

A task is the piece of work we can hand to a worker. For Mini Spark, the useful mental model is:

Task = operation + input partition

Or, if you prefer notation:

T = (f, P)

where f is the operation and P is the input partition. Executing the task produces a result:

TaskResult = f(P)

Real engines often pipeline several operations into one task, but this simpler definition gives us something concrete to schedule, watch, and retry.

For example:

Task 17 = count pizza types in partition 17

A task is not permanently tied to a worker. Alice, Bob, or Chris should all be able to run task 17 and get the same answer. If Alice disappears, the coordinator can give the same task to Bob.

So we have three different things:

Partition = a unit of data
Task      = a unit of work
Worker    = something that executes the work

A worker picks up a task, reads the task’s input partition, runs the operation, and stores the result. Another task may then use that result as its input.

Coordinating tasks with a DAG

Tasks are not always independent. Some cannot start until others have finished:

Task 1: count pizzas in partition 1
Task 2: count pizzas in partition 2
Task 3: combine the results of tasks 1 and 2

Task 3 depends on tasks 1 and 2, so it has to wait:

Task 1 ──┐
       ├──> Task 3
Task 2 ──┘

Once there are more than a handful of tasks, the coordinator needs a proper picture of these relationships. Mini Spark stores them in a directed acyclic graph, or DAG:

• Each node represents a task.
• Each edge represents a dependency between two tasks.
• The direction of an edge shows how results flow through the computation.
• The graph is acyclic because a task cannot eventually depend on its own result.

The pizza job looks roughly like this:

Count partition 1 ──┐
Count partition 2 ──┼──> Combine partial counts ──> Final result
Count partition 3 ──┘

The first three tasks have no dependencies, so they can run straight away. The combining task remains blocked until all three have finished. In scheduler terms, a task is ready when every dependency has completed successfully.

The coordinator can track every task through a small set of states:

PENDING → READY → RUNNING → COMPLETED
                        ↘ FAILED → READY

If a task fails, it can return to READY and run again. Anything downstream stays blocked until one attempt succeeds.

Each retry gets an attempt number. That lets the coordinator ignore a late answer from an older attempt. Our operations are deterministic and their intermediate files are scoped to a specific attempt, so rerunning these particular tasks is safe. That does not make arbitrary side effects retry-safe; it works because the operations in this project are deliberately constrained.

Sharing task results

After the first batch of tasks, each worker holds a set of partial counts:

Alice: (Margherita, 10), (Mushrooms, 17)
Bob:   (Margherita, 12), (Cheese, 8)
Chris: (Mushrooms, 9),   (Cheese, 11)

They are correct, but they are not the final answer. We still need to bring together the counts for each pizza type:

Margherita = 10 + 12
Mushrooms  = 17 + 9
Cheese     = 8 + 11

That means moving intermediate results between workers. This is where the example becomes more interesting than three people counting separate pieces of paper.

The simple solution

If we knew there were exactly three pizza types, we could create one aggregation task for each one:

Task 4: sum all Margherita counts
Task 5: sum all Mushrooms counts
Task 6: sum all Cheese counts

But we normally do not know all the keys in advance. There may be 3 pizza types, 30, or 3,000. Reading the entire file just to discover them would defeat the point, so we need a way to route keys without knowing the full list.

Task results become new datasets

A task’s output is just another dataset. A counting task might read:

(Alice, Margherita)
(Bob, Mushrooms)
(Chris, Margherita)

and produce:

(Margherita, 2)
(Mushrooms, 1)

Those partial counts become the input to the next group of tasks:

Input dataset
    ↓
Input partitions
    ↓
Counting tasks
    ↓
Intermediate dataset
    ↓
Intermediate partitions
    ↓
Aggregation tasks
    ↓
Final dataset

That pattern can repeat: tasks consume partitioned data and produce more partitioned data. Simple operations such as map or filter can usually keep the existing partitions. An aggregation cannot. Every partial count for Margherita must reach the same reducer, wherever it was produced.

A hash partitioner gives us that routing rule:

partition_id = hash(key) % number_of_partitions

For our dataset, the key is the pizza type:

hash("Margherita") % 3 → partition 2
hash("Mushrooms")  % 3 → partition 0
hash("Cheese")     % 3 → partition 1

Every worker uses the same function, so every Margherita record goes to partition 2, no matter where it started.

Different pizza types may land in the same partition, which is fine. The important rule is that the same key always lands in the same partition for this shuffle.

This gives us the essential guarantee:

same key → same partition → same aggregation task

Each map task writes one block per destination partition. Reducer 0 fetches block 0 from every mapper, reducer 1 fetches block 1, and so on. Moving those blocks across the network is the shuffle.

A stage runs one task per partition. A shuffle separates one stage from the next, while the DAG still keeps track of the individual task dependencies.

flowchart LR
    subgraph S1["Stage 1 — count each input partition"]
        P1["Partition 1"] --> T1["Task 1<br/>local pizza counts"]
        P2["Partition 2"] --> T2["Task 2<br/>local pizza counts"]
        P3["Partition 3"] --> T3["Task 3<br/>local pizza counts"]
    end

    T1 --> SH["Shuffle by pizza type"]
    T2 --> SH
    T3 --> SH

    subgraph S2["Stage 2 — combine matching pizza types"]
        SH --> R1["Task 4<br/>reduce partition 0"]
        SH --> R2["Task 5<br/>reduce partition 1"]
        SH --> R3["Task 6<br/>reduce partition 2"]
    end

    R1 --> OUT["Final pizza counts"]
    R2 --> OUT
    R3 --> OUT

Implementation

That is enough theory. The Python version will:

• Use independent workers that communicate over the network.
• Retry work after some worker and task failures.
• Implement the pizza count:
  • Read a CSV from the file system.
  • Perform the group-by count in a distributed fashion.
  • Write the result to the file system.

The goal is to make the moving parts visible, not to make them fast. There is no spill-to-disk, clever query planner, or production-grade storage layer hiding behind the example.

The overall architecture

The cluster has one coordinator and three workers. Each runs as a separate process with its own HTTP server. Docker gives us enough isolation to treat them like small, separate machines:

                 ┌─────────────────┐
                 │   Coordinator   │
                 │  owns the DAG,  │
                 │   schedules     │
                 └───────┬─────────┘
            ┌────────────┼────────────┐
            │            │            │
      ┌─────▼─────┐ ┌────▼──────┐ ┌───▼───────┐
      │  Worker 1 │ │  Worker 2 │ │  Worker 3 │
      │ (volume)  │ │ (volume)  │ │ (volume)  │
      └───────────┘ └───────────┘ └───────────┘

There are three choices worth calling out:

• Workers pull tasks. Each worker keeps asking the coordinator for something to do. The coordinator never has to initiate a connection to a worker.
• Shuffle data moves over HTTP. A worker stores intermediate blocks on its own volume. Downstream workers fetch those blocks over the network instead of reading them from shared storage.
• The coordinator owns scheduling state. Workers execute tasks, but the coordinator decides which task and attempt count as current.

Docker Compose gives each worker a named volume (worker1-data, worker2-data, and so on). The input and final output use the shared ./data mount. Shuffle blocks stay on the worker that created them, which is why losing a worker can also mean losing completed intermediate output.

The main settings live in config.toml:

[coordinator]
num_partitions = 3
num_reduce_partitions = 3
input_file = "/data/orders.csv"
output_file = "/data/result.csv"
partitioned_output = false
output_dir = "/data/result"
 
[coordinator.workers]
worker1 = "http://worker1:5000"
worker2 = "http://worker2:5000"
worker3 = "http://worker3:5000"

The coordinator uses the worker URLs when it tells a reduce task where its input blocks live.

Modelling the computation

A task needs a type, some dependencies, and a state. For retries, it also needs an attempt number and the worker currently responsible for that attempt:

class TaskType(str, Enum):
    MAP_COUNT = "map_count"
    REDUCE_SUM = "reduce_sum"
    WRITE_RESULT = "write_result"
 
 
class TaskState(str, Enum):
    PENDING = "pending"
    READY = "ready"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
 
 
class Task(BaseModel):
    id: str
    task_type: TaskType
    dependencies: list[str] = Field(default_factory=list)
    state: TaskState = TaskState.PENDING
    attempt: int = 0
    assigned_worker: Optional[str] = None
    started_at: Optional[float] = None
    last_heartbeat_at: Optional[float] = None
    output_worker_url: Optional[str] = None
 
    # map_count fields
    input_file: Optional[str] = None
    start_row: Optional[int] = None
    end_row: Optional[int] = None
    num_reduce_partitions: Optional[int] = None
 
    # reduce_sum fields
    reduce_partition_id: Optional[int] = None
    upstream_task_ids: list[str] = Field(default_factory=list)
 
    # write_result fields
    output_file: Optional[str] = None

output_worker_url is important. Once a task finishes, the coordinator records which worker has the files. Together with the job, task, and attempt IDs, that URL points a downstream task to the exact shuffle block it should fetch.

The state machine stays small:

PENDING → READY → RUNNING → COMPLETED
                     ↘ FAILED → READY (retry)

A task starts as PENDING. Once its dependencies are complete, it becomes READY. Giving it to a worker moves it to RUNNING; a failed attempt can send it back to READY for another try.

Building the DAG

CountPizza() builds the physical task DAG. It decides which tasks exist and how they depend on each other, but it does not assign them to workers. The three stages connect like this:

flowchart LR
    subgraph S1["Stage 1 — count each input partition"]
        M0["map_0"] & M1["map_1"] & M2["map_2"]
    end
    subgraph S2["Stage 2 — sum matching pizza types"]
        R0["reduce_0"] & R1["reduce_1"] & R2["reduce_2"]
    end
    M0 & M1 & M2 --> R0 & R1 & R2
    R0 & R1 & R2 --> W0["write_0"]

Every reduce task waits for every map task. With the default single output file, the write task then waits for every reducer. The code is mostly task construction and dependency wiring:

def CountPizza(input_file, num_partitions, num_reduce_partitions, output_file):
    total_rows = count_rows(input_file)
    base_size, remainder = divmod(total_rows, num_partitions)
 
    tasks = []
    start = 0
 
    # Stage 1: one map task per input partition
    for i in range(num_partitions):
        partition_size = base_size + (1 if i < remainder else 0)
        end = start + partition_size
        tasks.append(Task(
            id=f"map_{i}",
            task_type=TaskType.MAP_COUNT,
            input_file=input_file,
            start_row=start,
            end_row=end,
            num_reduce_partitions=num_reduce_partitions,
        ))
        start = end
 
    map_task_ids = [t.id for t in tasks]
 
    # Stage 2: one reduce task per output partition
    for i in range(num_reduce_partitions):
        tasks.append(Task(
            id=f"reduce_{i}",
            task_type=TaskType.REDUCE_SUM,
            dependencies=map_task_ids,        # wait for ALL map tasks
            reduce_partition_id=i,
            upstream_task_ids=map_task_ids,
        ))
 
    # Stage 3: a single task that writes the final CSV
    reduce_task_ids = [f"reduce_{i}" for i in range(num_reduce_partitions)]
    tasks.append(Task(
        id="write_0",
        task_type=TaskType.WRITE_RESULT,
        dependencies=reduce_task_ids,
        upstream_task_ids=reduce_task_ids,
        output_file=output_file,
    ))
 
    return tasks

If the row count does not divide evenly, the first few partitions get one extra row. That keeps the ranges as balanced as possible.

By default, one write task combines the reducer outputs into result.csv. Setting partitioned_output creates one writer per reduce partition instead, producing part-00000.csv, part-00001.csv, and so on.

A deliberate simplification

The example above begins by counting all the rows in the input file:

total_rows = count_rows(input_file)
base_size, remainder = divmod(total_rows, num_partitions)

This is not good distributed input handling. Before any worker starts, the coordinator scans the whole CSV just to count its rows. Then each worker opens the file at the beginning and skips forward to its range. Later partitions therefore reread all the rows before them.

I kept it this way because row ranges make the first version easy to follow, but the cost grows badly as we add partitions. A better implementation splits the file into byte ranges. Each worker seeks to its starting offset, moves to the next complete record, and reads from there. That is one of the first things to fix in part two.

The operations

Next are the functions the workers actually run. The hash partitioner must return the same answer in every process, so Python’s randomised built-in hash() is not suitable. MD5 is plenty for stable partition assignment here:

import hashlib
from itertools import islice
 
def hash_partition(key: str, num_partitions: int) -> int:
    h = int(hashlib.md5(key.encode()).hexdigest(), 16)
    return h % num_partitions

The map operation reads its CSV range, counts pizza types locally, and splits those counts into one block per reduce partition:

def map_count(input_file, start_row, end_row, num_reduce_partitions,
              job_id, task_id, attempt):
    with open(input_file) as f:
        reader = csv.reader(f)
        next(reader)                      # skip the header
 
        counts = {}
        for row in islice(reader, start_row, end_row):
            pizza_type = row[1].strip()
            counts[pizza_type] = counts.get(pizza_type, 0) + 1
 
    # Route each pizza type to its destination partition
    blocks = [{} for _ in range(num_reduce_partitions)]
    for pizza_type, count in counts.items():
        pid = hash_partition(pizza_type, num_reduce_partitions)
        blocks[pid][pizza_type] = blocks[pid].get(pizza_type, 0) + count
 
    # Write each block to this worker's own volume
    task_dir = task_output_dir(job_id, task_id, attempt)
    for i, block in enumerate(blocks):
        _write_json_atomic(os.path.join(task_dir, f"block_{i}.json"), block)

The routing line is hash_partition(pizza_type, num_reduce_partitions). Every mapper uses it, so all partial counts for "Margherita" end up in the same numbered block.

islice(reader, start_row, end_row) avoids loading the whole CSV into memory, but it does not jump directly to start_row. The CSV has no row index, so the reader still parses and discards everything before that point. This is the scaling problem mentioned above.

The reduce operation fetches its numbered block from every mapper and merges the dictionaries:

def reduce_sum(fetch_from, job_id, task_id, attempt):
    merged = {}
    for source in fetch_from:
        url = (
            f"{source['worker_url']}/block/{source['job_id']}/"
            f"{source['task_id']}/{source['attempt']}/{source['block_id']}"
        )
        block = httpx.get(url).json()          # shuffle read over HTTP
        for pizza_type, count in block.items():
            merged[pizza_type] = merged.get(pizza_type, 0) + count
 
    task_dir = task_output_dir(job_id, task_id, attempt)
    _write_json_atomic(os.path.join(task_dir, "result.json"), merged)

That loop is the entire shuffle read. The coordinator prepares fetch_from with the location and identifiers for each block. The reducer downloads them one at a time and adds the counts together. Writing works in much the same way: fetch reducer outputs, prepare a CSV, and let the coordinator publish it only if the result belongs to the current attempt.

Deliberate simplifications

These functions deliberately mix input, processing, and output. They also hold dictionaries in memory and transfer whole JSON objects. That works for pizza orders, but not for an unbounded input or a high-cardinality key space. A serious shuffle needs batching, bounded buffers, backpressure, spill-to-disk, and a merge strategy. Part two will start pulling those responsibilities apart.

The coordinator: scheduling tasks

The coordinator wraps the task list in a Job. Its scheduler is intentionally plain: scan the tasks and promote a PENDING task to READY when all of its dependencies are complete.

class Job:
    def __init__(self, job_id, tasks):
        self.tasks = {t.id: t for t in tasks}
 
    def get_ready_tasks(self):
        ready = []
        for task in self.tasks.values():
            if task.state == TaskState.PENDING:
                deps = [self.tasks[d] for d in task.dependencies]
                if all(d.state == TaskState.COMPLETED for d in deps):
                    task.state = TaskState.READY
            if task.state == TaskState.READY:
                ready.append(task)
        return ready

When a worker asks for work, it gets the first ready task:

@app.post("/task_request")
async def task_request(request: TaskRequest):
    ready_tasks = current_job.get_ready_tasks()
    if not ready_tasks:
        return {"task": None}
 
    task = ready_tasks[0]
    task.state = TaskState.RUNNING
    task.assigned_worker = request.worker_id
    task.started_at = time.time()
    task.last_heartbeat_at = task.started_at
    task.attempt += 1
 
    return {"task": build_task_spec(task, current_job).model_dump()}

build_task_spec turns the coordinator’s task record into instructions for a worker. A map task only needs its file range. Reduce and write tasks also need a fetch_from list, which the coordinator can build because it recorded where every upstream result was stored:

def build_task_spec(task, job):
    spec = TaskSpec(
        job_id=job.job_id,
        id=task.id,
        attempt=task.attempt,
        task_type=task.task_type,
    )
 
    if task.task_type == TaskType.REDUCE_SUM:
        spec.reduce_partition_id = task.reduce_partition_id
        for upstream_id in task.upstream_task_ids:
            upstream = job.tasks[upstream_id]
            spec.fetch_from.append(FetchSource(
                worker_url=upstream.output_worker_url,   # where the blocks live
                job_id=job.job_id,
                task_id=upstream_id,
                attempt=upstream.attempt,
                block_id=task.reduce_partition_id,
            ))
    # ... and similarly for MAP_COUNT and WRITE_RESULT
    return spec

Results are accepted only if they belong to the active job, current attempt, and assigned worker. Anything late is acknowledged and ignored:

@app.post("/task_result")
async def task_result(result: TaskResult):
    if not current_job or result.job_id != current_job.job_id:
        return {"ok": True, "accepted": False}
 
    task = current_job.tasks.get(result.task_id)
 
    if (not task
            or task.state != TaskState.RUNNING
            or result.attempt != task.attempt
            or result.worker_id != task.assigned_worker):
        return {"ok": True, "accepted": False}
 
    if result.success:
        task.state = TaskState.COMPLETED
        task.output_worker_url = WORKER_URLS[result.worker_id]
    else:
        task.state = TaskState.READY if task.attempt < MAX_RETRIES else TaskState.FAILED
    return {"ok": True}

The worker: executing tasks

The worker loop is not glamorous. Ask for a task, run it while sending heartbeats, report the result, and repeat. If the queue is empty, wait a second:

async def worker_loop():
    while True:
        async with httpx.AsyncClient() as client:
            data = (await client.post(
                f"{COORDINATOR_URL}/task_request",
                json={"worker_id": WORKER_ID},
            )).json()
 
            spec = data.get("task")
            if spec:
                result = await execute_with_heartbeats(client, spec)
                await client.post(f"{COORDINATOR_URL}/task_result", json={
                    "job_id": spec["job_id"],
                    "task_id": spec["id"],
                    "attempt": spec["attempt"],
                    "worker_id": WORKER_ID,
                    "success": result.success,
                    "failed_upstream_task_id": result.failed_upstream_task_id,
                    "failed_upstream_attempt": result.failed_upstream_attempt,
                    "output_path": result.output_path,
                })
            else:
                await asyncio.sleep(1)

Workers also serve their intermediate blocks to one another through a small endpoint:

@app.get("/block/{job_id}/{task_id}/{attempt}/{block_id}")
async def get_block(job_id: str, task_id: str, attempt: int, block_id: str):
    name = "result.json" if block_id == "result" else f"block_{block_id}.json"
    path = os.path.join(task_output_dir(job_id, task_id, attempt), name)
    with open(path) as f:
        return JSONResponse(content=json.load(f))

Fault tolerance

If a worker dies after receiving a task, it never sends a result and the task would otherwise stay RUNNING forever. To avoid that, workers renew a lease with regular heartbeats. A watchdog checks those leases and reclaims stale tasks:

async def timeout_checker():
    while True:
        await asyncio.sleep(5)
        for task in current_job.tasks.values():
            if task.state == TaskState.RUNNING and task.started_at:
                last_seen = task.last_heartbeat_at or task.started_at
                if time.time() - last_seen > TASK_TIMEOUT_SECONDS:
                    task.state = (TaskState.FAILED
                                  if task.attempt >= MAX_RETRIES
                                  else TaskState.READY)

Once a timed-out task is back in READY, another worker can take it. Attempt IDs stop an old worker from winning the race later, and attempt-scoped files stop retries from overwriting one another. The final output is published only after the coordinator accepts the matching attempt.

There is another failure mode: a worker can disappear after finishing a map task, taking its local shuffle blocks with it. When a reducer cannot fetch a block, it reports the exact upstream attempt that is missing. The coordinator invalidates that output, reruns the producer, and then retries the reducer, as long as there is still retry budget left.

This is useful worker-level recovery, but it is not full engine fault tolerance. The coordinator keeps the authoritative state in memory and remains a single point of failure. Restart it and the running job is gone.

Running it

Start the cluster and submit a job:

docker compose up --build -d
curl -X POST localhost:8000/jobs

Polling the job shows which worker ended up with each task. One run may look like this:

map_0    completed   worker3
map_1    completed   worker1
map_2    completed   worker2
reduce_0 completed   worker2
reduce_1 completed   worker3
reduce_2 completed   worker1
write_0  completed   worker1

With the default single-file output, data/result.csv contains:

PizzaType	Count
Cheese	17
Hawaiian	10
Margherita	34
Mushrooms	20
Pepperoni	19

With partitioned_output enabled, each reducer gets its own part-*.csv file under data/result/. A partition with no matching keys produces a file containing only the header.

Killing a worker

The failure path is easier to understand when you see it happen. Submit a job, then kill worker 1 while it is running a map task:

curl -X POST localhost:8000/jobs
docker kill mini-spark-worker1-1

At first, map_2 still appears as RUNNING:

map_2 running attempt=1 worker1   ← worker1 is gone

After the lease expires, the watchdog returns it to the queue and another worker picks it up:

map_2 completed attempt=2 worker2  ← recomputed elsewhere

The job completes with the same counts. This is the behaviour we wanted: losing a worker makes the job slower, but does not immediately make it wrong.

What we built and what we skipped

The first version now has:

• a hand-built physical task plan for one read, group, and count computation,
• partitions and tasks as the units of data and work,
• a DAG that drives ordering and parallelism,
• an HTTP shuffle that moves data between workers without shared storage, and
• limited worker failure recovery through attempt tracking, heartbeat-based leases, stale-result rejection, and re-execution.

It also cuts plenty of corners. There is no speculative execution for slow workers, no operator pipelining, no compression or spill-to-disk, and no recovery if the coordinator dies. The implementation is small precisely because it stops before many of the difficult production problems begin.

Where we go next

Think of this as part one. We now have the execution foundation of a small MapReduce engine: one fixed map–shuffle–reduce job split into tasks, scheduled across workers, and retried when worker-local output disappears.

It is not a general-purpose Spark-like engine yet. CountPizza() creates one physical graph directly. The CSV reader repeatedly scans earlier rows, and the shuffle holds whole JSON dictionaries in memory. Those are not small performance details; they are boundaries of the current design.

Part two will build on this foundation with:

• a lazy, user-programmable logical plan and a planner that turns it into physical stages and tasks,
• proper CSV input partitioning so workers can seek to independent byte ranges instead of repeatedly scanning preceding rows,
• a bounded, pipelined shuffle with streaming, batching, spill-to-disk, and backpressure,
• pipelines of arbitrary Python transformations, including the function serialisation and worker execution model they require, and
• distributed inner joins, including the shuffle and partitioning rules needed to bring matching keys together.

That should move the project from a purpose-built pizza counter towards a small programmable dataflow engine. It also raises the questions I have avoided so far: where to split stages, which operations can share a pipeline, how to ship Python functions safely, how to plan a join, and how to keep intermediate state bounded when the input is larger than memory.