The life of a YARN application

Thirty-six seconds, eighteen state changes

The YARN application lifecycle

Words you'll meet in this lesson

From our lab (Hadoop 3.5.0, ItemRevenue on 140 MB)

Architecture at a glance

Nine steps take one application from your terminal to its result. The numbers match the steps in Part 4.

The lifecycle at three levels

The three state machines

YARN tracks the same job at three levels. Each has its own states, and each appears in a different log.

How an application can end

Animated example: results day at a university board

The same states, on a different job: a board computes 12 lakh students' results overnight. Times are from our lab run (06:25:43 → 06:26:25); only the story around them is invented.

Following one application on the lab cluster

We use the Module 8 lab (HDFS + YARN, three workers) and the 140 MB orders_x30.csv. app_states.py polls the RM REST API every 0.2 s; docker logs shows the RM and NM logs. Both are in “Run it on your PC”.

1. Steps 1–3: the client asks for an ID, stages files, submits

   hadoop jar mr.jar ItemRevenue /user/hadoop/data/orders_x30.csv /out/lc1 &
   python3 app_states.py application_1792643020992_0136

   submitted application_1792643020992_0136 + 0.1 s state=ACCEPTED final=UNDEFINED containers=1 allocatedMB=512
   By the time our first poll arrived, the application had already passed NEW, NEW_SAVING and SUBMITTED. ACCEPTED with 1 container and 512 MB means the scheduler has already reserved the ApplicationMaster's container.

2. Step 4: the ResourceManager records and schedules

   docker logs lab8-resourcemanager-1 | grep application_1792643020992_0136

   06:25:43 RMAppImpl - application_…_0136 State change from NEW to NEW_SAVING on event = START 06:25:43 RMAppImpl - application_…_0136 State change from NEW_SAVING to SUBMITTED on event = APP_NEW_SAVED 06:25:43 RMAppImpl - application_…_0136 State change from SUBMITTED to ACCEPTED on event = APP_ACCEPTED 06:25:43 RMAppAttemptImpl - appattempt_…_0136_000001 State change from SCHEDULED to ALLOCATED_SAVING on event = CONTAINER_ALLOCATED 06:25:43 RMAppAttemptImpl - appattempt_…_0136_000001 State change from ALLOCATED to LAUNCHED on event = LAUNCHED
   Every “_SAVING” state is a write to the RM's state store, so a restarted RM can recover. The scheduler found room for the AM in the same second.

3. Step 5: the NodeManager localizes and launches the AM

   docker logs lab8-worker2-1 | grep container_1792643020992_0136_01_000001

   06:25:43 ContainerImpl - Container container_…_0136_01_000001 transitioned from NEW to LOCALIZING 06:25:43 ContainerImpl - Container container_…_0136_01_000001 transitioned from LOCALIZING to SCHEDULED 06:25:43 ContainerImpl - Container container_…_0136_01_000001 transitioned from SCHEDULED to RUNNING
   LOCALIZING copies the job jar and job.xml from the HDFS staging folder to the worker's disk. Container _000001 of attempt _01 is always the ApplicationMaster.

4. Step 6: the ApplicationMaster registers

   06:25:48 appattempt_…_000001 State change from LAUNCHED to RUNNING on event = REGISTERED
   06:25:48 application_…_0136 State change from ACCEPTED to RUNNING on event = ATTEMPT_REGISTERED
   +  5.0 s  state=RUNNING    final=UNDEFINED containers=1  allocatedMB=512

   The five seconds between LAUNCHED and REGISTERED are the MRAppMaster JVM starting and reading the job configuration. Only a registered AM may send allocate requests.

5. Steps 7–8: the AM gets containers and runs tasks

   +  6.0 s  state=RUNNING    final=UNDEFINED containers=2  allocatedMB=1024
   + 27.1 s  state=RUNNING    final=UNDEFINED containers=1  allocatedMB=512
   + 28.1 s  state=RUNNING    final=UNDEFINED containers=2  allocatedMB=1024

   One second after registering, the AM held a second container: the single map task (one 140 MB split, Lesson 8.2). When the map finished at +27 s its container was released, and a reduce container arrived at +28 s. The RM logged no application state changes in this phase: tasks are the AM's business.

6. Step 9: unregister, clean up, send the bill

   yarn application -status application_1792643020992_0136

   + 35.6 s state=FINISHED final=SUCCEEDED containers=1 allocatedMB=512 Application-Type : MAPREDUCE Queue : root.default State : FINISHED Final-State : SUCCEEDED AM Host : worker2 Aggregate Resource Allocation : 33224 MB-seconds, 64 vcore-seconds
   The AM unregistered at 06:26:19 (FINAL_SAVING, FINISHING); its container exited at 06:26:25 (EXITED_WITH_SUCCESS, DONE) and the app became FINISHED. 33,224 MB-seconds over about 36 s means roughly 920 MB held on average: the AM all the time, plus one task container most of the time.

Two ways it goes differently

# kill the second run 15 s after submitting it
yarn application -kill application_1792643020992_0137
# ask for a 3 GB ApplicationMaster on 2 GB NodeManagers
hadoop jar mr.jar ItemRevenue -D yarn.app.mapreduce.am.resource.mb=3072 …

A kill releases every container at once and records who killed it and from where. An impossible request is rejected at submission, so no application ever reaches ACCEPTED.

From Enter to FINISHED

Six steps of our 36-second run: what the polling script printed, and what the daemons were doing.

The lifecycle at work

Simplified; organisations differ in tools and policies.

Simulate the application and attempt state machines

A small model of YARN's RMApp and RMAppAttempt state machines. Each event moves an object to its next state, exactly as the lines in the ResourceManager log say. Run it and compare with our real log.

APP = {("NEW", "START"): "NEW_SAVING", ("NEW_SAVING", "APP_NEW_SAVED"): "SUBMITTED",
       ("SUBMITTED", "APP_ACCEPTED"): "ACCEPTED", ("ACCEPTED", "ATTEMPT_REGISTERED"): "RUNNING",
       ("RUNNING", "ATTEMPT_UNREGISTERED"): "FINAL_SAVING", ("FINAL_SAVING", "APP_UPDATE_SAVED"): "FINISHING",
       ("FINISHING", "ATTEMPT_FINISHED"): "FINISHED"}
ATTEMPT = {("NEW", "START"): "SUBMITTED", ("SUBMITTED", "ATTEMPT_ADDED"): "SCHEDULED",
           ("SCHEDULED", "CONTAINER_ALLOCATED"): "ALLOCATED_SAVING", ("ALLOCATED_SAVING", "ATTEMPT_NEW_SAVED"): "ALLOCATED",
           ("ALLOCATED", "LAUNCHED"): "LAUNCHED", ("LAUNCHED", "REGISTERED"): "RUNNING",
           ("RUNNING", "UNREGISTERED"): "FINAL_SAVING", ("FINAL_SAVING", "ATTEMPT_UPDATE_SAVED"): "FINISHING",
           ("FINISHING", "CONTAINER_FINISHED"): "FINISHED"}
state = {"app": "NEW", "attempt": "NEW"}

def fire(obj, event, time):
    table = APP if obj == "app" else ATTEMPT
    new = table[(state[obj], event)]          # an event that isn't allowed here raises KeyError
    print(f"{time}  {obj:7} {state[obj]:>16} -> {new:16} on {event}")
    state[obj] = new

for t, obj, ev in [("06:25:43", "app", "START"), ("06:25:43", "app", "APP_NEW_SAVED"), ("06:25:43", "app", "APP_ACCEPTED"),
                   ("06:25:43", "attempt", "START"), ("06:25:43", "attempt", "ATTEMPT_ADDED"),
                   ("06:25:43", "attempt", "CONTAINER_ALLOCATED"), ("06:25:43", "attempt", "ATTEMPT_NEW_SAVED"),
                   ("06:25:43", "attempt", "LAUNCHED"), ("06:25:48", "attempt", "REGISTERED"),
                   ("06:25:48", "app", "ATTEMPT_REGISTERED"), ("06:26:19", "attempt", "UNREGISTERED"),
                   ("06:26:19", "attempt", "ATTEMPT_UPDATE_SAVED"), ("06:26:19", "app", "ATTEMPT_UNREGISTERED"),
                   ("06:26:19", "app", "APP_UPDATE_SAVED"), ("06:26:25", "attempt", "CONTAINER_FINISHED"),
                   ("06:26:25", "app", "ATTEMPT_FINISHED")]:
    fire(obj, ev, t)

• new = table[(state[obj], event)] # an event that isn't allowed here raises KeyErrorA state machine only accepts certain events in each state. YARN logs “Can't handle this event at current state” in that case.
• ("06:25:43", "attempt", "LAUNCHED"), ("06:25:48", "attempt", "REGISTERED"),The 5-second gap: the MRAppMaster JVM starting up on worker2.
• ("06:25:48", "app", "ATTEMPT_REGISTERED"), ("06:26:19", "attempt", "UNREGISTERED"),Between these two lines the RM logs nothing about the app: the AM runs its own tasks.

Run it on your PC: watch, kill and break an application

Use the Module 8 lab (Lesson 8.1). Copy app_states.py and lifecycle_lab.sh into the namenode container (docker cp … lab8-namenode-1:/tmp/mr/) and run the script there. It runs ItemRevenue on orders_x30.csv while polling the ResourceManager, kills a second run, and asks for an impossible container. The RM and NM logs are read from your PC with docker logs.

# app_states.py: submit-and-watch. Polls the ResourceManager REST API every 0.2 s and prints every
# change of application state, plus how many containers the app holds. Usage: python3 app_states.py <app_id>
import json, sys, time, urllib.request
RM = "http://resourcemanager:8088/ws/v1/cluster/apps/"
app, t0, last = sys.argv[1], time.time(), None
while True:
    try:
        a = json.load(urllib.request.urlopen(RM + app))["app"]
    except Exception:
        time.sleep(0.2); continue
    now = (a["state"], a.get("finalStatus"), a.get("runningContainers", 0), a.get("allocatedMB", 0))
    if now != last:
        print(f"+{time.time() - t0:5.1f} s  state={now[0]:10} final={now[1]:9} containers={max(now[2], 0)}  allocatedMB={max(now[3], 0)}")
        last = now
    if a["state"] in ("FINISHED", "FAILED", "KILLED"):
        print("diagnostics:", (a.get("diagnostics") or "-")[:150])
        break
    time.sleep(0.2)

# Used by lifecycle_lab.sh:  python3 /tmp/mr/app_states.py application_<id>

# lifecycle_lab.sh: watch a YARN application move through its states, then kill one
cd /tmp/mr
run_and_watch() {   # output dir, extra args
  hdfs dfs -rm -r -f -skipTrash $1 > /dev/null 2>&1
  hadoop jar mr.jar ItemRevenue $2 /user/hadoop/data/orders_x30.csv $1 > /tmp/lc.log 2>&1 &
  until app=$(grep -o "application_[0-9_]*" /tmp/lc.log | head -1) && [ -n "$app" ]; do sleep 0.1; done
  echo "submitted $app"
  python3 /tmp/mr/app_states.py $app
}
echo "== 1. a normal job"
run_and_watch /out/lc1 ""
wait
echo "-- the same app as YARN reports it afterwards"
yarn application -status $(grep -o "application_[0-9_]*" /tmp/lc.log | head -1) 2>/dev/null | grep -E "Application-Type|Queue|State|Final-State|Start-Time|Finish-Time|Progress|AM Host|Aggregate Resource" | sed 's/^\s*//'
echo "== 2. kill a job while it runs"
hdfs dfs -rm -r -f -skipTrash /out/lc2 > /dev/null 2>&1
hadoop jar mr.jar ItemRevenue /user/hadoop/data/orders_x30.csv /out/lc2 > /tmp/lc2.log 2>&1 &
until app=$(grep -o "application_[0-9_]*" /tmp/lc2.log | head -1) && [ -n "$app" ]; do sleep 0.1; done
python3 /tmp/mr/app_states.py $app &
sleep 15; yarn application -kill $app 2>&1 | grep -v INFO
wait
grep -E "Job .* failed|killed|KILLED" /tmp/lc2.log | sed -E 's/^[0-9-]+ [0-9:]+ INFO  //' | head -2
echo "== 3. ask for an AM bigger than any node (3 GB on 2 GB NodeManagers)"
hadoop jar mr.jar ItemRevenue -D yarn.app.mapreduce.am.resource.mb=3072 /user/hadoop/data/canteen_orders.csv /out/lc3 2>&1 | grep -E "Exception|Invalid" | head -2 | cut -c1-250

# Run:  docker exec lab8-namenode-1 bash /tmp/mr/lifecycle_lab.sh

# In the mr-lab folder, after lifecycle_lab.sh (use your own application id):
app=application_1792643020992_0136; id=${app#application_}
docker logs lab8-resourcemanager-1 2>&1 | grep "$app\|appattempt_$id" | grep "State change"
docker logs lab8-worker2-1 2>&1 | grep "container_${id}_01_000001 transitioned"
# worker2 is the AM host shown by yarn application -status

A container's life on the NodeManager

The NodeManager keeps its own small state machine for each container. Follow the ApplicationMaster's container _000001 through the five transitions we saw in the worker2 log.

Worked examples: real problems, solved and traced in memory

Four lifecycle problems from different organisations: billing, waiting, retries and kills.

40,000 × 33,224 MB-s · ₹5 per GB-hour

1,328,960,000 MB-s = 360.50 GB-hours -> Rs 1,803

start/finish in ms · total 33,224 MB-s · AM 512 MB

elapsed 36.2 s, AM 18,531 MB-s, tasks 14,693 MB-s = 28.7 s of a 512 MB container

p(AM node fails) = 0.05 per attempt

max-attempts=1: fails 5.00% of the time
max-attempts=2: fails 0.25% of the time

deadline 06:00 · next job 3.5 h at 100%, longer with less

kill now, start 02:00 at 100%  done at 05:30 OK
start now beside it at 70%     done at 07:00 LATE
wait, start 03:00 at 100%      done at 06:30 LATE

Check yourself

Quick revision

Interview and viva questions

Practice problems

Lifecycle exercises built on the real states and numbers from our lab.

7 + 10 + 5

22 state changes for one 36-second job

print(7 + 10 + 5, "state changes for one 36-second job")

two timestamps

5 s

from datetime import datetime
f = "%H:%M:%S"
print((datetime.strptime("06:25:48", f) - datetime.strptime("06:25:43", f)).seconds, "s")

33,224 MB-s · 36.2 s

918 MB on average

print(round(33224 / 36.2), "MB on average")

one log line

('application_1792643020992_0136', 'ACCEPTED', 'RUNNING', 'ATTEMPT_REGISTERED')

import re
line = "06:25:48 RMAppImpl:921 - application_1792643020992_0136 State change from ACCEPTED to RUNNING on event = ATTEMPT_REGISTERED"
m = re.search(r"(\S+) State change from (\S+) to (\S+) on event = (\S+)", line)
print(m.groups())

3 apps from the RM REST API

0136 0.1 s
0137 0.1 s
0138 60.5 s

apps = {"0136": (1792650343192, 1792650343300), "0137": (1792650380000, 1792650380050), "0138": (1792650400000, 1792650460500)}
for a, (s, l) in apps.items():
    print(a, f"{(l - s) / 1000:.1f} s")

small table

False

allowed = {"ACCEPTED": {"RUNNING", "FAILED", "KILLED", "FINAL_SAVING"}, "RUNNING": {"FINAL_SAVING", "KILLED", "FAILED"}}
print("FINISHED" in allowed["ACCEPTED"])

3,000 × 1,536 MB × 40 s

50 GB-hours a day, Rs 75 a month

gb_h = 3000 * 1536 * 40 / 1024 / 3600
print(f"{gb_h:.0f} GB-hours a day, Rs {gb_h * 0.05 * 30:,.0f} a month")

app transitions from the RM log

NEW            0 s
NEW_SAVING     0 s
SUBMITTED      0 s
ACCEPTED       5 s
RUNNING       31 s
FINAL_SAVING   0 s
FINISHING      6 s

from datetime import datetime
log = [("06:25:43", "NEW"), ("06:25:43", "NEW_SAVING"), ("06:25:43", "SUBMITTED"), ("06:25:43", "ACCEPTED"),
       ("06:25:48", "RUNNING"), ("06:26:19", "FINAL_SAVING"), ("06:26:19", "FINISHING"), ("06:26:25", "FINISHED")]
t = lambda s: datetime.strptime(s, "%H:%M:%S")
for (a, s1), (b, _) in zip(log, log[1:]):
    print(f"{s1:12} {(t(b) - t(a)).seconds:3d} s")

Real-world scenario questions (asked in company interviews)

The YARN application lifecycle is asked in Hadoop, Spark and platform-engineering interviews.