What is a cluster? Masters and workers

The head chef who never cooks

What is a cluster?

Words you'll meet in this lesson

Masters and workers in Hadoop

Clusters at three levels

Kinds of clusters, and two ways to organise them

Master–worker vs peer-to-peer

How a master and its workers run one job

1. Workers join and register

   When a worker starts, it contacts the master: “I am dn4, on /rack2, with 32 cores, 256 GB RAM and 144 TB of disk.”

   The master adds a row to its worker table. In HDFS, the DataNode also sends a block report: the list of every block on its disks.

2. Heartbeats every few seconds

   last_seen = {"dn1": 120.0, "dn2": 121.5, "dn3": 119.0}   # time of last heartbeat (seconds)
   now = 122.0
   for w, t in last_seen.items():
       print(w, "silent for", now - t, "s")

   dn1 silent for 2.0 s dn2 silent for 0.5 s dn3 silent for 3.0 s
   The master only stores when it last heard from each worker. Silence for a few seconds is normal. Silence past the timeout means trouble.

3. A job arrives and is split into tasks

   A user submits a job on the master. The job is divided into many small, independent tasks, for example one per 128 MB piece of the input.

   Small tasks are the secret: if one fails, only that small piece is redone, not the whole job.

4. The master assigns tasks to free workers

   Each worker gets as many tasks as it has free slots. The rest wait in a queue.

   In YARN, a “slot” is a container: a fixed amount of memory and CPU on one node. The master picks workers that are free and, ideally, already hold the task's data (Lesson 1.5).

5. Workers report progress

   Heartbeats now carry “task 7: 60% done”, “task 3: finished”.

   This is how you see map 45% reduce 0% in your terminal: the master adds up progress from all workers.

6. A worker dies: its tasks are re-run

   Worker 3 loses power. Its heartbeats stop. After the timeout, the master marks it dead and gives its unfinished tasks to other workers.

   The job finishes a little later, but it finishes, and nobody had to do anything by hand. That is the core promise of a cluster built from ordinary machines.

A job with a dying worker: what Riya sees, and what the master does

Riya runs a job on a 3-worker cluster. Halfway through, worker 2 crashes. Left: her terminal. Right: the master's view.

Try it yourself: break a worker during a job

Now do it for real in the Cluster Playground. Run the job, then stop a node while its task bars are filling.

Masters and workers everywhere

Examples are simplified. Real deployments add more components (backups, monitoring, security) around these basic roles.

A tiny master and its workers in Python

This program plays the master. It has a queue of tasks and three workers, and gives each free worker the next task, round after round, until the queue is empty. Then try the challenges, including a worker that dies.

from collections import deque

queue = deque(["T1", "T2", "T3", "T4", "T5", "T6", "T7"])
workers = ["W1", "W2", "W3"]
done = {w: [] for w in workers}

round_no = 0
while queue:
    round_no += 1
    for w in workers:
        if queue:
            done[w].append(queue.popleft())
    print("after round", round_no, "->", done)

• queue = deque([...])A deque (double-ended queue) gives the next item quickly with popleft(): first in, first out, like a real job queue.
• done = {w: [] for w in workers}The master's table: for each worker, the list of tasks it has run.
• while queue:Keep going while tasks are waiting. An empty deque counts as False.

Run it on your PC: a master with real worker processes

Your computer can be a tiny cluster too: one master process and one worker process per CPU core. Python's ProcessPoolExecutor keeps a task queue and gives tasks to whichever worker is free, exactly like a cluster master. Save this as workers.py and run it.

# workers.py: a master process hands out tasks to worker processes on your own PC
import os, time
from concurrent.futures import ProcessPoolExecutor, as_completed

def task(n):
    """One task: count the even numbers in its own range of 20 lakh numbers."""
    start = n * 2_000_000
    evens = sum(1 for x in range(start, start + 2_000_000) if x % 2 == 0)
    return n, os.getpid(), evens

if __name__ == "__main__":               # this part runs only in the master
    workers = os.cpu_count()
    print("Master PID", os.getpid(), "starting", workers, "workers")
    t0 = time.time()
    with ProcessPoolExecutor(max_workers=workers) as pool:
        futures = [pool.submit(task, n) for n in range(8)]      # 8 tasks in the queue
        total = 0
        for f in as_completed(futures):
            n, pid, evens = f.result()
            total += evens
            print(f"  task {n} done by worker PID {pid}")
    print("Total evens:", total)
    print(f"Time: {time.time() - t0:.1f} s")

# In a terminal:  python workers.py

The master's timeout check, one line at a time

The master loops over its table of last heartbeats. Watch silent change for each worker and see which ones end up in the dead list.

Worked examples: real problems, solved and traced in memory

Three jobs a cluster master really does: share out work, notice dead workers, and recover their tasks. Each solution runs live, and the trace shows the master's table changing in memory.

files part-1 … part-7 · workers W1, W2, W3

W1 ['part-1', 'part-4', 'part-7']
W2 ['part-2', 'part-5']
W3 ['part-3', 'part-6']

now = 2000 s · dn1 1998, dn2 1200, dn3 1999, dn4 1500, dn5 1360

Dead : ['dn2', 'dn5']
Alive: 3 of 5

W1 [T1, T4] · W2 [T2, T5, T6] · W3 [T3] · dead W2

{'W1': ['T1', 'T4', "T5'"], 'W3': ['T3', "T2'", "T6'"]}

Check yourself

Quick revision

Interview and viva questions

Practice problems

Small programs that do the master's job: count, check health, assign and recover. The business problems are taken from real cluster operations.

[("nn1","master"),("rm1","master"),("dn1","worker"),("dn2","worker"),("dn3","worker")]

{'master': 2, 'worker': 3}

nodes = [("nn1", "master"), ("rm1", "master"), ("dn1", "worker"), ("dn2", "worker"), ("dn3", "worker")]
count = {}
for name, role in nodes:
    count[role] = count.get(role, 0) + 1
print(count)

3 s interval · 500 nodes

One node: 1200
500 nodes: 600000

per_node = 3600 // 3
print("One node:", per_node)
print("500 nodes:", per_node * 500)

recheck 300 s · heartbeat 3 s

630 seconds = 10 min 30 s

t = 2 * 300 + 10 * 3
print(t, "seconds =", t // 60, "min", t % 60, "s")

N = 50, W = 6, S = 4

3 waves

import math
N, W, S = 50, 6, 4
print(math.ceil(N / (W * S)), "waves")

W1 [30, 32, 31] · W2 [29, 30] · W3 [95, 90] · W4 [33]

W3 is slow: 92 s vs cluster 46 s

times = {"W1": [30, 32, 31], "W2": [29, 30], "W3": [95, 90], "W4": [33]}
all_t = [t for ts in times.values() for t in ts]
avg = sum(all_t) / len(all_t)
for w, ts in times.items():
    a = sum(ts) / len(ts)
    if a > 1.5 * avg:
        print(w, "is slow:", round(a), "s vs cluster", round(avg), "s")

["dn4","dn2","dn4","dn7","dn4","dn2","dn9","dn2","dn2"]

Stop using: ['dn2', 'dn4'] {'dn4': 3, 'dn2': 4, 'dn7': 1, 'dn9': 1}

failures = ["dn4", "dn2", "dn4", "dn7", "dn4", "dn2", "dn9", "dn2", "dn2"]
count = {}
for n in failures:
    count[n] = count.get(n, 0) + 1
bad = sorted(n for n, c in count.items() if c >= 3)
print("Stop using:", bad, count)

400 tasks × 6 min · 20 workers × 8 slots · 8 tasks redone

Waves: 3 | minutes: 18
Finish at about 1:18 am

import math
tasks, minutes = 400 + 8, 6
slots = 19 * 8                     # plan for one worker lost
waves = math.ceil(tasks / slots)
total = waves * minutes
print("Waves:", waves, "| minutes:", total)
print(f"Finish at about {1 + total // 60}:{total % 60:02d} am")

60 TB lost · 2 GB/s

8.5 hours

tb = 60
gb_per_s = 2
hours = tb * 1024 / gb_per_s / 3600
print(round(hours, 1), "hours")

Real-world scenario questions (asked in company interviews)

Interview situations about masters, workers and heartbeats. Think of your answer, then compare it with the model answer.