Resource Managers

The previous lesson decided which computation would run and counted what would not, but it never touched who would run that computation. The partitions were handed to the workers in turn, the number of workers was taken as four, and TASK_OVERHEAD was quietly added to the makespan once. In a real cluster none of these three is in the hands of the driver.

This lesson measures the component that stands in between: the resource manager. The driver submits a job, and the resource manager decides how many tasks to split the job into, how many tasks will run at the same time, and when a waiting job takes its turn. The number measured is this: when every task has a fixed cost of 50 units, how far do increasing the number of tasks and the concurrency bring the makespan down, and where does the gain lose to this cost.

• DC49. The cluster, the job and the resource manager are fictional; no real resource manager, cluster manager or cloud service is called. The seed is 20260218, and the measurement does not depend on the run.
• DC50. Makespan is the model’s own unit, not raw milliseconds: it is the work finished by the busiest slot.
• DC51. TASK_OVERHEAD = 50 is a choice, and the finding of this lesson rests directly on it; every result is written together with its value. The cost is paid per task: if a worker is given two tasks, it is paid twice.
• DC52. A task is the share given to a worker. A slot is a task place that can run at the same time in the cluster; when a slot frees up it takes the next task. The number of concurrent tasks is the number of slots.
• DC53. The partition key is subscriber and there is no shuffle in this lesson; the row work is 2200 in every setup. The case in which a wrong key makes adding workers meaningless belongs to the second topic of this course.
• DC54. The resource manager puts the large task first and places every task in the emptiest slot.
• DC55. The units counted are: tasks, concurrent tasks, makespan, row work, the total fixed cost and the total paid (row work plus fixed cost). The baseline is one task in one slot.
• DC56. The resource request, the limit and the queueing procedure were measured in the Container Orchestration course, and autoscaling in the Compute and Networking in the Cloud course, and they are not repeated. The question here is only what the number of tasks and the number of slots do to the makespan and to the fixed cost.
• DC57. Resolution: rows 2200, tasks at most 32, slots at most 16. The makespan and the fixed cost are whole numbers, and no ratio is printed.

What the Resource Manager Decides

What the driver submits is a computation plan; it does not say on how many machines it will run. That decision is made by the resource manager, and the decision has three parts. The first is the number of tasks: into how many shares the job will be split. The second is the number of slots: how many of these shares will run at the same time. The third is the order: when this job will start if the cluster is full of other jobs.

All three decisions share one cost. Giving work to a task is not free: the plan is sent, a slot is set aside, and an executor is prepared on the worker side. In this model that cost is called TASK_OVERHEAD and its value is 50. The value is a choice, but the moment it is chosen it has a consequence: increasing the number of tasks raises the fixed cost by 50 per task, while increasing the number of slots does not change it at all. The two knobs are not the same knob.

# FICTIONAL. The source, the first three stages of the pipeline, the cluster and the resource manager are
# defined in the lesson; no real resource manager or cluster manager is called.
SEED, M32 = 20260218, 0xFFFFFFFF


def rng(t):
    x = ((t ^ (t >> 16)) * 2246822507) & M32
    x = ((x ^ (x >> 13)) * 3266489909) & M32
    s = [(x ^ (x >> 16)) & M32]

    def next_value():
        s[0] = (s[0] * 1664525 + 1013904223) & M32
        return s[0] / 4294967296
    return next_value


REGION = ["dogu", "bati", "kuzey", "guney", "merkez"]
TYPE = ["residential", "commercial"]
PERIODS = [f"2026-{a:02d}" for a in range(1, 13)]
SUBSCRIBERS = {}
for i in range(200):
    r = rng(SEED + 37 * i)
    SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)],
                                    "start": 10000 + int(r() * 60000)}


def generate_source():
    records = []
    for i, (sub, a) in enumerate(SUBSCRIBERS.items()):
        e = a["start"]
        for p in PERIODS:
            r = rng(SEED + 101 * i + 7 * PERIODS.index(p))
            e += 4 + int(r() * 46)
            records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"],
                            "index": e, "unit": "m3"})
    return records


def s1_screen(records):
    REQUIRED = ("subscriber", "period", "region", "index", "unit")
    keep = [k for k in records if all(a in k for a in REQUIRED)]
    return keep, len(records) - len(keep)


def s2_dedupe(records):
    seen, keep = set(), []
    for k in records:
        key = (k["subscriber"], k["period"])
        if key in seen:
            continue
        seen.add(key)
        keep.append(k)
    return keep, len(records) - len(keep)


def s3_consumption(records):
    table = {(k["subscriber"], k["period"]): k for k in records}
    out, dropped = [], 0
    for k in records:
        i = PERIODS.index(k["period"])
        if i == 0:
            continue
        prev = table.get((k["subscriber"], PERIODS[i - 1]))
        if prev is None:
            dropped += 1
            continue
        try:
            f = int(k["index"]) - int(prev["index"])
        except (TypeError, ValueError):
            dropped += 1
            continue
        out.append(dict(k, m3=f))
    return out, dropped


WORKERS = 4
NETWORK_COST = 3          # the cost of moving one row over the network , as a multiple of local work
TASK_OVERHEAD = 50       # the fixed cost of handing work to a task , a choice


def rows():
    v, _ = s1_screen(generate_source())
    v, _ = s2_dedupe(v)
    v, _ = s3_consumption(v)
    return v


def trace_id(m):
    h = 2166136261
    for c in str(m):
        h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF
    return h


def distribute(records, key, workers=WORKERS):
    buckets = {i: [] for i in range(workers)}
    for k in records:
        buckets[trace_id(k[key]) % workers].append(k)
    return buckets


def cost(buckets, moved=0, workers=WORKERS):
    """Makespan = the busiest worker's work (parallel). Total work is counted separately."""
    n = [len(v) for v in buckets.values()]
    return {"duration": max(n) + moved * NETWORK_COST // workers + TASK_OVERHEAD,
            "total_work": sum(n) + moved * NETWORK_COST,
            "moved": moved}

Task, Slot and Fixed Cost

The scheduler carries the core of a real resource manager: the tasks are put in order from large to small, every task is placed in the emptiest slot, and every placement pays its own fixed cost. The makespan is the work finished by the busiest slot.

S = rows()


def shares(records, tasks):
    """The job is split into 'tasks' shares; the same key always goes to the same share."""
    return sorted((len(v) for v in distribute(records, "subscriber", tasks).values()), reverse=True)


def schedule(shares, concurrent, fixed=TASK_OVERHEAD):
    """Resource manager: at most 'concurrent' tasks run at the same time. A slot that frees up
    takes the next task , and every task pays its own fixed cost."""
    slots = [0] * concurrent
    for share in shares:
        i = slots.index(min(slots))
        slots[i] += share + fixed
    return {"tasks": len(shares), "duration": max(slots), "row_work": sum(shares),
            "fixed": len(shares) * fixed, "paid": sum(shares) + len(shares) * fixed}


print("cost (4 workers , no shuffle):", cost(distribute(S, "subscriber", 4)))
print("schedule (4 tasks , 4 slots) :", schedule(shares(S, 4), 4))
print("baseline (1 task , 1 slot)   :", schedule(shares(S, 1), 1))

cost (4 workers , no shuffle): {'duration': 611, 'total_work': 2200, 'moved': 0}
schedule (4 tasks , 4 slots) : {'tasks': 4, 'duration': 611, 'row_work': 2200, 'fixed': 200, 'paid': 2400}
baseline (1 task , 1 slot)   : {'tasks': 1, 'duration': 2250, 'row_work': 2200, 'fixed': 50, 'paid': 2250}

The first two rows give the same makespan: 611. This is a check: the course’s cost function and the scheduler of this lesson produce the same result for four tasks in four slots. The difference is in the third number. cost adds the fixed cost to the makespan once, since exactly one task falls on each worker; schedule counts it as a separate item and writes the total fixed cost of four tasks as 200. The baseline is the last row: one task, one slot, makespan 2250, fixed cost 50.

The Two Knobs Are Not the Same Knob

The matrix below puts the number of tasks in the rows and the number of slots in the columns. Every cell is the makespan; the left column is the total fixed cost paid by each row and is independent of the number of slots.

CONC = (1, 2, 4, 8, 16)
print(f"{'tasks':>6}{'fixed cost':>12}", "".join(f"{'conc=' + str(e):>9}" for e in CONC))
for g in (4, 8, 16, 32):
    p = shares(S, g)
    print(f"{g:>6}{g * TASK_OVERHEAD:>12}", "".join(f"{schedule(p, e)['duration']:>9}" for e in CONC))

 tasks  fixed cost    conc=1   conc=2   conc=4   conc=8  conc=16
     4         200      2400     1200      611      611      611
     8         400      2600     1300      650      336      336
    16         800      3000     1500      750      386      215
    32        1600      3800     1900      950      475      243

Reading along a row shows the slot knob: the job of sixteen tasks finishes with a makespan of 3000 in one slot and 215 in sixteen slots, and the fixed cost paid is 800 at both ends. Increasing the concurrency does not touch the fixed cost; it only shortens the time in which it is paid.

Reading along a column shows the second knob, and it works in the opposite direction. In four slots, if the job is split into four tasks the makespan is 611, into eight 650, into sixteen 750, into thirty-two 950. The job did not change; the row work is still 2200. The only thing that rises is the fixed cost: from 200 to 1600. Every task beyond the number of slots pays a 50 and buys no parallelism, because there is no free place to run.

The two knobs are tied to each other, and the most expensive corner of the table says so. To be able to use sixteen slots, the job must be split into at least sixteen tasks; that is, the precondition of dropping to a makespan of 215 is a fixed cost of 800 units. Splitting into thirty-two tasks raises the makespan in the same sixteen slots from 215 to 243 and the fixed cost to 1600. Saturation is in the same place too: the job of four tasks finishes with a makespan of 611 in eight or sixteen slots as well, and the extra slots sit idle.

Where the Makespan Gain Loses

The diagonal of the matrix represents the most frequently made decision: one slot for every task, that is, choosing how many workers to split the job into. To see where the gain loses, two columns are put side by side: the makespan gained and the fixed cost added.

print(f"{'tasks=conc':>12}{'skew':>6}{'makespan':>10}{'makespan gain':>15}"
      f"{'added fixed':>13}{'paid':>6}")
previous = None
for g in (1, 2, 4, 8, 16):
    p, z = shares(S, g), schedule(shares(S, g), g)
    c = round(max(p) / (sum(p) / len(p)), 4)
    gain = "-" if previous is None else str(previous[0] - z["duration"])
    added = "-" if previous is None else str(z["fixed"] - previous[1])
    print(f"{g:>12}{c:>6}{z['duration']:>10}{gain:>15}{added:>13}{z['paid']:>6}")
    previous = (z["duration"], z["fixed"])

  tasks=conc  skew  makespan  makespan gain  added fixed  paid
           1   1.0      2250              -            -  2250
           2   1.0      1150           1100           50  2300
           4  1.02       611            539          100  2400
           8  1.04       336            275          200  2600
          16   1.2       215            121          400  3000

The first three steps are beyond dispute: going from one task to two lowers the makespan by 1100 units and adds 50 units of fixed cost. From two to four the gain is 539, the cost 100. From four to eight the gain is 275, the cost 200. While the gain is halved at every step, the cost is doubled at every step; that is, the two numbers are running toward each other.

They meet in the step from eight to sixteen, and the order reverses: the makespan gain is 121, the added fixed cost 400. The place where the makespan gain loses is between eight and sixteen tasks. This does not mean the makespan does not fall; it really drops from 336 to 215. What it means is this: while the total paid rises from 2600 to 3000, the row work does not change at all, that is, all of the added 400 units is not work but the cost of handing out work. Someone who looks at this row without being told which one was measured sees a pure speedup.

The skew column carries a second warning. At sixteen tasks even the subscriber key produces a skew of 1.2: the heaviest share is 165 rows, the lightest 110. As the share gets smaller its equality deteriorates and the fixed cost grows; the two effects work in the same direction.

Two Jobs on the Same Cluster

The real work of the resource manager is not to speed up a single job but to share the cluster. Two jobs arrive at a cluster of eight slots: A is a job of eight tasks that processes all the rows, B is a job of four tasks that processes only the residential subscribers. In the priority policy A takes the whole cluster and B waits for its turn; in the shared policy the cluster is divided in two.

RESIDENTIAL = [k for k in S if k["type"] == "residential"]
A, B = shares(S, 8), shares(RESIDENTIAL, 4)
print("job A rows:", sum(A), " tasks:", len(A), "  job B rows:", sum(B), " tasks:", len(B))
for policy in ("priority", "shared"):
    if policy == "priority":                     # first A takes the whole cluster
        a = schedule(A, 8)["duration"]
        b = a + schedule(B, 8)["duration"]
    else:                                        # the cluster is divided in two
        a, b = schedule(A, 4)["duration"], schedule(B, 4)["duration"]
    print(f"{policy:<12} A finishes {a:>5}  B finishes {b:>5}  both {max(a, b):>5}"
          f"  fixed cost {(len(A) + len(B)) * TASK_OVERHEAD:>5}")

job A rows: 2200  tasks: 8   job B rows: 1078  tasks: 4
priority     A finishes   336  B finishes   683  both   683  fixed cost   600
shared       A finishes   650  B finishes   347  both   650  fixed cost   600

The policy changes the order of finishing from top to bottom. In the priority setup A finishes at 336 and B waits until 683; in the shared setup A is delayed to 650 while B finishes at 347. The moment when both are finished drops from 683 to 650; that is, the shared policy keeps the cluster a little fuller. There is no superior policy; what is chosen is which job is delayed.

The last column is the summing sentence of this lesson: the fixed cost is 600 in both policies. The scheduling policy determines who finishes when, and it does not determine the fixed cost paid. That cost is a function of the number of tasks alone, and twelve tasks pay 50 twelve times. What can be gained by changing the resource manager is the order, not the work.

Summary

• The resource manager decides three things: the number of tasks, the number of slots and the order; the first two are separate knobs and they behave differently toward the fixed cost.
• Increasing the number of slots does not change the fixed cost at all: the job of sixteen tasks finishes with a makespan of 3000 in one slot and 215 in sixteen slots, and the fixed cost paid is 800 at both ends.
• Increasing the number of tasks raises the fixed cost by 50 per task: in four slots the makespan rises from 611 to 950 while the row work stays at 2200, because a task beyond the number of slots buys no parallelism.
• When every task is given a slot, the makespan gain is halved at every step and the added fixed cost is doubled at every step; between eight and sixteen tasks the gain stays at 121 and loses to the 400 added units, and the total paid rises from 2600 to 3000.
• At sixteen tasks even the subscriber key produces a skew of 1.2; the inequality of the shares and the growth of the fixed cost work in the same direction.
• When two jobs run on the same cluster, the policy changes the order of finishing (A 650 and B 347 instead of A 336 and B 683) but does not change the fixed cost paid: 600 in both setups.

Next Step

In five lessons the job was split. A cluster was built and separated by the names driver, executor and task; the blocks were placed and which worker the read fell on was counted; the map was separated from the reduce; which part of the computation would never run was measured; and who decides into how many tasks the job is split was written down. In all these lessons one number stayed the same and was never questioned: the row work is 2200. We split the job, distributed it, scheduled it, but never touched the job itself.

Yet the slowness of a distributed job most often does not come from the number of rows. Moving from one partition key to another carries the rows over the network, and in this model that movement is counted at three times the cost of a local operation; there are two ways to join two relations, and the cheaper one changes with size; when a key does not divide the shares equally, some workers sit idle. The next topic counts these three separately, starting with this question: in the same job, with the same number of workers, only because the partition key changed, how many rows pass through the network and to where does the total work rise from 2200.