Distributed computing basics for big data

One lakh answer sheets, one week

What is distributed computing?

Words you'll meet in this lesson

The five-step pattern

Distributed computing at three levels

Building blocks and shapes of distributed systems

Six building blocks

Two kinds of parallelism

Two common architectures

Count chai orders on three machines

1. Partition the data

   orders = ["chai", "samosa", "chai", "poha", "chai", "samosa"]
   blocks = [orders[0:2], orders[2:4], orders[4:6]]
   print(blocks)

   [['chai', 'samosa'], ['chai', 'poha'], ['chai', 'samosa']]
   In a real cluster, a 30 GB file becomes 240 blocks of 128 MB, and each block is stored on three different nodes.

2. Each machine counts only its block

   block = ["chai", "poha"]           # machine 2's block
   counts = {}
   for item in block:
       counts[item] = counts.get(item, 0) + 1
   print(counts)

   {'chai': 1, 'poha': 1}
   All three machines run this at the same time, each on the block stored on its own disk. Nothing is sent over the network yet: that's data locality.

3. Send back small results

   Machine 1 sends {'chai': 1, 'samosa': 1}, machine 2 sends {'chai': 1, 'poha': 1}, machine 3 sends {'chai': 1, 'samosa': 1}.

   Each block may be 128 MB, but its partial result is a few bytes: one number per item. The network carries the small summaries, not the data.

4. Combine

   partials = [{"chai": 1, "samosa": 1}, {"chai": 1, "poha": 1}, {"chai": 1, "samosa": 1}]
   total = {}
   for p in partials:
       for item, n in p.items():
           total[item] = total.get(item, 0) + n
   print(total)

   {'chai': 3, 'samosa': 2, 'poha': 1}
   The same answer as counting all six orders on one machine. Counts combine by simple addition, which is why counting is the classic first distributed program.

5. A machine fails

   Machine 2 crashes halfway through its block.

   The master notices that machine 2 stopped reporting (it misses its regular heartbeat). Another node holds a copy of block 2, so the master re-runs just that one task there. Blocks 1 and 3 are not redone.

A distributed job: what Riya sees, and what the cluster does

Step through the six frames. Left: the job's progress page. Right: the master and four workers.

What can go wrong in a distributed system

Real-time scenarios: a cluster at work

Three live scenarios: whether to move the data or the program, a job held up by one giant partition, and an advanced one about a worker that goes silent and then comes back.

Move the data, or move the program?

A worker needs to process a 100 GB block stored on another machine. It could copy the data over the network, or the program could be sent to the machine that has the data. Watch the difference.

Data skew: one giant partition holds up the whole job

TiffinGo's daily report splits orders by city, one city per worker. Mumbai has six times more orders than any other city. Watch the progress bars, then see two standard fixes.

Slow or dead? The worker that came back

A cluster has no way to look inside a silent machine. Watch how heartbeats and timeouts handle a worker that goes quiet, and the surprising bug that appears when it comes back.

Where distributed computing runs every day

Examples are simplified to show the pattern. Real systems add many more details.

A tiny cluster in Python

Twelve canteen orders, three “machines”. The program partitions the data, lets each machine count only its own block, then combines the partial counts. On a real cluster the three counts would run at the same time on different computers. Then try the challenges: real parallel speed-up, a failure, Amdahl's law, a straggler and the average-of-averages trap.

orders = ["chai", "samosa", "chai", "vada pav", "chai", "poha",
          "samosa", "chai", "coffee", "chai", "samosa", "poha"]
machines = 3

# 1. PARTITION: split the data into one block per machine
size = len(orders) // machines
blocks = [orders[i * size:(i + 1) * size] for i in range(machines)]

# 2. PROCESS: each machine counts only its own block
def count_block(block):
    counts = {}
    for item in block:
        counts[item] = counts.get(item, 0) + 1
    return counts

partials = [count_block(b) for b in blocks]
for i, p in enumerate(partials, 1):
    print("Machine", i, "->", p)

# 3. COMBINE: merge the small partial results
total = {}
for p in partials:
    for item, n in p.items():
        total[item] = total.get(item, 0) + n
print("Final     ->", total)

• orders[i * size:(i + 1) * size]Slicing gives machine 1 items 0–3, machine 2 items 4–7, machine 3 items 8–11: one partition each.
• def count_block(block):The same small function is sent to every machine. Shipping code to the data is data locality.
• partials = [count_block(b) for b in blocks]Here the calls run one after another. On a cluster each call runs on a different node at the same time.
• total[item] = total.get(item, 0) + nCounts combine by adding. That is why counting is the classic first distributed program.

Run it on your PC: use all your CPU cores

A plain Python loop uses one core. This program counts prime numbers below 20 lakh, first on one core, then split into blocks and run on all your cores with multiprocessing. It is a real parallel program on one machine, using the same split–process–combine pattern a cluster uses across machines. Save it as use_all_cores.py.

import os, time
from multiprocessing import Pool


def count_primes(block):
    """Count the prime numbers in one block (a range of numbers)."""
    lo, hi = block
    count = 0
    for n in range(max(lo, 2), hi):
        if all(n % d for d in range(2, int(n ** 0.5) + 1)):
            count += 1
    return count


if __name__ == "__main__":           # needed on Windows for multiprocessing
    cores = os.cpu_count()
    N = 2_000_000
    step = N // cores
    blocks = [(i * step, (i + 1) * step) for i in range(cores)]

    start = time.time()
    serial = sum(count_primes(b) for b in blocks)
    t1 = time.time() - start
    print(f"Serial   (1 core) : {serial} primes in {t1:.1f} s")

    start = time.time()
    with Pool(cores) as pool:
        parallel = sum(pool.map(count_primes, blocks))
    t2 = time.time() - start
    print(f"Parallel ({cores} cores): {parallel} primes in {t2:.1f} s")
    print(f"Speed-up: {t1 / t2:.1f}x")

    # Big numbers take longer to test, so the last block is the slowest.
    # Many small blocks spread the work more evenly between the cores.
    small = N // (cores * 8)
    many = [(i * small, (i + 1) * small) for i in range(cores * 8)]
    start = time.time()
    with Pool(cores) as pool:
        balanced = sum(pool.map(count_primes, many))
    t3 = time.time() - start
    print(f"Parallel, {len(many)} small blocks: {balanced} primes in {t3:.1f} s, speed-up {t1 / t3:.1f}x")

# In a terminal:  python use_all_cores.py

Partition, count, combine, one line at a time

The mini cluster with six orders and two machines. Watch count_block() get its own frame for each block and return a small dictionary, then watch the partial results merge into total.

Worked examples: real problems, solved and traced in memory

Four problems, from beginner to advanced: splitting fairly, combining correctly, recovering from failures and fixing a skewed partition. Each runs live and is traced in memory.

records R1..R10, 3 machines

Machine 1 : ['R1', 'R2', 'R3', 'R4']
Machine 2 : ['R5', 'R6', 'R7']
Machine 3 : ['R8', 'R9', 'R10']

A: 8 × 100 · B: 2 × 1000

Partials: [(800, 8), (2000, 2)]
True average: 280.0

T1 done, T2 done, T3 failed, T4 done · healthy workers W1, W2

Retrying T3 on W1
All done: True

Pune 2, Delhi 2, Chennai 2, Kolkata 2, Mumbai 12 (GB)

Hot key: Mumbai | skew ratio: 3.0
Partitions now: 10 | largest: 2 GB

Check yourself

Quick revision

Interview and viva questions

Practice problems

Split–process–combine exercises. The last three are business tasks about time, cost and correctness.

[1..8]

[[1, 2, 3], [4, 5, 6], [7, 8]]

nums = list(range(1, 9))
chunks = [nums[i:i + 3] for i in range(0, len(nums), 3)]
print(chunks)

two dictionaries

{'chai': 6, 'poha': 1, 'samosa': 3}

p1 = {"chai": 4, "poha": 1}
p2 = {"chai": 2, "samosa": 3}
total = dict(p1)
for item, n in p2.items():
    total[item] = total.get(item, 0) + n
print(total)

[1200, 950, 1430, 880]

Total: 4460

print("Total:", sum([1200, 950, 1430, 880]))

blocks [120, 450, 80], [990, 30], [300, 710]

Local: [450, 990, 710]
Global max: 990

blocks = [[120, 450, 80], [990, 30], [300, 710]]
local_max = [max(b) for b in blocks]
print("Local:", local_max)
print("Global max:", max(local_max))

p = 0.95, n = 20

Speed-up: 10.26
Limit: 20 x

p, n = 0.95, 20
print("Speed-up:", round(1 / ((1 - p) + p / n), 2))
print("Limit:", round(1 / (1 - p)), "x")

now = 1000 s

Dead workers: ['W3', 'W5']
Tasks to re-run: [4, 5, 7]

now = 1000
last_beat = {"W1": 998, "W2": 990, "W3": 955, "W4": 999, "W5": 960}
tasks = {"W1": [1, 2], "W2": [3], "W3": [4, 5], "W4": [6], "W5": [7]}
TIMEOUT = 30
dead = [w for w, t in last_beat.items() if now - t > TIMEOUT]
rerun = [task for w in dead for task in tasks[w]]
print("Dead workers:", dead)
print("Tasks to re-run:", rerun)

results (task, attempt, orders)

Naive total  : 670
Correct total: 435

results = [(1, "a", 120), (2, "a", 95), (3, "a", 140),
           (3, "b", 140), (4, "a", 80), (2, "b", 95)]
committed = {}
for task, attempt, orders in results:
    if task not in committed:
        committed[task] = (attempt, orders)
naive = sum(r[2] for r in results)
correct = sum(o for _, o in committed.values())
print("Naive total  :", naive)
print("Correct total:", correct)

500 GB, 1250 MB/s

Copying data: 6.8 minutes

mb = 500 * 1024
print("Copying data:", round(mb / 1250 / 60, 1), "minutes")

[4, 5, 4, 21, 5, 4]

Job time: 21 minutes
Average task: 7.2 minutes

times = [4, 5, 4, 21, 5, 4]
print("Job time:", max(times), "minutes")
print("Average task:", round(sum(times) / len(times), 1), "minutes")

store ids with overlaps

Sum of counts (wrong): 8
Distinct customers   : 5

stores = [{"C1", "C2", "C3"}, {"C2", "C4"}, {"C1", "C4", "C5"}]
print("Sum of counts (wrong):", sum(len(s) for s in stores))
print("Distinct customers   :", len(set().union(*stores)))

Real-world scenario questions (asked in company interviews)

Distributed-computing problems that show up in real jobs and interviews. Decide your answer first, then open the model answer.