Schedulers: FIFO, Capacity and Fair

A ten-second report that waited two minutes

The YARN scheduler

Words you'll meet in this lesson

Architecture at a glance

The queue tree of our lab, and where its containers land. The scheduler walks this tree every time a NodeManager reports free space.

From our lab (Hadoop 3.5.0, 3 × 2048 MB, SleepJob: big = 30 × 40 s, small = 2 × 5 s, 15 s later)

Schedulers at three levels

The three schedulers compared

The settings, in the files our lab used

yarn.resourcemanager.scheduler.class = ...scheduler.capacity.CapacityScheduler
yarn.scheduler.capacity.root.queues = prod,dev
yarn.scheduler.capacity.root.prod.capacity = 70
yarn.scheduler.capacity.root.dev.capacity = 30
yarn.scheduler.capacity.root.dev.maximum-capacity = 100
yarn.scheduler.capacity.root.dev.user-limit-factor = 4     # only in the "ULF 4" runs
yarn.resourcemanager.scheduler.monitor.enable = true       # only in the "preemption" runs

<allocations>
  <queue name="prod"><weight>1.0</weight><schedulingPolicy>fair</schedulingPolicy></queue>
  <queue name="dev"><weight>1.0</weight><schedulingPolicy>fair</schedulingPolicy></queue>
</allocations>

Animated example: the 9 AM regulatory report at a bank

A bank must send a regulatory report by 9:00 every morning, from the same cluster where analysts run ad-hoc queries. Watch the same morning under three set-ups. (A story with round numbers, built on what our lab showed.)

How an application lands in a queue

Six runs, one pair of jobs

Each run recreates only the ResourceManager with another scheduler (rm.sh); HDFS and the NodeManagers keep running and re-register. Then sched_lab.sh starts the big job, waits 15 s and starts the small one. Both are Hadoop's built-in SleepJob, so the work is identical every time.

1. FIFO: the head of the line takes everything

   bash rm.sh fifo
   docker exec lab8-namenode-1 bash /tmp/mr/sched_lab.sh default default

   scheduler: scheduler.fifo.FifoScheduler big queue=default AM waited 0.6 s, job took 198.8 s, SUCCEEDED, preempted containers: 0 small queue=default AM waited 120.4 s, job took 159.6 s, SUCCEEDED, preempted containers: 0
   The small job couldn't even start its ApplicationMaster for two minutes: FIFO gives every free container to the oldest app that still wants one.

2. Capacity with queues: the small job flies, the big one crawls

   bash rm.sh capacity
   docker exec lab8-namenode-1 bash /tmp/mr/sched_lab.sh dev prod

   Queue Name : prod Capacity: 70.0, MaximumCapacity: 100.0, CurrentCapacity: 0.0 Queue Name : dev Capacity: 30.000002, MaximumCapacity: 100.0, CurrentCapacity: 0.0 big queue=root.dev AM waited 0.6 s, job took 430.0 s, SUCCEEDED, preempted containers: 0 small queue=root.prod AM waited 0.1 s, job took 24.2 s, SUCCEEDED, preempted containers: 0
   Prod had room the whole time. But the big job took more than twice as long as under FIFO, even though dev's maximum is 100%.

3. Why so slow? Measure the big job alone

   curl -s http://resourcemanager:8088/ws/v1/cluster/metrics   # allocatedMB, containersAllocated

   capacity allocated 2560 of 6144 MB, containers 5 capacity allocated 2560 of 6144 MB, containers 5 fair allocated 6144 of 6144 MB, containers 6
   Under Capacity the job never grew past about 30% (5 containers of 512 MB) with the rest of the cluster idle. The cause is user-limit-factor, default 1: a single user may use at most 1 × the queue's capacity. Under Fair it took all 6,144 MB, but in only 6 containers: Fair rounds requests up to its 1,024 MB increment, so each 384 MB map got a 1 GB container.

4. Capacity with user-limit-factor 4: elastic again

   bash rm.sh capacity-ulf      # adds root.dev.user-limit-factor=4
   docker exec lab8-namenode-1 bash /tmp/mr/sched_lab.sh dev prod

   big queue=root.dev AM waited 0.7 s, job took 207.6 s, SUCCEEDED, preempted containers: 0 small queue=root.prod AM waited 51.0 s, job took 121.5 s, SUCCEEDED, preempted containers: 0
   Dev borrowed the whole idle cluster, so the big job is back to FIFO speed. The price: when prod's job arrived, every container was busy, and the scheduler could only wait for dev's 40-second tasks to finish.

5. Add preemption: borrowed containers are taken back

   bash rm.sh capacity-ulf-preempt   # + yarn.resourcemanager.scheduler.monitor.enable=true
   docker exec lab8-namenode-1 bash /tmp/mr/sched_lab.sh dev prod

   big queue=root.dev AM waited 0.7 s, job took 215.4 s, SUCCEEDED, preempted containers: 2 small queue=root.prod AM waited 20.7 s, job took 77.9 s, SUCCEEDED, preempted containers: 0
   The preemption monitor (ProportionalCapacityPreemptionPolicy) saw prod below its 70% guarantee, waited, then killed 2 of dev's containers. Prod's wait dropped from 51 s to 21 s; the big job paid 8 seconds for the redone work.

6. Fair Scheduler with two equal queues

   bash rm.sh fair
   docker exec lab8-namenode-1 bash /tmp/mr/sched_lab.sh dev prod

   scheduler: scheduler.fair.FairScheduler big queue=root.dev AM waited 1.0 s, job took 390.6 s, SUCCEEDED, preempted containers: 0 small queue=root.prod AM waited 37.5 s, job took 152.8 s, SUCCEEDED, preempted containers: 0
   Without preemption, Fair also waits for containers to finish, and with 1 GB containers the cluster ran only half as many tasks at once. Our lab's small maps simply don't suit Fair's default 1 GB increment (yarn.resource-types.memory-mb.increment-allocation).

The same 12 containers, three schedulers

Six steps through the real runs: what the terminal reported, and how the scheduler filled the cluster's 12 container slots.

Schedulers at work

Simplified; organisations differ in tools and policies.

Simulate the four schedulers

A small model of our lab: 12 container slots, the big job (30 tasks × 40 s) at time 0 and the small job (2 tasks × 5 s) at 15 s. Each ApplicationMaster takes a slot and needs about 5 s before it asks for task containers. It is a model, so the seconds differ from the lab, but the pattern is the same: FIFO hurts the small job, capacity with user-limit-factor 1 hurts the big job, preemption gets the small job going fastest without holding the big job back for long.

import math

def simulate(policy, slots=12):
    jobs = {"big":   dict(arrive=0,  left=30, secs=40, run=[], am=None, end=None, queue="dev"),
            "small": dict(arrive=15, left=2,  secs=5,  run=[], am=None, end=None, queue="prod")}
    if policy == "fair":
        slots //= 2                 # Fair rounds 384 MB maps up to 1 GB: half as many containers
    preempted = 0
    for t in range(1000):
        for j in jobs.values():     # tasks that finished free their slot
            j["run"] = [e for e in j["run"] if e > t]
            if j["am"] is not None and j["left"] == 0 and not j["run"] and j["end"] is None:
                j["end"] = t
        active = [n for n, j in jobs.items() if j["arrive"] <= t and j["end"] is None]
        used = lambda n: len(jobs[n]["run"]) + (jobs[n]["am"] is not None)
        free = slots - sum(used(n) for n in jobs)
        def limit(n):
            if policy == "capacity-ulf1" and jobs[n]["queue"] == "dev":
                return math.ceil(0.3 * slots) + 1       # one user <= 1 x 30% of the cluster
            if policy == "fair" and len(active) == 2:
                return slots // 2
            return slots
        def wants(n):
            j = jobs[n]
            return j["am"] is None or (j["left"] > 0 and t >= j["am"] + 5)
        if policy == "preempt" and t >= 30 and wants("small") and free == 0 and jobs["big"]["run"]:
            jobs["big"]["run"].pop(); jobs["big"]["left"] += 1; preempted += 1; free += 1
        order = active                                  # FIFO: order of arrival
        if policy in ("capacity-ulf4", "preempt"):
            order = sorted(active, key=lambda n: jobs[n]["queue"] != "prod")   # prod is under its guarantee
        if policy == "fair":
            order = sorted(active, key=lambda n: used(n))                       # furthest below fair share first
        for n in order:
            while free > 0 and wants(n) and used(n) < limit(n):
                if jobs[n]["am"] is None:
                    jobs[n]["am"] = t
                else:
                    jobs[n]["run"].append(t + jobs[n]["secs"]); jobs[n]["left"] -= 1
                free -= 1
            if policy == "fifo" and wants(n):
                break                                   # nobody passes the head of the line
        if all(j["end"] is not None for j in jobs.values()):
            break
    s = jobs["small"]
    return jobs["big"]["end"], s["am"] - s["arrive"], s["end"] - s["arrive"], preempted

for p in ["fifo", "capacity-ulf1", "capacity-ulf4", "preempt", "fair"]:
    big, wait, small, pre = simulate(p)
    print(f"{p:14} big {big:4d} s | small: AM waited {wait:3d} s, took {small:3d} s | preempted {pre}")

• return math.ceil(0.3 * slots) + 1 # one user <= 1 x 30% of the clusteruser-limit-factor 1: our big job alone held 2,560 of 6,144 MB (5 containers) in the lab.
• if policy == "preempt" and t >= 30 and wants("small") and free == 0 and jobs["big"]["run"]:Preemption: after a wait, take a container back from the over-using queue. In the lab 2 were preempted.
• break # nobody passes the head of the lineFIFO in one line: if the first app still wants containers, nobody behind it gets any.

Run it on your PC: switch schedulers and race two jobs

Use the Module 8 lab cluster. Put rm.sh, the sched-*.env files and fair-scheduler.xml in the mr-lab folder next to docker-compose.yaml. rm.sh recreates only the ResourceManager with another scheduler; HDFS and the NodeManagers keep running and re-register within a few seconds. Then copy sched_lab.sh into the namenode container and race the jobs. Each run takes 3–8 minutes.

# rm.sh fifo|capacity|capacity-preempt|capacity-ulf|capacity-ulf-preempt|fair|drf|default : recreate ONLY the ResourceManager with another scheduler
# (run in the mr-lab folder; HDFS and the NodeManagers keep running)
mode=$1
envs="./config"
case $mode in
  fifo) envs="$envs, ./sched-fifo.env" ;;
  capacity) envs="$envs, ./sched-capacity.env" ;;
  capacity-preempt) envs="$envs, ./sched-capacity.env, ./sched-capacity-preempt.env" ;;
  capacity-ulf) envs="$envs, ./sched-capacity.env, ./sched-capacity-ulf.env" ;;
  capacity-ulf-preempt) envs="$envs, ./sched-capacity.env, ./sched-capacity-ulf.env, ./sched-capacity-preempt.env" ;;
  fair) envs="$envs, ./sched-fair.env" ;;
  drf) envs="$envs, ./sched-drf.env" ;;
  default) ;;
esac
cat > rm-override.yaml <<Y
services:
  resourcemanager:
    env_file: [$envs]
    volumes: [./fair-scheduler.xml:/opt/fair-scheduler.xml:ro]
Y
docker compose -p lab8 -f docker-compose.yaml -f rm-override.yaml up -d resourcemanager 2>&1 | tail -1
until [ "$(docker exec lab8-resourcemanager-1 yarn node -list 2>/dev/null | grep -c RUNNING)" = 3 ]; do sleep 3; done
docker exec lab8-resourcemanager-1 bash -c 'echo "scheduler: $(grep -o "scheduler\.[a-z]*\.[A-Za-z]*Scheduler" /opt/hadoop/etc/hadoop/yarn-site.xml | tail -1)"; mapred queue -list 2>/dev/null | grep -E "Queue Name|Scheduling Info" | paste - - | sed "s/Scheduling Info : //"'

# The files rm.sh combines with ./config:
# ---- sched-fifo.env ----
# YARN-SITE.XML_yarn.resourcemanager.scheduler.class=org.apache.hadoop.yarn.server.resourcemanager.scheduler.fifo.FifoScheduler
# ---- sched-capacity.env ----
# YARN-SITE.XML_yarn.resourcemanager.scheduler.class=org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler
# CAPACITY-SCHEDULER.XML_yarn.scheduler.capacity.root.queues=prod,dev
# CAPACITY-SCHEDULER.XML_yarn.scheduler.capacity.root.prod.capacity=70
# CAPACITY-SCHEDULER.XML_yarn.scheduler.capacity.root.dev.capacity=30
# CAPACITY-SCHEDULER.XML_yarn.scheduler.capacity.root.dev.maximum-capacity=100
# CAPACITY-SCHEDULER.XML_yarn.scheduler.capacity.root.prod.maximum-capacity=100
# CAPACITY-SCHEDULER.XML_yarn.scheduler.capacity.maximum-am-resource-percent=0.5
# ---- sched-capacity-ulf.env ----
# CAPACITY-SCHEDULER.XML_yarn.scheduler.capacity.root.dev.user-limit-factor=4
# CAPACITY-SCHEDULER.XML_yarn.scheduler.capacity.root.prod.user-limit-factor=4
# ---- sched-capacity-preempt.env ----
# YARN-SITE.XML_yarn.resourcemanager.scheduler.monitor.enable=true
# CAPACITY-SCHEDULER.XML_yarn.scheduler.capacity.root.dev.disable_preemption=false
# ---- sched-fair.env ----
# YARN-SITE.XML_yarn.resourcemanager.scheduler.class=org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler
# YARN-SITE.XML_yarn.scheduler.fair.allocation.file=/opt/fair-scheduler.xml
# YARN-SITE.XML_yarn.scheduler.fair.user-as-default-queue=false
# ---- fair-scheduler.xml ----
# <?xml version="1.0"?>
# <!-- fair-scheduler.xml: two queues with equal weight; inside a queue, apps share fairly -->
# <allocations>
#   <queue name="prod"><weight>1.0</weight><schedulingPolicy>fair</schedulingPolicy></queue>
#   <queue name="dev"><weight>1.0</weight><schedulingPolicy>fair</schedulingPolicy></queue>
#   <queueMaxAMShareDefault>0.5</queueMaxAMShareDefault>
# </allocations>

# Run, in the mr-lab folder:
#   bash rm.sh capacity

# sched_lab.sh BIG_QUEUE SMALL_QUEUE : a big job fills the cluster, a small job arrives 15 s later. Who waits?
# Uses Hadoop's built-in "sleep" test job: tasks that just sleep, so timings are easy to read.
# Run inside the namenode container.
T=$(ls /opt/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-client-jobclient-*-tests.jar)
export HADOOP_CLASSPATH=$T
SLEEP="yarn org.apache.hadoop.mapreduce.SleepJob -libjars $T"
BQ=${1:-default}; SQ=${2:-default}
# big: 30 map tasks of 40 s each (the cluster runs about 11 at a time)
$SLEEP -D mapreduce.job.queuename=$BQ -D mapreduce.job.name=big -m 30 -r 1 -mt 40000 -rt 1000 > /tmp/s_big.log 2>&1 &
sleep 15
# small: 2 map tasks of 5 s each
$SLEEP -D mapreduce.job.queuename=$SQ -D mapreduce.job.name=small -m 2 -r 1 -mt 5000 -rt 1000 > /tmp/s_small.log 2>&1 &
wait
for j in big small; do
  app=$(grep -o "application_[0-9_]*" /tmp/s_$j.log | head -1)
  curl -s http://resourcemanager:8088/ws/v1/cluster/apps/$app | python3 -c '
import sys, json; a = json.load(sys.stdin)["app"]
print("%-5s queue=%-12s AM waited %5.1f s, job took %6.1f s, %s, preempted containers: %s" % (sys.argv[1], a["queue"],
      (a["launchTime"] - a["startedTime"]) / 1000, a["elapsedTime"] / 1000, a["finalStatus"], a.get("numNonAMContainerPreempted", 0)))' $j
done
grep -m1 -i "preempted" /tmp/s_big.log | sed -E 's/^[0-9-]+ [0-9:]+ INFO  //' | cut -c1-160

# In the mr-lab folder:
#   docker cp sched_lab.sh lab8-namenode-1:/tmp/mr/
#   bash rm.sh fifo     && docker exec lab8-namenode-1 bash /tmp/mr/sched_lab.sh default default
#   bash rm.sh capacity && docker exec lab8-namenode-1 bash /tmp/mr/sched_lab.sh dev prod
#   … capacity-preempt, fair, capacity-ulf, capacity-ulf-preempt the same way
# The "probe" lines: start only the big job in dev and read
#   curl -s http://resourcemanager:8088/ws/v1/cluster/metrics   every 15 s

How the Capacity Scheduler limits one user

The limit our big job hit, computed step by step: a queue's guaranteed share, its maximum, and the user limit that applies when one user is alone in the queue.

Worked examples: real problems, solved and traced in memory

Four scheduler design problems from different industries: sizing queues, guarantee vs elasticity, the user limit, and the cost of preemption.

60 × 96 GB · 50 / 20 / 30 · 4 GB containers

billing     2880 GB -> 720 containers of 4 GB
fraud       1152 GB -> 288 containers of 4 GB
analytics   1728 GB -> 432 containers of 4 GB

1,000 containers · idle 40% for 20 h

8,000 container-hours a day left idle

400 containers · capacity 25% · max 60%

user-limit-factor 1: up to 100 containers
user-limit-factor 2: up to 200 containers
user-limit-factor 3: up to 240 containers

50 containers × 6 min lost · 250 days

300 container-minutes a day, 1,250 container-hours a year
lab: big job 7.8 s slower, small job 43.6 s faster

Check yourself

Quick revision

Interview and viva questions

Practice problems

Scheduler exercises built on the six real runs.

159.6 s · 24.2 s · 10 s

FIFO: 16.0x the work time
Capacity: 2.4x the work time

for name, t in [("FIFO", 159.6), ("Capacity", 24.2)]:
    print(f"{name}: {t / 10:.1f}x the work time")

70% · 30%

prod 4300.799999999999 MB, dev 1843.1999999999998 MB

print("prod", 6144 * 0.70, "MB, dev", 6144 * 0.30, "MB")

430.0 · 207.6

222.4 s longer, 2.07x

print(f"{430.0 - 207.6:.1f} s longer, {430.0 / 207.6:.2f}x")

small 121.5 → 77.9 · big 207.6 → 215.4

small saved 43.6 s, big lost 7.8 s

print(f"small saved {121.5 - 77.9:.1f} s, big lost {215.4 - 207.6:.1f} s")

20 × 48 GB · 3 GB · 35%

112 of 320 containers

total = 20 * 48 // 3
print(int(total * 0.35), "of", total, "containers")

weights 3 : 1 · 200

{'prod': 150, 'adhoc': 50}
{'prod': 200}

for active in [{"prod": 3, "adhoc": 1}, {"prod": 3}]:
    t = sum(active.values())
    print({q: 200 * w // t for q, w in active.items()})

A: 10 × ₹6 L + 10 × ₹1.5 L/yr · B: 2% × 40% × 50 nodes

A: Rs 75 lakh in year 1, B: Rs 0.6 lakh a year of redone work

a = 10 * 6 + 10 * 1.5
b = 0.02 * 0.40 * 50 * 1.5
print(f"A: Rs {a:.0f} lakh in year 1, B: Rs {b:.1f} lakh a year of redone work")

a nested dict of capacities

root level: 100
prod children: 100 {'etl': '42% of cluster', 'reports': '18% of cluster'}

tree = {"prod": (60, {"etl": 70, "reports": 30}), "adhoc": (25, {}), "ds": (15, {})}
print("root level:", sum(c for c, _ in tree.values()))
for q, (c, kids) in tree.items():
    if kids:
        print(f"{q} children:", sum(kids.values()), {k: f"{c * v / 100:.0f}% of cluster" for k, v in kids.items()})

Real-world scenario questions (asked in company interviews)

Scheduler questions come up for every Hadoop admin and data-engineering role, usually as “a job is stuck” or “design the queues”.