YARN architecture: ResourceManager, NodeManager, ApplicationMaster

Who decides where your code runs?

YARN: Yet Another Resource Negotiator

Words you'll meet in this lesson

From our lab (Hadoop 3.5.0, Module 8 cluster)

YARN at three levels

The parts of YARN, one by one

Architecture at a glance

The picture is our lab during the 422 MB job: the AM and one map on worker1, two maps on worker2, one on worker3, and 1,536 MB still free on worker3.

Not only MapReduce: the ApplicationMasters you'll meet

Animated example: a 10 AM rush on a shared cluster

Looking inside a live YARN cluster

We use the Module 8 lab (NameNode, ResourceManager, JobHistory Server, three workers each running a DataNode and a NodeManager). All commands are in yarn_arch_lab.sh under “Run it on your PC”.

1. Which Java processes run where?

   for c in resourcemanager namenode historyserver worker1; do
     echo "$c: $(docker exec lab8-$c-1 jps | awk '{print $2}' | grep -v Jps)"
   done

   resourcemanager: ResourceManager namenode: NameNode RunJar historyserver: JobHistoryServer worker1: DataNode NodeManager
   Every daemon is a separate JVM. Masters run one daemon each; a worker runs both a DataNode (storage) and a NodeManager (compute) so tasks can read local blocks. RunJar on the namenode was the client of a job that was still finishing when we looked.

2. What does each NodeManager offer?

   docker exec lab8-resourcemanager-1 yarn node -list -showDetails

   worker2:40977 RUNNING worker2:8042 1 Configured Resources : <memory:2048, vCores:2> Allocated Resources : <memory:512, vCores:1> Resource Utilization by Node : PMem:5621 MB, VMem:5621 MB, VCores:1.6533333 Resource Utilization by Containers : PMem:466 MB, VMem:4334 MB, VCores:1.04 worker1:36427 RUNNING worker1:8042 0 Configured Resources : <memory:2048, vCores:2> Allocated Resources : <memory:0, vCores:0>
   “Configured” is what the NodeManager offers to YARN (from yarn-site.xml), not the machine's real RAM: the node itself was using 5.6 GB, mostly for the DataNode and other processes. One 512 MB container was still running on worker2.

3. Ask the ResourceManager for cluster totals

   curl -s http://resourcemanager:8088/ws/v1/cluster/metrics

   activeNodes 3 totalMB 6144 totalVirtualCores 6 allocatedMB 512 appsRunning 1 appsCompleted 125 containersAllocated 1
   The RM's whole job is this bookkeeping: 3 × 2048 MB = 6144 MB and 3 × 2 = 6 vcores to hand out. The same numbers appear on the web UI at port 8088.

4. Start a job and look again

   hadoop jar mr.jar ItemRevenue /user/hadoop/data/orders_x90.csv /out/yarn_arch &
   docker exec lab8-worker1-1 jps

   1 DataNode 1 MRAppMaster 1 NodeManager 1 YarnChild allocatedMB 2560 of 6144 | allocatedVirtualCores 5 | containersAllocated 5 | appsRunning 1
   Two new processes on worker1: the job's MRAppMaster and a YarnChild (a map task). Five containers hold 2,560 MB: the AM's 512 MB plus four maps. Each map asked for 384 MB, but YARN rounds every request up to a multiple of the minimum allocation (256 MB), so each got 512 MB.

5. List the job's containers

   yarn applicationattempt -list application_1792643020992_0135
   yarn container -list appattempt_1792643020992_0135_000001

   container_1792643020992_0135_01_000001 RUNNING worker1:36427 container_1792643020992_0135_01_000005 RUNNING worker2:40977 container_1792643020992_0135_01_000004 RUNNING worker3:44495 container_1792643020992_0135_01_000003 RUNNING worker1:36427 container_1792643020992_0135_01_000002 RUNNING worker2:40977
   Container …_000001 is always the ApplicationMaster. The number in the middle (_01_) is the attempt: if the AM dies, YARN starts attempt 02 with new containers.

6. Per-node usage

   curl -s http://resourcemanager:8088/ws/v1/cluster/nodes

   worker2 used 1024 MB / 2048 MB, 2 vcores used, 2 containers worker1 used 1024 MB / 2048 MB, 2 vcores used, 2 containers worker3 used 512 MB / 2048 MB, 1 vcores used, 1 containers Job:1765 - Job job_1792643020992_0135 completed successfully
   The Scheduler spread the containers as NodeManagers heartbeated in. In an earlier run in this session worker1 even showed “3 vcores used” on a 2-vcore node: the default resource calculator only counts memory. Lesson 11.4 switches that on.

From hadoop jar to five containers

Six steps of our 422 MB job, as the terminal shows them and as YARN handles them.

YARN at work

Simplified, illustrative examples; organisations differ in tools and policies.

Be the ResourceManager for one job

This model keeps the RM's books for our lab: three NodeManagers with 2,048 MB each, a 256 MB minimum allocation, one AM of 512 MB and four maps asking for 384 MB. Containers are granted as NodeManagers heartbeat in, one per heartbeat, in the order they arrived in our run.

import math
MIN_ALLOC = 256                                   # yarn.scheduler.minimum-allocation-mb
nodes = {"worker1": 2048, "worker2": 2048, "worker3": 2048}
used = {n: 0 for n in nodes}

def normalize(mb):                                # round UP to a multiple of the minimum
    return math.ceil(mb / MIN_ALLOC) * MIN_ALLOC

requests = [("AM", 512)] + [(f"map{i}", 384) for i in range(4)]
heartbeats = ["worker1", "worker2", "worker3", "worker1", "worker2"]   # who reported free space next

for (name, ask), node in zip(requests, heartbeats):
    size = normalize(ask)
    if used[node] + size <= nodes[node]:
        used[node] += size
        print(f"{name:5} asks {ask} MB -> gets {size} MB on {node}")

print("per node:", used)
print("allocated", sum(used.values()), "of", sum(nodes.values()), "MB in", len(requests), "containers")

• return math.ceil(mb / MIN_ALLOC) * MIN_ALLOCThe Scheduler normalizes every request: 384 MB becomes 512 MB, which is why 4 maps + AM held 2,560 MB.
• heartbeats = ["worker1", "worker2", "worker3", "worker1", "worker2"] # who reported free space nextThe Scheduler assigns containers when a NodeManager heartbeats and reports free space.
• if used[node] + size <= nodes[node]:A container must fit entirely on one node: YARN never splits a container across machines.

Run it on your PC: look inside YARN

This uses the Module 8 lab (docker compose -p lab8 up -d in the mr-lab folder) and the mr.jar and orders_x90.csv from Lesson 8.1. Unlike most scripts in Module 8, this one runs on your PC in the mr-lab folder, because it looks at several containers with docker exec.

# yarn_arch_lab.sh: look at YARN's parts while a job runs (run on your PC, in the mr-lab folder)
echo "== 1. Java processes (daemons) on each machine"
for c in resourcemanager namenode historyserver worker1; do
  echo "$c: $(docker exec lab8-$c-1 jps 2>/dev/null | awk '{print $2}' | grep -v Jps | sort | tr '\n' ' ')"
done
echo "== 2. what the ResourceManager knows about its NodeManagers"
docker exec lab8-resourcemanager-1 yarn node -list -showDetails 2>/dev/null | grep -E "Node-Id|worker|Resource Utilization|Configured Resources|Allocated" | head -12
echo "== 3. cluster metrics (RM REST API)"
docker exec lab8-namenode-1 curl -s http://resourcemanager:8088/ws/v1/cluster/metrics | python3 -c '
import sys, json; m = json.load(sys.stdin)["clusterMetrics"]
for k in ["activeNodes", "totalMB", "totalVirtualCores", "allocatedMB", "appsRunning", "appsCompleted", "containersAllocated"]: print(f"  {k:20} {m[k]}")'
echo "== 4. start a job and look again while it runs"
docker exec -d lab8-namenode-1 bash -c 'cd /tmp/mr9; hdfs dfs -rm -r -f -skipTrash /out/yarn_arch >/dev/null 2>&1; hadoop jar /tmp/mr/mr.jar ItemRevenue /user/hadoop/data/orders_x90.csv /out/yarn_arch > /tmp/yarn_arch.log 2>&1'
sleep 20
echo "-- processes on worker1 now:"
docker exec lab8-worker1-1 jps 2>/dev/null | awk '{print $2}' | grep -v Jps | sort | uniq -c
docker exec lab8-namenode-1 curl -s http://resourcemanager:8088/ws/v1/cluster/metrics | python3 -c '
import sys, json; m = json.load(sys.stdin)["clusterMetrics"]
print("  allocatedMB", m["allocatedMB"], "of", m["totalMB"], "| allocatedVirtualCores", m["allocatedVirtualCores"], "| containersAllocated", m["containersAllocated"], "| appsRunning", m["appsRunning"])'
app=$(docker exec lab8-resourcemanager-1 yarn application -list 2>/dev/null | grep -o "application_[0-9_]*" | head -1)
att=$(docker exec lab8-resourcemanager-1 yarn applicationattempt -list $app 2>/dev/null | grep -o "appattempt_[0-9_]*" | head -1)
echo "-- containers of $att:"
docker exec lab8-resourcemanager-1 yarn container -list $att 2>/dev/null | grep "^container_" | awk -F'\t' '{print "  " $1, $4, $5}' | tr -s ' ' | head -8
echo "-- per-node view (RM REST /ws/v1/cluster/nodes)"
docker exec lab8-namenode-1 curl -s http://resourcemanager:8088/ws/v1/cluster/nodes | python3 -c '
import sys, json
for n in json.load(sys.stdin)["nodes"]["node"]:
    print("  %-8s used %4d MB / %d MB, %d vcores used, %d containers" % (n["nodeHostName"], n["usedMemoryMB"], n["usedMemoryMB"] + n["availMemoryMB"], n["usedVirtualCores"], n["numContainers"]))'
docker exec lab8-namenode-1 bash -c 'while ! grep -qE "completed|failed" /tmp/yarn_arch.log; do sleep 2; done; grep -E "completed|Launched map" /tmp/yarn_arch.log | sed -E "s/^[0-9-]+ [0-9:]+ INFO  //"'

# Run on your PC, in the mr-lab folder:
#   bash yarn_arch_lab.sh

The Scheduler on each heartbeat

Watch the waiting requests and each node's free memory as three heartbeats arrive. A request is only granted on a heartbeat from a node with enough free space.

Worked examples: real problems, solved and traced in memory

Four YARN sizing and design problems from different industries.

10 nodes × 32 GB · 50 maps × 2 GB · AM 2 GB

16 per node, 160 in the cluster, 159 for maps -> one wave: True

400 × 1,100 MB · minimum 1,024 MB

each gets 2048 MB, waste 370 GB across 400 containers

heartbeat 1 s · expiry 600 s · 6 tasks × 4 min

RM marks the node LOST after 10 min; 24 task-minutes must be redone

30 × (56 GB, 16 vcores) · Spark + MR

memory 766/1680 GB fits: True
vcores 382/480 fits: True

Check yourself

Quick revision

Interview and viva questions

Practice problems

YARN architecture exercises using the real numbers from our cluster.

3 × (2048 MB, 2 vcores)

6144 MB, 6 vcores

print(3 * 2048, "MB,", 3 * 2, "vcores")

AM 512 · 4 × 384 → rounded

2560

import math
map_size = math.ceil(384 / 256) * 256
print(512 + 4 * map_size)

2,560 of 6,144 MB

42%

print(f"{2560 / 6144:.0%}")

5 requests

100 -> 256
256 -> 256
300 -> 512
600 -> 768
1100 -> 1280

import math
for ask in [100, 256, 300, 600, 1100]:
    print(ask, "->", math.ceil(ask / 256) * 256)

3 nodes, 4 heartbeats

granted on w2
granted on w3
granted on w2
still waiting: 1 {'w1': 512, 'w2': 0, 'w3': 0}

free = {"w1": 512, "w2": 2048, "w3": 1024}
waiting = 4
for node in ["w1", "w2", "w3", "w2"]:
    if waiting and free[node] >= 1024:
        free[node] -= 1024; waiting -= 1
        print("granted on", node)
print("still waiting:", waiting, free)

2 RMs + 50 NMs

52 YARN daemons (plus one AM per running application)

print(2 + 50, "YARN daemons (plus one AM per running application)")

3 × 20 nodes at 30%

separate: 60 nodes; shared average demand: 18 nodes -> a 30-node shared cluster leaves headroom

separate = 3 * 20
avg_demand = 3 * 20 * 0.3
print(f"separate: {separate} nodes; shared average demand: {avg_demand:.0f} nodes -> a 30-node shared cluster leaves headroom")

6144 MB · 10% · 512 MB AMs

AM limit 614 MB -> 1 AM(s) at once (YARN always allows at least one)

limit = 6144 * 0.1
print(f"AM limit {limit:.0f} MB -> {int(limit // 512)} AM(s) at once (YARN always allows at least one)")

Real-world scenario questions (asked in company interviews)

YARN architecture questions come up in every Hadoop and Spark interview.