Move the code, not the data

The weekly report that took 22 hours

Data locality: move the code, not the data

Words you'll meet in this lesson

Why it matters: the numbers

Times are back-of-the-envelope estimates. Real networks are shared, and disks and protocols add overhead.

Locality at three levels

Three levels of locality

Ways systems put compute near data

How Hadoop sends one map task to its data

1. The job asks: where is the input?

   block_locations = {               # what the NameNode answers
       "blk_1": ["dn1", "dn4", "dn5"],
       "blk_2": ["dn2", "dn4", "dn6"],
   }
   print(block_locations["blk_1"])

   ['dn1', 'dn4', 'dn5']
   The client computes the input splits (normally one per block) and records the nodes that hold each one. This list travels with the job.

2. The ApplicationMaster asks for containers on those nodes

   For the task of blk_1, it asks YARN: “a container on dn1, dn4 or dn5, or failing that, on /rack1 or /rack2.”

   YARN resource requests can name specific hosts and racks. The scheduler tries to satisfy the most local one first.

3. Only the program travels

   The job jar (say 50 KB) and the configuration are copied to the chosen node, and the task JVM starts there.

   Hadoop puts job files in HDFS and each NodeManager downloads them once per job, shared by all tasks on that node. Even with 1,000 tasks, this is tiny compared with the input data.

4. The task reads its block from local disk

   The map task opens blk_1. Because dn1 itself holds a replica, the HDFS client reads it from dn1's disk.

   With short-circuit local reads enabled, the task reads the block file straight from disk without going through the DataNode's network socket, which is faster still.

5. Only small results leave the node

   The map task writes its output (for example counts per city) to dn1's local disk. Reducers fetch only these small outputs.

   128 MB of logs might become a few KB of counts. That is the shuffle: the only big network step of a MapReduce job, and much smaller than moving the input.

Same report, two ways: what Kabir sees, and what crosses the network

First the old way (copy all data to one server), then the Hadoop way (send the program to the data). Left: Kabir's terminal. Right: what moves across the cluster.

Try it yourself: locality on, locality off

Run the same job twice: once with Hadoop's normal locality-aware scheduling, and once with locality switched off, so tasks go to any free node and must pull their block over the network (red packets).

Where moving the code wins

Examples are simplified; figures are illustrative.

Move the data or move the code? Do the maths

Compare the time to bring all the data to one machine with the time to send a small program to every node and let each read its own share in parallel. Then try the challenges, including a tiny locality-aware scheduler.

data_tb = 10
nodes = 40
network_gbps = 10          # link into the single reporting server
local_read_gb_s = 1.0      # each node, many disks in parallel
jar_kb = 50

move_data_h = data_tb * 1024 * 8 / network_gbps / 3600
move_code_s = jar_kb * 8 / 1024 / 1024 / network_gbps
local_min = data_tb * 1024 / nodes / local_read_gb_s / 60

print(f"Move the data : {move_data_h:.1f} hours of transfer")
print(f"Move the code : {move_code_s:.5f} s to send the jar")
print(f"Then each node reads its share in {local_min:.1f} minutes")

• move_data_h = data_tb * 1024 * 8 / network_gbps / 3600TB → GB (× 1024), bytes → bits (× 8), ÷ link speed in Gbit/s = seconds, ÷ 3600 = hours.
• move_code_s = jar_kb * 8 / 1024 / 1024 / network_gbpsThe same formula for a 50 KB jar: KB → Gbit is ÷ 1024 ÷ 1024 after × 8.
• local_min = data_tb * 1024 / nodes / local_read_gb_s / 60Each node reads only its own 1/40th of the data, and all nodes read at the same time.

A locality-aware scheduler, one line at a time

Watch the scheduler give each task to a free node that holds its block, and fall back to any free node when none does. The free list shrinks as slots are used.

Worked examples: real problems, solved and traced in memory

Three locality problems from real cluster work: a transfer estimate, a scheduling decision, and a job report built from task placements. Each runs live and is traced in memory.

2 TB · 200 Mbps vs 1000 Mbps

Wi-Fi        : 23.3 hours
Wired 1 Gbps : 4.7 hours

M1 [dn1, dn3] · M2 [dn1, dn2] · M3 [dn1] · M4 [dn4] · free slots dn1 2, dn2 1, dn3 1

M1 -> dn1 node-local
M2 -> dn1 node-local
M3 -> dn2 remote
M4 -> dn3 remote

[node, node, rack, node, off, node, rack, node, node, off] · 128 MB per block

Data-local map tasks = 6
Rack-local map tasks = 2
Off-rack map tasks   = 2
Data-local share: 60%, input over network: 0.5 GB

Check yourself

Quick revision

Interview and viva questions

Practice problems

Transfer estimates and small scheduling programs. The business problems are the kind of questions asked when a team argues “just copy the data”.

5 TB · 10 Gbps

1.14 hours

print(round(5 * 1024 * 8 / 10 / 3600, 2), "hours")

200 KB · 4 TB

21,474,836 times bigger

data_kb = 4 * 1024 * 1024 * 1024
print(f"{data_kb / 200:,.0f} times bigger")

block on dn2 · rack {dn2: r1, dn3: r1, dn7: r2}

dn2 node-local
dn3 rack-local
dn7 off-rack

rack = {"dn2": "r1", "dn3": "r1", "dn7": "r2"}
for n in ["dn2", "dn3", "dn7"]:
    kind = "node-local" if n == "dn2" else "rack-local" if rack[n] == rack["dn2"] else "off-rack"
    print(n, kind)

client dn6 · replicas dn1, dn5, dn9 · racks r1: dn1-dn4, r2: dn5-dn8, r3: dn9

dn5

rack = {"dn1": "r1", "dn5": "r2", "dn6": "r2", "dn9": "r3"}
def dist(a, b):
    return 0 if a == b else 2 if rack[a] == rack[b] else 4
reps = ["dn1", "dn5", "dn9"]
print(min(reps, key=lambda r: dist("dn6", r)))

1250 tasks · 1180 data-local

94.4% data-local; target met: True

total, local = 1250, 1180
pct = local / total * 100
print(f"{pct:.1f}% data-local; target met: {pct >= 90}")

2000 × 128 MB · 3% vs 85% remote

with locality : 7.5 GB over the network
without       : 212.5 GB over the network

blocks, mb = 2000, 128
for label, remote in [("with locality", 0.03), ("without", 0.85)]:
    print(f"{label:14}: {blocks * remote * mb / 1024:.1f} GB over the network")

30 TB vs 2 GB · 10 Gbps

Copy all data   : 6.8 hours every night
Copy the result : 1.6 seconds
Recommendation  : aggregate on the cluster, move only the result

link = 10            # Gbps
full = 30 * 1024 * 8 / link / 3600
agg = 2 * 8 / link
print(f"Copy all data   : {full:.1f} hours every night")
print(f"Copy the result : {agg:.1f} seconds")
print("Recommendation  : aggregate on the cluster, move only the result")

600 readers · at most 20 per replica

Replicas needed: 30

import math
readers, limit = 600, 20
print("Replicas needed:", math.ceil(readers / limit))

Real-world scenario questions (asked in company interviews)

Locality questions that interviewers use to separate people who have memorised definitions from people who understand clusters.