--- title: 'Resource Managers' source: 'https://academia.sh/en/courses/distributed-data-processing/resource-managers' course: 'Distributed Data Processing' language: en updated: '2026-10-21T05:03:00+00:00' license: 'CC BY-SA 4.0' objectives: - 'Defines the number of tasks and the number of concurrent tasks as two separate decisions and measures the effect of each on the fixed cost.' - 'Counts at which number of tasks the makespan gain loses to the added fixed cost.' - 'Shows that in two jobs running on the same cluster the scheduling policy changes the order of finishing but does not change the fixed cost.' --- # Resource Managers Counting that the number of tasks and the number of concurrent tasks are two separate knobs in scheduling a job on a cluster, and where the fixed cost of fifty units per task goes: with one task in one slot the makespan is 2250 and the fixed cost 50, while sixteen tasks in sixteen slots bring the makespan down to 215 but raise the fixed cost to 800; increasing concurrency does not change the fixed cost at all, whereas increasing the number of tasks raises it by fifty units per task; splitting the job into more tasks at the same concurrency raises the makespan from 611 to 950; going from eight tasks to sixteen the makespan gain stays at 121 and loses to the 400 units of added fixed cost; and when two jobs run on the same cluster the priority and the shared policy change the order of finishing but do not change the 600 units of fixed cost paid. 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. ```python # 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. ```python 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**. ```python 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. ```python 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. ```python 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.