Resource Planning & Translation

fleet

How much hardware does this load need? — the second layer, wrapping the first.

slot sizes one cell; this sizes the fleet that runs all of them. Slots per unit, units for the fan-out, the autoscaling [min, max]. Still runtime-neutral: a caller supplies the shape of one schedulable unit (UnitShape — a Ray worker node, a Spark executor, a Dataproc worker) and gets back a RuntimeResourcePlan carrying both the decision and its evidence.

Three axes bound a density, and the smallest one wins. slots_per_unit = unit.cores // slot.cores is the design's formula and it is right up to the point where the cells do not fit: eight NeuralProphet cells at 4 GiB each do not run on a 30 GiB node no matter how many cores it has, and a node's card stops being the scarce resource as soon as the fraction is small enough that its cores run out first. So the density is min(device, cores, memory) over whichever of the three have a basis — the same rule the Serverless translation states explicitly (floor(usable_python_mem / peak_rss)), applied on the runtime that had been taking one axis at a time.

The device axis in particular used to stand alone: a GPU slot's density was accelerators x floor(1 / gpu_fraction) with no core term at all, so an n1-standard-8 with one T4 at the 0.1 fraction floor reported ten concurrent cells onto seven usable cores. Ten cells were never going to run; the arithmetic just never said so.

Adding the core term stopped the over-count and left the mirror-image problem behind it: that 0.1 floor is a constant, so on a sixteen-core node it capped a card at ten cells and left five cores idle, with the device named as the axis to go and fix. catalog.device_floor_fraction now derives the floor from the node instead, and _defining_axis knows that a fraction which came from the cores is not a statement about the card.

Nameplate is not schedulable, on either axis. schedulable_memory_bytes takes the plasma store and the OS off the RAM; schedulable_cores takes _RESERVED_CORES_PER_UNIT off the vCPUs for the node's own agents. Every bound here goes through one of the two.

And a bound is not a rule until some scheduler enforces it. min(device, cores, memory) answers "what do these three axes allow", which on Ray is not the same question as "what will this pool actually run" — nothing on the Ray path ever requests memory, so the scheduler packs on cores and cards alone and a memory-derived density is one the pool sails straight past. So slots_per_unit keeps the three-axis arithmetic, enforced_bounds drops whatever the runtime will not hold a task to, and plan_fleet sizes from the latter. See _UNENFORCED_AXES.

RuntimeResourcePlan dataclass

RuntimeResourcePlan(
    runtime,
    family,
    slot,
    unit,
    n_cells,
    slots_per_unit,
    derived_units,
    saturating_units,
    min_units,
    max_units,
    target_cells_per_slot,
)

One family's sizing decision on one runtime, with the evidence attached (pure).

Everything a runtime needs to launch the family's work, plus everything an auditor needs to reconstruct why: the slot and its provenance, the unit it packs into, the load it was sized for, and the resulting fleet bounds. Stamped whole into job_telemetry.

derived_units is the fan-out's implied unit count after the ceiling clamp — the number the fixed path provisions and the autoscaling path uses as its reference size. saturating_units is the count that would run every cell at once with no ceiling at all; the gap between the two is exactly how much the ceiling is throttling this run, and it is a number worth being able to read off a record rather than infer from a wall clock.

assigned_cores property

assigned_cores

Cores Ray will actually hand this plan's task — what its thread pools may use.

Not always slot.cores. A GPU task requests num_gpus and no num_cpus, so Ray assigns it the default of one core however wide the slot's core figure happens to be. Capping the thread pools at what the task is given rather than at what was planned for it is what keeps this number equal to the OMP_NUM_THREADS Ray sets beside it.

binding_axis property

binding_axis

Which resource holds this pool's density down — memory/device/cores.

Measured over the axes this runtime enforces, so it names the axis that actually held the pool down rather than one the scheduler was never asked about. On Ray that means an oversized memory figure can no longer be reported as the binding axis; it is surfaced by density_note instead, which is where an unenforced number belongs.

"scheduler" when the stored density did not come from this module's arithmetic at all (plan_fleet's density override — Serverless), because then no axis here is the one that decided.

density_note property

density_note

One line about the memory axis, when there is something to say (pure; else None).

A pool bound by cores is a pool doing the obvious thing, and saying so on every run is noise. Memory is the axis that goes wrong quietly, and it now does so in two opposite ways, so this says which one is happening.

It binds. slots_per_unit takes the min of the bounds and returns a single integer, so a slot sized at 97% of a node collapses the pool to one cell per unit and reports it as a density with no indication that the other axis had room to spare. That is exactly what a live 10,000-series Ray run did — one cell per node, seven of eight cores and 90% of every T4 idle, and nothing in the record said why (see profiling.source._without_driver_rss for the cause). Naming both sides of the comparison, and what the tightest other axis would have packed, is the difference between "this family is genuinely memory-heavy, buy bigger nodes" and "the slot is mis-measured".

It is measured but unenforced. On Ray nothing ever requests memory, so a large per-cell figure no longer shrinks the plan — and the silent failure inverts. A reader who knows the min takes three axes will assume the low number won, and size a quota request for a fleet twice the one the run needs. A 10,000-cell run advertised 3,334 nodes on a harvested 6.05 GiB per cell and then packed seven cells a node. So when the figure would have bound and does not, the note says so in as many words rather than going quiet.

slots_at_ceiling property

slots_at_ceiling

Concurrent cells the fleet could hold if it scaled all the way to max_units.

The demand an autoscaler must actually see before it will grow to its ceiling. See tasks_for_ceiling for why that matters.

task_options property

task_options

The Ray @ray.remote.options(**...) mapping this plan implies.

A GPU slot requests num_gpus and lets Ray default num_cpus to 1, exactly as the engine did before — several cells pack onto one device by summing fractions against its capacity of 1.0. A CPU slot requests num_cpus explicitly.

No memory, on Ray, ever — the request is omitted here rather than merely happening to be absent. The engine sizes its tasks from the driver-side pre-pass, which drops the memory axis for a live-proven reason, so in practice slot.memory_bytes has always been None by the time this property is reached from a submitting path. That made "Ray is never asked for memory" true by accident of which profile reaches which call site, and plan_fleet now sizes a fleet on the strength of it (_UNENFORCED_AXES). An invariant two modules depend on should not rest on a coincidence one edit away from being untrue, so the omission is stated where the request would be made. A measured footprint that a runtime will not schedule against is evidence, and it survives as density_note and in slot.to_dict(); it is not a resource request.

The task also carries a per-pool thread cap, and it has to ride here rather than on the job. Ray already sets OMP_NUM_THREADS per task, to that task's assigned cores, but it sets only that one — so OpenBLAS, MKL, NumExpr and vecLib fall back to counting the machine's cores and each cell claims the whole node. Seven cells packed onto seven cores then run forty-nine threads against seven, which costs throughput and, worse, makes cpu_seconds / fit_seconds report clean occupancy on a fleet that is thrashing. The remedy has to reach the worker before it imports numpy, because these pools are sized at import and cannot be resized afterwards, which rules out setting them inside the cell.

A task-level runtime_env is the seam that works: Ray merges its env_vars over the job's per key and inherits every other field, so working_dir and the uv install are untouched and neither is re-staged. Putting it here rather than in code_delivery.build_runtime_env is what lets the CPU and GPU pools carry different caps, which a single job-level value cannot express — the cost is that the two pools stop sharing worker processes.

Only meaningful for runtime == "ray"; the Spark translations emit properties, not options, and will carry their own accessor.

total_slots property

total_slots

Concurrent cells the derived fleet can hold — slots_per_unit x derived_units.

to_dict

to_dict()

JSON-safe dict of the whole decision — the record stamped into telemetry.

Source code in src/scale_forecasting/resources/fleet.py

def to_dict(self) -> dict[str, Any]:
    """JSON-safe dict of the whole decision — the record stamped into telemetry."""
    return {
        "runtime": self.runtime,
        "family": self.family,
        "n_cells": self.n_cells,
        "slot": self.slot.to_dict(),
        "unit": self.unit.to_dict(),
        "slots_per_unit": self.slots_per_unit,
        "derived_units": self.derived_units,
        "saturating_units": self.saturating_units,
        "min_units": self.min_units,
        "max_units": self.max_units,
        "target_cells_per_slot": self.target_cells_per_slot,
        "total_slots": self.total_slots,
        "slots_at_ceiling": self.slots_at_ceiling,
        "binding_axis": self.binding_axis,
        # What the runtime was actually asked for, beside what was decided. `binding_axis`
        # says which bound won; these two say what that turned into and, when memory won,
        # how much of another axis it left on the floor. Both were previously derivable only
        # by re-running the arithmetic against a plan nobody had kept.
        "task_options": self.task_options,
        "density_note": self.density_note,
    }

UnitShape dataclass

UnitShape(cores, memory_bytes=None, accelerators=0)

One schedulable unit of a fleet — a Ray worker node, later a Spark executor (pure).

The runtime-neutral description of the thing slots are packed into. memory_bytes is the unit's nameplate RAM (machine_memory_bytes); the schedulable share is applied during packing, not here, so the raw number stays legible in telemetry. accelerators is devices per unit — compute.accelerator_count for a Ray GPU pool, 0 for a CPU pool.

to_dict

to_dict()

JSON-safe dict for telemetry.

Source code in src/scale_forecasting/resources/fleet.py

def to_dict(self) -> dict[str, Any]:
    """JSON-safe dict for telemetry."""
    return {
        "cores": self.cores,
        "memory_bytes": self.memory_bytes,
        "accelerators": self.accelerators,
    }

enforced_bounds

enforced_bounds(slot, unit, runtime)

_bounds less the axes runtime will not hold a task to (pure).

Never empty: cores always has a basis and is enforced everywhere, so filtering can only ever remove the optional axes around it.

Source code in src/scale_forecasting/resources/fleet.py

def enforced_bounds(slot: ResourceSlot, unit: UnitShape, runtime: str) -> dict[str, int]:
    """`_bounds` less the axes ``runtime`` will not hold a task to (pure).

    Never empty: ``cores`` always has a basis and is enforced everywhere, so filtering can only
    ever remove the optional axes around it.
    """
    ignored = _UNENFORCED_AXES.get(runtime, frozenset())
    return {axis: value for axis, value in _bounds(slot, unit).items() if axis not in ignored}

max_slot_memory_bytes

max_slot_memory_bytes(unit)

The largest memory ask one slot may make of a unit (pure; unknown → None).

schedulable_memory_bytes with _MAX_SLOT_MEMORY_FRACTION of headroom on top, and the distinction between the two is the difference between a fleet that runs and one that hangs. The schedulable figure is an estimate of the scheduler's ceiling, derived from a machine type's nameplate RAM; the scheduler's real ceiling is derived from what the container's OS reports, which is lower. Clamping a slot to exactly the estimate therefore produces a task request that is plausibly just above what any node can offer, and an unplaceable Ray task does not fail — it queues forever while the autoscaler explains, once per second, that no node type can fulfil it.

Use this wherever a slot is being clamped to fit; use schedulable_memory_bytes where a unit's capacity is being divided up (_memory_bound, container sizing). Same node, two different questions.

Source code in src/scale_forecasting/resources/fleet.py

def max_slot_memory_bytes(unit: UnitShape) -> int | None:
    """The largest memory ask one slot may make of a unit (pure; unknown → ``None``).

    `schedulable_memory_bytes` with `_MAX_SLOT_MEMORY_FRACTION` of headroom on top, and the
    distinction between the two is the difference between a fleet that runs and one that hangs.
    The schedulable figure is an *estimate* of the scheduler's ceiling, derived from a machine
    type's nameplate RAM; the scheduler's real ceiling is derived from what the container's OS
    reports, which is lower. Clamping a slot to exactly the estimate therefore produces a task
    request that is plausibly just above what any node can offer, and an unplaceable Ray task does
    not fail — it queues forever while the autoscaler explains, once per second, that no node type
    can fulfil it.

    Use this wherever a slot is being *clamped to fit*; use `schedulable_memory_bytes` where a
    unit's capacity is being *divided up* (`_memory_bound`, container sizing). Same node, two
    different questions.
    """
    schedulable = schedulable_memory_bytes(unit)
    return int(schedulable * _MAX_SLOT_MEMORY_FRACTION) if schedulable else None

plan_fleet

plan_fleet(
    slot,
    *,
    runtime,
    n_cells,
    unit,
    target_cells_per_slot=_DEFAULT_TARGET_CELLS_PER_SLOT,
    min_units=1,
    max_units=1,
    density=None,
)

The fleet half of plan_resources, over a slot the caller already sized (pure).

Split out because a pool is not a family: a shared Ray CPU pool runs several families through one worker, so its slot comes from merge_slots rather than from a single resource_slot call. Both entry points then need the identical fleet arithmetic, and duplicating it is how the two paths would quietly drift apart.

The slot is taken as given — a caller that assembles one by hand is responsible for having clamped it to the unit (schedulable_memory_bytes is the ceiling to clamp against). slots_per_unit still floors the density at 1, so an unclamped slot yields an inefficient plan rather than a stalled pool.

density overrides the cells-per-unit figure for a runtime whose own scheduler is the authority on it: Serverless divides an executor's cores by spark.task.cpus and honours nothing else, so a fleet sized off slots_per_unit's device arithmetic would be sized off a density the platform never grants (spark_tasks_per_executor). Left None — every Ray caller — the derivation stands.

Without an override the density comes from enforced_bounds rather than slots_per_unit, which is the same principle one level finer. The override drops a whole derivation because another scheduler owns it; this drops a single axis because this scheduler is never asked about it. Both exist so that the number this function reports is a number the pool can actually reach — see _UNENFORCED_AXES for the Ray memory case that made it necessary.

Source code in src/scale_forecasting/resources/fleet.py

def plan_fleet(
    slot: ResourceSlot,
    *,
    runtime: str,
    n_cells: int,
    unit: UnitShape,
    target_cells_per_slot: int = _DEFAULT_TARGET_CELLS_PER_SLOT,
    min_units: int = 1,
    max_units: int = 1,
    density: int | None = None,
) -> RuntimeResourcePlan:
    """The fleet half of `plan_resources`, over a slot the caller already sized (pure).

    Split out because a *pool* is not a family: a shared Ray CPU pool runs several families
    through one worker, so its slot comes from `merge_slots` rather than from a single
    `resource_slot` call. Both entry points then need the identical fleet arithmetic, and
    duplicating it is how the two paths would quietly drift apart.

    The slot is taken as given — a caller that assembles one by hand is responsible for
    having clamped it to the unit (`schedulable_memory_bytes` is the ceiling to clamp
    against). `slots_per_unit` still floors the density at 1, so an unclamped slot
    yields an inefficient plan rather than a stalled pool.

    ``density`` overrides the cells-per-unit figure for a runtime whose own scheduler is the
    authority on it: Serverless divides an executor's cores by ``spark.task.cpus`` and honours
    nothing else, so a fleet sized off `slots_per_unit`'s device arithmetic would be sized off a
    density the platform never grants (`spark_tasks_per_executor`). Left ``None`` — every Ray
    caller — the derivation stands.

    Without an override the density comes from `enforced_bounds` rather than `slots_per_unit`,
    which is the same principle one level finer. The override drops a whole derivation because
    another scheduler owns it; this drops a single *axis* because this scheduler is never asked
    about it. Both exist so that the number this function reports is a number the pool can
    actually reach — see `_UNENFORCED_AXES` for the Ray memory case that made it necessary.
    """
    per_unit = (
        max(1, density)
        if density is not None
        else max(1, min(enforced_bounds(slot, unit, runtime).values()))
    )
    cells_per_unit = max(1, per_unit * max(1, target_cells_per_slot))
    saturating = math.ceil(n_cells / per_unit) if n_cells > 0 else 0
    if n_cells <= 0:
        derived = 0
    else:
        derived = max(min_units, min(math.ceil(n_cells / cells_per_unit), max_units))

    return RuntimeResourcePlan(
        runtime=runtime,
        family=slot.family,
        slot=slot,
        unit=unit,
        n_cells=max(0, n_cells),
        slots_per_unit=per_unit,
        derived_units=derived,
        saturating_units=saturating,
        min_units=min_units,
        max_units=max_units,
        target_cells_per_slot=target_cells_per_slot,
    )

plan_resources

plan_resources(
    profile,
    family,
    runtime,
    n_cells,
    *,
    unit,
    use_gpu=False,
    device_bytes=None,
    static_gpu_fraction=None,
    target_cells_per_slot=_DEFAULT_TARGET_CELLS_PER_SLOT,
    min_units=1,
    max_units=1,
)

Size one family's fleet on one runtime from its measured profile (pure).

Three steps, each independently testable: size the slot (resource_slot), pack slots into a unit (slots_per_unit), then widen the fleet until the load flows through at the target density.

derived_units = ceil(n_cells / (slots_per_unit x target_cells_per_slot)), clamped into [min_units, max_units] — the same shape as the node-count arithmetic the Ray engine already used, now over a measured slot rather than an assumed one. An empty pool (n_cells <= 0) derives 0 units and is never floored to the minimum: a family with no work should not provision hardware.

saturating_units answers the different question of how wide the fleet would have to be to run every cell simultaneously, and is deliberately left unclamped so the record shows when the ceiling — not the work — is what bounded the run.

runtime is carried rather than branched on: the slot and the fleet math are the same for every runtime, and what differs is only how the numbers are spelled when handed over (RuntimeResourcePlan.task_options for Ray; spark.* properties for the Spark runtimes). Keeping the divergence at the edge is what makes one profile usable three ways.

The slot is clamped to the unit before packing, so a family measured larger than any available node yields a schedulable — if inefficient — plan with the clamp recorded, instead of a task the scheduler will never place.

Source code in src/scale_forecasting/resources/fleet.py

def plan_resources(
    profile: ComputeProfile | None,
    family: str,
    runtime: str,
    n_cells: int,
    *,
    unit: UnitShape,
    use_gpu: bool = False,
    device_bytes: int | None = None,
    static_gpu_fraction: float | None = None,
    target_cells_per_slot: int = _DEFAULT_TARGET_CELLS_PER_SLOT,
    min_units: int = 1,
    max_units: int = 1,
) -> RuntimeResourcePlan:
    """Size one family's fleet on one runtime from its measured profile (pure).

    Three steps, each independently testable: size the slot (`resource_slot`), pack
    slots into a unit (`slots_per_unit`), then widen the fleet until the load flows
    through at the target density.

    ``derived_units = ceil(n_cells / (slots_per_unit x target_cells_per_slot))``, clamped
    into ``[min_units, max_units]`` — the same shape as the node-count arithmetic the Ray
    engine already used, now over a *measured* slot rather than an assumed one. An empty
    pool (``n_cells <= 0``) derives 0 units and is never floored to the minimum: a family
    with no work should not provision hardware.

    ``saturating_units`` answers the different question of how wide the fleet would have to
    be to run every cell simultaneously, and is deliberately left unclamped so the record
    shows when the ceiling — not the work — is what bounded the run.

    ``runtime`` is carried rather than branched on: the slot and the fleet math are the same
    for every runtime, and what differs is only how the numbers are *spelled* when handed
    over (`RuntimeResourcePlan.task_options` for Ray; ``spark.*`` properties for the
    Spark runtimes). Keeping the divergence at the edge is what makes one profile usable
    three ways.

    The slot is clamped to the unit before packing, so a family measured larger than any
    available node yields a schedulable — if inefficient — plan with the clamp recorded,
    instead of a task the scheduler will never place.
    """
    slot = resource_slot(
        profile,
        family,
        use_gpu=use_gpu,
        device_bytes=device_bytes,
        static_gpu_fraction=static_gpu_fraction,
        min_gpu_fraction=device_floor_fraction(schedulable_cores(unit), unit.accelerators),
        max_cores=unit.cores if unit.cores > 0 else None,
        max_memory_bytes=max_slot_memory_bytes(unit),
    )
    return plan_fleet(
        slot,
        runtime=runtime,
        n_cells=n_cells,
        unit=unit,
        target_cells_per_slot=target_cells_per_slot,
        min_units=min_units,
        max_units=max_units,
    )

schedulable_cores

schedulable_cores(unit)

vCPUs a unit will actually run cells on — nameplate minus the reserve (pure; >= 1).

The core-axis twin of schedulable_memory_bytes, and it exists for the same reason: the number on the machine type is what Google bills, not what the scheduler has left to give once the node's own agents are running. See _RESERVED_CORES_PER_UNIT for why the reserve is a flat core rather than a share.

Floored at 1 so a single-core unit still holds a cell. A unit that holds zero cells is not a conservative plan, it is a pool that never starts.

Source code in src/scale_forecasting/resources/fleet.py

def schedulable_cores(unit: UnitShape) -> int:
    """vCPUs a unit will actually run cells on — nameplate minus the reserve (pure; >= 1).

    The core-axis twin of `schedulable_memory_bytes`, and it exists for the same reason: the
    number on the machine type is what Google bills, not what the scheduler has left to give
    once the node's own agents are running. See `_RESERVED_CORES_PER_UNIT` for why the reserve
    is a flat core rather than a share.

    Floored at 1 so a single-core unit still holds a cell. A unit that holds zero cells is not
    a conservative plan, it is a pool that never starts.
    """
    return max(1, unit.cores - _RESERVED_CORES_PER_UNIT)

schedulable_memory_bytes

schedulable_memory_bytes(unit)

The share of a unit's RAM a scheduler will actually hand out (pure; unknown → None).

_SCHEDULABLE_MEMORY_FRACTION of nameplate. None when the unit's memory is unknown (an unparseable machine type), which callers must read as no memory bound rather than as a unit with no memory. Exposed because every caller that clamps a slot has to clamp against the same ceiling the packing arithmetic uses.

Source code in src/scale_forecasting/resources/fleet.py

def schedulable_memory_bytes(unit: UnitShape) -> int | None:
    """The share of a unit's RAM a scheduler will actually hand out (pure; unknown → ``None``).

    ``_SCHEDULABLE_MEMORY_FRACTION`` of nameplate. ``None`` when the unit's memory is unknown
    (an unparseable machine type), which callers must read as *no memory bound* rather than
    as a unit with no memory. Exposed because every caller that clamps a slot has to clamp
    against the same ceiling the packing arithmetic uses.
    """
    return int(unit.memory_bytes * _SCHEDULABLE_MEMORY_FRACTION) if unit.memory_bytes else None

slots_per_unit

slots_per_unit(slot, unit)

Concurrent cells one unit holds — the smallest of its three bounds (pure).

• device — accelerators x floor(1 / gpu_fraction), only when the slot carries a fraction. A GPU node's cores and RAM are sized around its cards, so this usually binds first; "usually" is exactly why it cannot be the only term.
• cores — floor(schedulable_cores / slot.cores), on every slot including a GPU one. A cell needs a core to run on whether or not it also needs a card, and the card does not supply one: an n1-standard-8 + 1 T4 at the 0.1 fraction floor is ten cells by the device bound and seven by this one, and seven is the number the node can actually run.
• memory — floor(schedulable_memory / slot.memory_bytes), when both sides are known. What stops eight 4 GiB cells landing on a 30 GiB node, and what a Spark executor's heap really does impose on the tasks sharing it.

An axis with no basis is absent, not zero — an unmeasured memory footprint must not shrink a fleet it knows nothing about, which is the property that keeps turning the profiler on from ever making a run worse.

This is the arithmetic, not the plan. It answers "what do these three axes allow", which is not the same question as "what will this runtime hold a task to" — on Ray they differ, because nothing there ever requests memory. plan_fleet sizes from enforced_bounds for that reason and callers wanting the number a pool will actually reach should do the same. A density reported but not honoured is a density the pool never reaches, and it used to be reported here: see _UNENFORCED_AXES.

Always at least 1. A slot too big for its unit has already been clamped to fit by resource_slot, so the floor here is a belt-and-braces guard against a caller that assembled a slot by hand.

Source code in src/scale_forecasting/resources/fleet.py

def slots_per_unit(slot: ResourceSlot, unit: UnitShape) -> int:
    """Concurrent cells one unit holds — the smallest of its three bounds (pure).

    * **device** — ``accelerators x floor(1 / gpu_fraction)``, only when the slot carries a
      fraction. A GPU node's cores and RAM are sized around its cards, so this usually binds
      first; "usually" is exactly why it cannot be the only term.
    * **cores** — ``floor(schedulable_cores / slot.cores)``, on **every** slot including a GPU
      one. A cell needs a core to run on whether or not it also needs a card, and the card does
      not supply one: an ``n1-standard-8`` + 1 T4 at the 0.1 fraction floor is ten cells by the
      device bound and seven by this one, and seven is the number the node can actually run.
    * **memory** — ``floor(schedulable_memory / slot.memory_bytes)``, when both sides are known.
      What stops eight 4 GiB cells landing on a 30 GiB node, and what a Spark executor's heap
      really does impose on the tasks sharing it.

    An axis with no basis is *absent*, not zero — an unmeasured memory footprint must not shrink
    a fleet it knows nothing about, which is the property that keeps turning the profiler on from
    ever making a run worse.

    **This is the arithmetic, not the plan.** It answers "what do these three axes allow", which
    is not the same question as "what will this runtime hold a task to" — on Ray they differ,
    because nothing there ever requests memory. `plan_fleet` sizes from `enforced_bounds` for
    that reason and callers wanting the number a pool will actually reach should do the same.
    A density reported but not honoured is a density the pool never reaches, and it used to be
    reported here: see `_UNENFORCED_AXES`.

    Always at least 1. A slot too big for its unit has already been clamped to fit by
    `resource_slot`, so the floor here is a belt-and-braces guard against a caller
    that assembled a slot by hand.
    """
    return max(1, min(_bounds(slot, unit).values()))

tasks_for_ceiling

tasks_for_ceiling(plan)

Concurrent tasks the fan-out must produce before the autoscaler will reach its ceiling.

An autoscaler grows on pending demand. Ray adds nodes because tasks are queued and cannot be placed; if the run only ever submits as many tasks as the current fleet can hold, nothing is ever pending and the pool sits at its minimum — the "we enabled autoscaling and nothing scaled" failure, which looks like a platform problem and is actually an arithmetic one.

So a run that wants to be able to reach max_units must split its work into at least slots_per_unit x max_units tasks. Callers compare this against their chunk/bucket count and raise it if it falls short. Returns 0 for an unused pool (no cells, no ceiling worth reaching).

Source code in src/scale_forecasting/resources/fleet.py

def tasks_for_ceiling(plan: RuntimeResourcePlan) -> int:
    """Concurrent tasks the fan-out must produce before the autoscaler will reach its ceiling.

    An autoscaler grows on *pending demand*. Ray adds nodes because tasks are queued and
    cannot be placed; if the run only ever submits as many tasks as the current fleet can
    hold, nothing is ever pending and the pool sits at its minimum — the "we enabled
    autoscaling and nothing scaled" failure, which looks like a platform problem and is
    actually an arithmetic one.

    So a run that wants to be able to reach ``max_units`` must split its work into at least
    ``slots_per_unit x max_units`` tasks. Callers compare this against their chunk/bucket
    count and raise it if it falls short. Returns 0 for an unused pool (no cells, no
    ceiling worth reaching).
    """
    if plan.n_cells <= 0:
        return 0
    return plan.slots_at_ceiling

slot

How big is one cell of this family? — the first of the package's two layers.

Cores, host memory, GPU fraction: three numbers, runtime-neutral by construction. The same slot describes a cell whether it runs as a Ray task or a Spark partition, and the translators above turn it into whichever knobs their platform accepts.

The seam is the point. Sizing arithmetic that lives inside an engine can only be tested with that engine running. Everything in this package is pure — no Ray, no Spark, no GPU, no GCP. An engine hands over a ComputeProfile and gets back a decision plus the evidence for it.

Absence propagates; it is never filled in silently. A ResourceSlot records which axes came from measurement (measured) and which fell back to a static default (assumed), plus every clamp applied (notes). An unmeasured memory axis requests no memory — today's behaviour — rather than requesting a number nobody took. That is the contract FamilyCost keeps, carried one layer out so "we sized this off nothing" survives into telemetry.

What measurement actually changed, honestly. Of the three axes, the interesting result is that cores is usually 1, and that is a finding rather than a null one. The probe pins every native thread pool to one thread (profiling.measure._pinned_intraop_threads), which is exactly the environment a Ray task runs in — Ray exports OMP_NUM_THREADS = the task's num_cpus. So the measurement says the hardcoded num_cpus=1 was right, and now says it from evidence instead of from assumption. The axes that move a run are the other two: a memory request Ray had never been given at all, and a GPU fraction derived from a real footprint rather than from ray_io._NOMINAL_AUTO_FRACTION.

ResourceSlot dataclass

ResourceSlot(
    family,
    cores,
    memory_bytes,
    gpu_fraction,
    device_bytes,
    measured=(),
    assumed=(),
    notes=(),
)

What one cell of one family needs, and how much of that was actually measured (pure).

Runtime-neutral by construction: three numbers plus their provenance. cores is always present (it has a defensible static default); memory_bytes and gpu_fraction are None when there is no basis, and None means request nothing on this axis — not zero, and not a guess.

measured and assumed partition the axes that carry a value, so a reader of the telemetry can tell a 4 GiB request that came from a fit apart from one that came from a table. notes records every clamp in plain words; a slot that was silently trimmed to fit the machine is a sizing decision, and sizing decisions have to be auditable.

basis property

basis

"measured" if any axis came from a fit, else "static" — the headline.

to_dict

to_dict()

JSON-safe dict for telemetry — plain scalars and lists, no custom encoder.

Source code in src/scale_forecasting/resources/slot.py

def to_dict(self) -> dict[str, Any]:
    """JSON-safe dict for telemetry — plain scalars and lists, no custom encoder."""
    return {
        "family": self.family,
        "basis": self.basis,
        "cores": self.cores,
        "memory_bytes": self.memory_bytes,
        "gpu_fraction": self.gpu_fraction,
        "device_bytes": self.device_bytes,
        "measured": list(self.measured),
        "assumed": list(self.assumed),
        "notes": list(self.notes),
    }

merge_slots

merge_slots(slots, *, family)

Collapse several families' slots into the one slot a shared pool needs (pure).

A Ray CPU pool runs whatever lands on it — statistical cells and ML cells go through the same worker — so its slot has to hold the heaviest of them. Same roll-up rule FamilyCost applies across a family's models, one level out: max per axis, because the slot must fit whichever cell arrives, not the average one.

Provenance is resolved per axis by asking where the winning number came from: an axis is measured iff a contributor that measured it supplied the max. That is the honest answer — a 4 GiB max taken from a real fit is measured evidence even if a lighter family beside it had none. What the lighter family's gap does earn is a note, so a reader can see that the pool was sized off a subset of the families that will run on it. Without that note "measured" would over-claim; with it, both facts are on the record.

family is the merged label ("statistical+ml"). Raises on an empty sequence: a pool with no families is a caller bug, not a slot.

Source code in src/scale_forecasting/resources/slot.py

def merge_slots(slots: Sequence[ResourceSlot], *, family: str) -> ResourceSlot:
    """Collapse several families' slots into the one slot a shared pool needs (pure).

    A Ray CPU pool runs whatever lands on it — statistical cells and ML cells go through the
    same worker — so its slot has to hold the heaviest of them. Same roll-up rule
    `FamilyCost` applies across a family's models, one level out: **max per axis**,
    because the slot must fit whichever cell arrives, not the average one.

    Provenance is resolved per axis by asking *where the winning number came from*: an axis is
    ``measured`` iff a contributor that measured it supplied the max. That is the honest
    answer — a 4 GiB max taken from a real fit is measured evidence even if a lighter family
    beside it had none. What the lighter family's gap does earn is a **note**, so a reader can
    see that the pool was sized off a subset of the families that will run on it. Without that
    note "measured" would over-claim; with it, both facts are on the record.

    ``family`` is the merged label (``"statistical+ml"``). Raises on an empty sequence: a pool
    with no families is a caller bug, not a slot.
    """
    if not slots:
        raise ValueError("merge_slots needs at least one slot")

    def pick(values: list[tuple[float | None, bool]]) -> tuple[float | None, bool]:
        """The max across contributors, plus whether a *measuring* contributor supplied it."""
        present = [(value, was_measured) for value, was_measured in values if value is not None]
        if not present:
            return None, False
        best = max(value for value, _ in present)
        return best, any(was_measured for value, was_measured in present if value == best)

    def axis_of(slot: ResourceSlot, axis: str) -> tuple[float | None, bool]:
        raw = getattr(slot, axis)
        return (None if raw is None else float(raw)), axis in slot.measured

    cores, cores_measured = pick([axis_of(s, "cores") for s in slots])
    memory, memory_measured = pick([axis_of(s, "memory_bytes") for s in slots])
    fraction, fraction_measured = pick([axis_of(s, "gpu_fraction") for s in slots])

    measured: list[str] = []
    assumed: list[str] = []
    for axis, value, was_measured in (
        ("cores", cores, cores_measured),
        ("memory_bytes", memory, memory_measured),
        ("gpu_fraction", fraction, fraction_measured),
    ):
        if axis == "gpu_fraction" and value is None:
            continue  # a CPU pool has no device axis at all, measured or otherwise
        (measured if was_measured else assumed).append(axis)

    notes = [note for slot in slots for note in slot.notes]
    for axis in ("cores", "memory_bytes", "gpu_fraction"):
        if axis == "gpu_fraction" and fraction is None:
            continue
        blind = sorted(s.family for s in slots if axis not in s.measured)
        if blind and len(blind) < len(slots):
            notes.append(f"{axis} sized without a measurement for {', '.join(blind)}")

    return ResourceSlot(
        family=family,
        cores=max(1, int(cores or _DEFAULT_SLOT_CORES)),
        memory_bytes=None if memory is None else int(memory),
        gpu_fraction=fraction,
        device_bytes=next((s.device_bytes for s in slots if s.device_bytes is not None), None),
        measured=tuple(measured),
        assumed=tuple(assumed),
        notes=tuple(notes),
    )

resource_slot

resource_slot(
    profile,
    family,
    *,
    use_gpu=False,
    device_bytes=None,
    static_gpu_fraction=None,
    min_gpu_fraction=None,
    max_cores=None,
    max_memory_bytes=None,
)

Size one cell of family from the profile, falling back per-axis (pure).

Each axis resolves independently, because they fail independently: a family can have a solid wall-time and RSS measurement and no GPU number at all (nothing ran on a device), or the reverse. Resolving them as a unit would throw away good evidence because of a neighbouring gap.

• cores — FamilyCost.slot_cores, else _DEFAULT_SLOT_CORES. Clamped to max_cores when given: a task asking for more cores than any node has is not slow, it is unschedulable, and Ray will sit on it forever rather than fail. Clamping and recording beats hanging.
• memory — FamilyCost.slot_rss_bytes (the absolute footprint x the profile's memory margin), else None. Clamped to max_memory_bytes for the same unschedulable-task reason.
• gpu_fraction — None unless use_gpu. When measured and device_bytes is known: slot_gpu_bytes / device_bytes, clamped to the band. Otherwise static_gpu_fraction (the operator's pin), and failing that the nominal. min_gpu_fraction is the bottom of that band — pass catalog.device_floor_fraction for the unit so the floor cannot cap a device below what the node's cores would run; omitting it keeps the flat default, which is right only for a direct caller with no node in mind.

profile may be None (profiling off, or the pre-pass produced nothing) and the family may simply be absent from it; both take every fallback, which reproduces the pre-profiler behaviour exactly. That equivalence is the safety property: turning the profiler on can add information, never remove it.

The margin already lives inside the slot_* properties, so nothing here multiplies again. Note that the measured GPU path applies the profile's memory_margin while the legacy ray_io.calibrate_gpu_fraction path applies compute.gpu_safety_margin; both default to 1.3, so the two agree unless an operator moves one of them.

Source code in src/scale_forecasting/resources/slot.py

def resource_slot(
    profile: ComputeProfile | None,
    family: str,
    *,
    use_gpu: bool = False,
    device_bytes: int | None = None,
    static_gpu_fraction: float | None = None,
    min_gpu_fraction: float | None = None,
    max_cores: int | None = None,
    max_memory_bytes: int | None = None,
) -> ResourceSlot:
    """Size one cell of ``family`` from the profile, falling back per-axis (pure).

    Each axis resolves independently, because they fail independently: a family can have a
    solid wall-time and RSS measurement and no GPU number at all (nothing ran on a device),
    or the reverse. Resolving them as a unit would throw away good evidence because of a
    neighbouring gap.

    * **cores** — ``FamilyCost.slot_cores``, else `_DEFAULT_SLOT_CORES`. Clamped to
      ``max_cores`` when given: a task asking for more cores than any node has is not slow,
      it is *unschedulable*, and Ray will sit on it forever rather than fail. Clamping and
      recording beats hanging.
    * **memory** — ``FamilyCost.slot_rss_bytes`` (the absolute footprint x the profile's
      memory margin), else ``None``. Clamped to ``max_memory_bytes`` for the same
      unschedulable-task reason.
    * **gpu_fraction** — ``None`` unless ``use_gpu``. When measured *and* ``device_bytes``
      is known: ``slot_gpu_bytes / device_bytes``, clamped to the band. Otherwise
      ``static_gpu_fraction`` (the operator's pin), and failing that the nominal.
      ``min_gpu_fraction`` is the bottom of that band — pass `catalog.device_floor_fraction`
      for the unit so the floor cannot cap a device below what the node's cores would run;
      omitting it keeps the flat default, which is right only for a direct caller with no node
      in mind.

    ``profile`` may be ``None`` (profiling off, or the pre-pass produced nothing) and the
    family may simply be absent from it; both take every fallback, which reproduces the
    pre-profiler behaviour exactly. That equivalence is the safety property: turning the
    profiler on can add information, never remove it.

    The margin already lives inside the ``slot_*`` properties, so nothing here multiplies
    again. Note that the measured GPU path applies the profile's ``memory_margin`` while the
    legacy `ray_io.calibrate_gpu_fraction` path applies ``compute.gpu_safety_margin``;
    both default to 1.3, so the two agree unless an operator moves one of them.
    """
    cost = profile.for_family(family) if profile is not None else None
    measured: list[str] = []
    assumed: list[str] = []
    notes: list[str] = []

    cores = cost.slot_cores if cost is not None else None
    if cores is None:
        cores = _DEFAULT_SLOT_CORES
        assumed.append("cores")
    else:
        measured.append("cores")
    if max_cores is not None and cores > max_cores:
        notes.append(f"cores {cores} exceeded the unit's {max_cores}; clamped")
        cores = max_cores
    cores = max(1, cores)

    memory_bytes = cost.slot_rss_bytes if cost is not None else None
    if memory_bytes is None:
        assumed.append("memory_bytes")
    else:
        measured.append("memory_bytes")
        if max_memory_bytes is not None and memory_bytes > max_memory_bytes:
            notes.append(
                f"memory {memory_bytes} exceeded the unit's schedulable {max_memory_bytes}; clamped"
            )
            memory_bytes = max_memory_bytes

    gpu_fraction = _resolve_gpu_fraction(
        cost,
        use_gpu=use_gpu,
        device_bytes=device_bytes,
        static_gpu_fraction=static_gpu_fraction,
        min_gpu_fraction=min_gpu_fraction,
        measured=measured,
        assumed=assumed,
        notes=notes,
    )

    return ResourceSlot(
        family=family,
        cores=cores,
        memory_bytes=memory_bytes,
        gpu_fraction=gpu_fraction,
        device_bytes=device_bytes if use_gpu else None,
        measured=tuple(measured),
        assumed=tuple(assumed),
        notes=tuple(notes),
    )

serverless

Serverless Spark — the arithmetic, snapped to the knobs the service will accept.

The first of the two Spark translators (cluster is the other; they are peers and share their platform facts through catalog, never through each other).

The knobs are discrete. Serverless takes executor.cores from {4, 8, 16} — a different set again on GPU — and rejects anything else at submit, minutes after the operator walked away. So this module never emits a raw arithmetic result: it computes the ideal, snaps to a legal neighbour in whichever direction is safe for that knob (memory up, tasks-per-device down), and keeps both numbers, so an audit can see what the arithmetic wanted and what the platform allowed.

It also exports the thread-pin caps itself. A Ray task inherits OMP_NUM_THREADS for free; a Spark executor pins nothing, so N concurrent Python workers each grab the whole executor. Setting the caps is what makes a profile measured under a pin describe the executor it is being used to size.

ServerlessTranslation dataclass

ServerlessTranslation(
    executor_cores,
    ideal_executor_cores,
    tasks_per_device,
    properties,
    notes=(),
)

One plan spelled as Dataproc Serverless properties, with the ideals it snapped from.

properties is merged straight into RuntimeConfig.properties. Everything beside it exists so an audit can answer why: what the arithmetic wanted (ideal), what the platform's legal-value table allowed (the properties), and where the two disagreed enough to matter (notes).

to_dict

to_dict()

JSON-safe record of the translation — what telemetry stamps alongside the plan.

Source code in src/scale_forecasting/resources/serverless.py

def to_dict(self) -> dict[str, Any]:
    """JSON-safe record of the translation — what telemetry stamps alongside the plan."""
    return {
        "executor_cores": self.executor_cores,
        "ideal_executor_cores": self.ideal_executor_cores,
        "tasks_per_device": self.tasks_per_device,
        "properties": dict(self.properties),
        "notes": list(self.notes),
    }

plan_serverless

plan_serverless(
    profile,
    families,
    n_cells,
    *,
    gpu=False,
    device_bytes=None,
    static_gpu_fraction=None,
    target_cells_per_slot=_DEFAULT_TARGET_CELLS_PER_SLOT,
    max_executors=None,
    tier="standard",
    pin_threads=True,
)

Size a Serverless batch and spell it as properties, in one call (pure).

Two passes, because the executor's shape and the fleet's size each depend on the other. The slot fixes executor.cores on its own — that is the whole point of the inversion, the cores come from the measured fraction rather than from the fleet — so pass one translates against the widest legal executor purely to learn the core count, and pass two builds the real serverless_unit from it and plans the fleet properly. The same shape ray_io.plan_cluster uses for its autoscaling ceiling, and for the same reason.

Pass one deliberately clamps against the widest shape rather than a nominal one: clamping to a small executor first would cap the measured slot before it had a chance to ask for a bigger executor, and the batch would then be sized to the clamp instead of to the work. The slot is re-derived against the chosen shape in pass two, so the plan that comes out is still clamped to the executor it will actually run on.

max_executors is the operator's explicit cap (--max-executors). Left None the ceiling is the saturating count — enough executors to run every cell at once — rather than a derived guess. That asymmetry is deliberate: Serverless already defaults maxExecutors to 1000, so a ceiling we invent can only ever make a large run slower than leaving it alone, while a floor we raise can only make it faster.

families is every family the batch will run, merged the same way a shared Ray pool's is (merge_slots): one executor shape has to hold whichever cell arrives.

Source code in src/scale_forecasting/resources/serverless.py

def plan_serverless(
    profile: ComputeProfile | None,
    families: Sequence[str],
    n_cells: int,
    *,
    gpu: bool = False,
    device_bytes: int | None = None,
    static_gpu_fraction: float | None = None,
    target_cells_per_slot: int = _DEFAULT_TARGET_CELLS_PER_SLOT,
    max_executors: int | None = None,
    tier: str = "standard",
    pin_threads: bool = True,
) -> tuple[RuntimeResourcePlan, ServerlessTranslation]:
    """Size a Serverless batch and spell it as properties, in one call (pure).

    Two passes, because the executor's shape and the fleet's size each depend on the other.
    The slot fixes ``executor.cores`` on its own — that is the whole point of the inversion,
    the cores come from the measured fraction rather than from the fleet — so pass one
    translates against the *widest* legal executor purely to learn the core count, and pass two
    builds the real `serverless_unit` from it and plans the fleet properly. The same shape
    `ray_io.plan_cluster` uses for its autoscaling ceiling, and for the same reason.

    Pass one deliberately clamps against the widest shape rather than a nominal one: clamping
    to a small executor first would cap the measured slot *before* it had a chance to ask for a
    bigger executor, and the batch would then be sized to the clamp instead of to the work. The
    slot is re-derived against the chosen shape in pass two, so the plan that comes out is still
    clamped to the executor it will actually run on.

    ``max_executors`` is the operator's explicit cap (``--max-executors``). Left ``None`` the
    ceiling is the **saturating** count — enough executors to run every cell at once — rather
    than a derived guess. That asymmetry is deliberate: Serverless already defaults
    ``maxExecutors`` to 1000, so a ceiling we invent can only ever make a large run *slower*
    than leaving it alone, while a floor we raise can only make it faster.

    ``families`` is every family the batch will run, merged the same way a shared Ray pool's
    is (`merge_slots`): one executor shape has to hold whichever cell arrives.
    """
    label = "+".join(families) or "cpu"

    def sized_for(unit: UnitShape) -> ResourceSlot:
        return merge_slots(
            [
                resource_slot(
                    profile,
                    family,
                    use_gpu=gpu,
                    device_bytes=device_bytes,
                    static_gpu_fraction=static_gpu_fraction,
                    max_cores=unit.cores,
                    max_memory_bytes=max_slot_memory_bytes(unit),
                )
                for family in (families or ["cpu"])
            ],
            family=label,
        )

    widest = serverless_unit((_SERVERLESS_L4_CORES if gpu else _SERVERLESS_CPU_CORES)[-1], gpu=gpu)
    first = translate_serverless(
        plan_fleet(sized_for(widest), runtime="serverless", n_cells=n_cells, unit=widest),
        tier=tier,
        pin_threads=pin_threads,
    )

    unit = serverless_unit(first.executor_cores, gpu=gpu)
    slot = sized_for(unit)
    # Density comes from the executor's task slots, not from `slots_per_unit`'s device
    # arithmetic — see `spark_tasks_per_executor`. Both the ceiling and the plan use it, so
    # the fleet and the properties describe one machine rather than two.
    per_unit = spark_tasks_per_executor(slot, unit.cores)
    ceiling = max_executors or (math.ceil(n_cells / per_unit) if n_cells > 0 else 1)
    plan = plan_fleet(
        slot,
        runtime="serverless",
        n_cells=n_cells,
        unit=unit,
        target_cells_per_slot=target_cells_per_slot,
        min_units=_SERVERLESS_MIN_EXECUTORS,
        max_units=max(_SERVERLESS_MIN_EXECUTORS, ceiling),
        density=per_unit,
    )
    return plan, translate_serverless(plan, tier=tier, pin_threads=pin_threads)

serverless_gpu_executor_memory_mb

serverless_gpu_executor_memory_mb(cores)

The spark.executor.memory a Serverless L4 executor may ask for at cores (pure).

The per-config maximum is stated per core and applies to spark.executor.memory and spark.executor.memoryOverhead summed, not to memory alone. Overhead is service-owned on the GPU path at 40% of memory, so the memory we are allowed to name is the max divided by 1.4 rather than the max itself — ask for the max and the derived overhead pushes the pair over it, which the service rejects at submit with INVALID_ARGUMENT.

At the 4-core shape this returns 9560 MB, which is exactly what Serverless resolves a GPU batch to when left to itself (observed on sf-smoke-03-serverless-gpu-92763e0f2242-deep-learning-a1, 2026-09-09). That agreement is the evidence that both constants above are right, and it is why naming this value explicitly is a restatement of the platform's own default rather than a second opinion about it.

Source code in src/scale_forecasting/resources/serverless.py

def serverless_gpu_executor_memory_mb(cores: int) -> int:
    """The ``spark.executor.memory`` a Serverless L4 executor may ask for at ``cores`` (pure).

    The per-config maximum is stated per core and applies to ``spark.executor.memory`` *and*
    ``spark.executor.memoryOverhead`` **summed**, not to memory alone. Overhead is service-owned
    on the GPU path at 40% of memory, so the memory we are allowed to name is the max divided by
    1.4 rather than the max itself — ask for the max and the derived overhead pushes the pair over
    it, which the service rejects at submit with ``INVALID_ARGUMENT``.

    At the 4-core shape this returns 9560 MB, which is exactly what Serverless resolves a GPU batch
    to when left to itself (observed on `sf-smoke-03-serverless-gpu-92763e0f2242-deep-learning-a1`,
    2026-09-09). That agreement is the evidence that both constants above are right, and it is why
    naming this value explicitly is a restatement of the platform's own default rather than a
    second opinion about it.
    """
    return int(cores * _SERVERLESS_L4_MB_PER_CORE / (1 + _SERVERLESS_PYSPARK_OVERHEAD_RATIO))

serverless_unit

serverless_unit(cores, *, gpu)

The executor a given core count buys, as a UnitShape (pure).

Serverless bills the executor as a shape rather than a machine type, so memory follows the tier band rather than a GCE ratio: the standard-tier per-core maximum is what an executor can actually be given. accelerators comes from the L4 table, where the device count is a property of the core count and not a separate knob.

Source code in src/scale_forecasting/resources/serverless.py

def serverless_unit(cores: int, *, gpu: bool) -> UnitShape:
    """The executor a given core count buys, as a `UnitShape` (pure).

    Serverless bills the executor as a shape rather than a machine type, so memory follows
    the tier band rather than a GCE ratio: the standard-tier per-core maximum is what an
    executor can actually be given. ``accelerators`` comes from the L4 table, where the
    device count is a property of the core count and not a separate knob.
    """
    per_core = _SERVERLESS_L4_MB_PER_CORE if gpu else _SERVERLESS_MAX_MB_PER_CORE["standard"]
    return UnitShape(
        cores=cores,
        memory_bytes=cores * per_core * _MIB,
        accelerators=_SERVERLESS_L4_GPUS_PER_EXECUTOR.get(cores, 0) if gpu else 0,
    )

snap_to_legal

snap_to_legal(value, choices, *, up)

Nearest legal value to value in the given direction (pure).

up=True returns the smallest choice >= value (falling back to the largest when the ideal is off the top of the table); up=False the largest <= value (falling back to the smallest). Direction is the caller's call because it encodes which way is safe, and that differs per knob: memory snaps up, tasks-per-device snaps down.

Source code in src/scale_forecasting/resources/serverless.py

def snap_to_legal(value: float, choices: Sequence[int], *, up: bool) -> int:
    """Nearest legal value to ``value`` in the given direction (pure).

    ``up=True`` returns the smallest choice ``>= value`` (falling back to the largest when
    the ideal is off the top of the table); ``up=False`` the largest ``<= value`` (falling
    back to the smallest). Direction is the caller's call because it encodes which way is
    *safe*, and that differs per knob: memory snaps up, tasks-per-device snaps down.
    """
    legal = sorted(choices)
    if not legal:
        raise ValueError("snap_to_legal needs at least one legal value")
    if up:
        return next((c for c in legal if c >= value), legal[-1])
    return next((c for c in reversed(legal) if c <= value), legal[0])

spark_tasks_per_executor

spark_tasks_per_executor(slot, cores)

Cells one Serverless executor runs at once (pure) — the number Spark will really honour.

spark.task.cpus divides an executor's cores into task slots, and that division is the only concurrency a Spark executor enforces. Nothing on the batch path tells Spark about devices or about per-task memory, so the cores are the whole story here, and both the Serverless planner and the translation size against this one number rather than each deriving their own.

This is the cores bound alone, and on a GPU slot it can exceed the device bound. slots_per_unit answers the same question for a Ray pool and takes the smallest of three terms — device (accelerators x floor(1 / gpu_fraction)), cores, and memory. This takes one. At a measured fraction of 0.5, one card seats 2 cells while an 8-core executor with a 1-core slot reports 8, and the extra 6 will contend for the device rather than be refused by the scheduler. That divergence is deliberate, not an oversight:

• Ray is told a fraction and enforces it. Spark is not, so a bound stated here would have to be imposed by raising spark.task.cpus — a real change to executor shape and count.
• The fraction is usually a nominal fallback rather than a measurement. Enforcing a 2-cells-per-card bound derived from a guess would multiply the GPU-executor count of every unprofiled run, which is the live-green path. Whether to enforce it is therefore gated on "gpu_fraction" in slot.measured, and on comparing the cost both ways on live hardware — correctness is offline, value is not.

Until then the honest reading of this number is "task slots the executor will hand out", not "cells the hardware can serve".

Source code in src/scale_forecasting/resources/serverless.py

def spark_tasks_per_executor(slot: ResourceSlot, cores: int) -> int:
    """Cells one Serverless executor runs at once (pure) — the number Spark will really honour.

    ``spark.task.cpus`` divides an executor's cores into task slots, and that division is the
    *only* concurrency a Spark executor enforces. Nothing on the batch path tells Spark about
    devices or about per-task memory, so the cores are the whole story here, and both the
    Serverless planner and the translation size against this one number rather than each
    deriving their own.

    **This is the cores bound alone, and on a GPU slot it can exceed the device bound.**
    `slots_per_unit` answers the same question for a Ray pool and takes the smallest of three
    terms — device (``accelerators x floor(1 / gpu_fraction)``), cores, and memory. This takes
    one. At a measured fraction of 0.5, one card seats 2 cells while an 8-core executor with a
    1-core slot reports 8, and the extra 6 will contend for the device rather than be refused by
    the scheduler. That divergence is deliberate, not an oversight:

    * Ray is *told* a fraction and enforces it. Spark is not, so a bound stated here would have
      to be imposed by raising ``spark.task.cpus`` — a real change to executor shape and count.
    * The fraction is usually a nominal fallback rather than a measurement. Enforcing a
      2-cells-per-card bound derived from a guess would multiply the GPU-executor count of every
      unprofiled run, which is the live-green path. Whether to enforce it is therefore gated on
      ``"gpu_fraction" in slot.measured``, and on comparing the cost both ways on live hardware —
      correctness is offline, *value* is not.

    Until then the honest reading of this number is "task slots the executor will hand out", not
    "cells the hardware can serve".
    """
    return max(1, cores // max(1, min(slot.cores, cores)))

translate_serverless

translate_serverless(
    plan, *, tier="standard", pin_threads=True
)

Spell one RuntimeResourcePlan as Dataproc Serverless properties (pure).

The same three measured numbers as the Ray translation, solved backwards. Ray is told a fraction directly; Serverless derives the fraction from spark.executor.cores, so here we solve for the cores that yield the fraction we measured and snap to a legal value.

Four things this sets that the submit path never has:

1. executor.cores from evidence. GPU: the largest legal shape whose tasks-per-device fits the measured footprint (_serverless_gpu_cores). CPU: the slot's core requirement snapped up to a legal value — which lands on the 4-core default whenever the cores axis measures 1, and says so from evidence.
2. Memory in memoryOverhead, not spark.executor.memory. Every fit runs in the Python worker, outside the JVM heap, and PySpark charges that to overhead — which is why the service defaults it to 40% rather than 10%. Sizing executor.memory from peak_rss would hand the JVM memory it never touches while the Python worker got OOM-killed. On the GPU path the service rejects an explicit overhead, so the control is inverted: set executor.memory such that the 40% derivation lands on the Python memory we need — then clamp, because dividing by 0.4 puts an ordinary deep-learning footprint past the per-config maximum.
3. A warm, fast autoscaler. initialExecutors at the derived count instead of the floor of 2, and executorAllocationRatio at 1.0 instead of the default 0.3.
4. Concurrency stated once, consistently. spark.task.cpus (when the family really is threaded), the catalog.intraop_env_vars pin, and the memory budget all derive from the same tasks_per_executor. A Spark executor pins nothing on its own, so without it the sizing is sound and the run still thrashes. Ray reaches the same place by a different route — fleet.RuntimeResourcePlan.task_options puts the pin on a per-task runtime_env — but it does not get it for free, which is what the old note here said.

Unmeasured axes stay absent, as everywhere else: with no memory measurement the memory properties are simply not emitted and Serverless' own defaults apply, which is exactly today's behaviour. tier is "standard" or "premium" and only widens the legal per-core memory band.

Source code in src/scale_forecasting/resources/serverless.py

def translate_serverless(
    plan: RuntimeResourcePlan, *, tier: str = "standard", pin_threads: bool = True
) -> ServerlessTranslation:
    """Spell one `RuntimeResourcePlan` as Dataproc Serverless properties (pure).

    The same three measured numbers as the Ray translation, solved backwards. Ray is told a
    fraction directly; Serverless derives the fraction from ``spark.executor.cores``, so here
    we solve for the cores that *yield* the fraction we measured and snap to a legal value.

    Four things this sets that the submit path never has:

    1. **``executor.cores`` from evidence.** GPU: the largest legal shape whose
       tasks-per-device fits the measured footprint (`_serverless_gpu_cores`). CPU: the
       slot's core requirement snapped up to a legal value — which lands on the 4-core
       default whenever the cores axis measures 1, and says so from evidence.
    2. **Memory in ``memoryOverhead``, not ``spark.executor.memory``.** Every fit runs in the
       Python worker, outside the JVM heap, and PySpark charges that to overhead — which is
       why the service defaults it to 40% rather than 10%. Sizing ``executor.memory`` from
       ``peak_rss`` would hand the JVM memory it never touches while the Python worker got
       OOM-killed. On the **GPU** path the service rejects an explicit overhead, so the
       control is inverted: set ``executor.memory`` such that the 40% derivation lands on
       the Python memory we need — then clamp, because dividing by 0.4 puts an ordinary
       deep-learning footprint past the per-config maximum.
    3. **A warm, fast autoscaler.** ``initialExecutors`` at the derived count instead of the
       floor of 2, and ``executorAllocationRatio`` at 1.0 instead of the default 0.3.
    4. **Concurrency stated once, consistently.** ``spark.task.cpus`` (when the family really
       is threaded), the `catalog.intraop_env_vars` pin, and the memory budget all derive from
       the same ``tasks_per_executor``. A Spark executor pins nothing on its own, so without it
       the sizing is sound and the run still thrashes. Ray reaches the same place by a different
       route — `fleet.RuntimeResourcePlan.task_options` puts the pin on a per-task
       ``runtime_env`` — but it does not get it for free, which is what the old note here said.

    Unmeasured axes stay absent, as everywhere else: with no memory measurement the memory
    properties are simply not emitted and Serverless' own defaults apply, which is exactly
    today's behaviour. ``tier`` is ``"standard"`` or ``"premium"`` and only widens the
    legal per-core memory band.
    """
    slot = plan.slot
    notes: list[str] = []
    properties: dict[str, str] = {}

    if slot.gpu_fraction is not None:
        cores, ideal_cores, per_device = _serverless_gpu_cores(slot.gpu_fraction)
        wanted = 1.0 / max(slot.gpu_fraction, _MIN_GPU_FRACTION)
        if per_device > wanted:
            notes.append(
                f"serverless packs {per_device} cells per L4 at its smallest legal shape; "
                f"the measurement allows {wanted:.1f} — expect device pressure"
            )
    else:
        cores = snap_to_legal(slot.cores, _SERVERLESS_CPU_CORES, up=True)
        ideal_cores = float(slot.cores)
        per_device = None
    properties["spark.executor.cores"] = str(cores)
    properties["spark.driver.cores"] = str(snap_to_legal(4, _SERVERLESS_CPU_CORES, up=True))

    # Cores per task, and therefore cells running at once inside one executor. Spark's default
    # is one task per core, which is right for the single-threaded case the probe measures and
    # wrong for a genuinely threaded family: without ``spark.task.cpus`` Spark would hand that
    # family one core and then watch it spawn threads it was never given room for. These three
    # numbers have to agree — tasks, the thread pin, and the memory budgeted for them — or the
    # translation over-packs on one axis while under-packing another.
    task_cpus = max(1, min(slot.cores, cores))
    tasks_per_executor = spark_tasks_per_executor(slot, cores)
    if task_cpus > 1:
        properties["spark.task.cpus"] = str(task_cpus)
    if pin_threads:
        for name, value in intraop_env_vars(task_cpus).items():
            properties[f"spark.executorEnv.{name}"] = value
    else:
        notes.append(
            "native thread pools left uncapped so effective_cores can be measured; "
            "executors are deliberately oversubscribed — do not size a real run from this shape"
        )

    if slot.memory_bytes:
        python_mb = math.ceil(tasks_per_executor * slot.memory_bytes / _MIB)
        jvm_mb = cores * _SPARK_JVM_MB_PER_CORE
        if per_device is None:
            per_core = (python_mb + jvm_mb) / cores
            ceiling = _SERVERLESS_MAX_MB_PER_CORE.get(tier, _SERVERLESS_MAX_MB_PER_CORE["standard"])
            if per_core > ceiling:
                notes.append(
                    f"{per_core:.0f}m per core exceeds the {tier}-tier ceiling of {ceiling}m; "
                    "clamped — use premium compute or a smaller slot"
                )
                python_mb = max(1, int(cores * ceiling) - jvm_mb)
            elif per_core < _SERVERLESS_MIN_MB_PER_CORE:
                python_mb = cores * _SERVERLESS_MIN_MB_PER_CORE - jvm_mb
            properties["spark.executor.memory"] = f"{jvm_mb}m"
            properties["spark.executor.memoryOverhead"] = f"{python_mb}m"
        else:
            # GPU: overhead is service-owned at 40% of memory, so ask for the memory whose
            # 40% is the Python footprint we need, and let the service derive the rest. The
            # result is still bounded — the inversion divides by 0.4, so an ordinary
            # deep-learning footprint lands well past the per-config maximum if left raw.
            memory_mb = math.ceil(python_mb / _SERVERLESS_PYSPARK_OVERHEAD_RATIO)
            ceiling = serverless_gpu_executor_memory_mb(cores)
            if memory_mb > ceiling:
                notes.append(
                    f"{memory_mb}m exceeds the extrapolated L4 maximum of {ceiling}m at "
                    f"{cores} cores; clamped — the cells will contend for host memory"
                )
                memory_mb = ceiling
            memory_mb = max(memory_mb, cores * _SERVERLESS_MIN_MB_PER_CORE)
            properties["spark.executor.memory"] = f"{memory_mb}m"
            notes.append("gpu batch: memoryOverhead is service-owned, sized via executor.memory")
    else:
        notes.append("memory unmeasured; serverless memory defaults left in place")

    lo, initial, hi = _executor_counts(plan)
    properties["spark.dynamicAllocation.enabled"] = "true"
    properties["spark.dynamicAllocation.minExecutors"] = str(lo)
    properties["spark.dynamicAllocation.initialExecutors"] = str(initial)
    properties["spark.dynamicAllocation.maxExecutors"] = str(hi)
    ratio = _SERVERLESS_FAST_ALLOCATION_RATIO if hi > initial else None
    if ratio is not None:
        properties["spark.dynamicAllocation.executorAllocationRatio"] = str(ratio)
    if plan.max_units > _SERVERLESS_MAX_EXECUTORS:
        notes.append(f"executor ceiling clamped to the platform max of {_SERVERLESS_MAX_EXECUTORS}")

    return ServerlessTranslation(
        executor_cores=cores,
        ideal_executor_cores=ideal_cores,
        tasks_per_device=per_device,
        properties=properties,
        notes=tuple(notes) + slot.notes,
    )

cluster

Dataproc cluster — the worker is the unit, and it is billed whole.

The second Spark translator, and the differences from serverless are all consequences of one fact: here we choose the machine, and we pay for the whole of it whether or not the arithmetic filled it. Serverless rents executors from a legal-value table; a cluster rents machines from GCE, so the shapes come from catalog's machine tables and the knobs are continuous within what the worker actually has.

The two translators are peers. Everything they must agree about — the JVM heap floor, the thread-pin variable names, the mebibyte — lives one layer down in catalog, so neither imports the other and neither can drift.

ClusterTranslation dataclass

ClusterTranslation(
    executor_cores,
    tasks_per_executor,
    worker_count,
    ideal_workers,
    properties,
    notes=(),
)

One plan spelled as Dataproc cluster job properties plus a worker count.

The cluster analog of ServerlessTranslation, and it carries one thing more: on a batch the fleet is the properties, while a cluster's ceiling is a physical worker count fixed at create. So worker_count is a separate output that the create path consumes and properties is what the job carries.

ideal_workers is what the fan-out asked for before _CLUSTER_MAX_WORKERS (or the operator's cap) applied. When the two differ the run is throttled by a spend ceiling rather than by its own shape, and that is worth being able to read off a record.

to_dict

to_dict()

JSON-safe record of the translation — what telemetry stamps alongside the plan.

Source code in src/scale_forecasting/resources/cluster.py

def to_dict(self) -> dict[str, Any]:
    """JSON-safe record of the translation — what telemetry stamps alongside the plan."""
    return {
        "executor_cores": self.executor_cores,
        "tasks_per_executor": self.tasks_per_executor,
        "worker_count": self.worker_count,
        "ideal_workers": self.ideal_workers,
        "properties": dict(self.properties),
        "notes": list(self.notes),
    }

cluster_unit

cluster_unit(machine_type, *, accelerators=0)

The worker a GCE machine type buys, as a UnitShape (pure).

Straight from the machine name, because on a cluster that name is the knob — there is no legal-value table to snap to as there is on Serverless, and no tier band: an n1-standard-8 worker has eight cores and 30 GiB, and the sizing question is only how to divide them. accelerators is devices per worker (AcceleratorConfig.accelerator_count, 1 on every GPU cluster we build today).

Source code in src/scale_forecasting/resources/cluster.py

def cluster_unit(machine_type: str, *, accelerators: int = 0) -> UnitShape:
    """The worker a GCE machine type buys, as a `UnitShape` (pure).

    Straight from the machine name, because on a cluster that name *is* the knob — there is no
    legal-value table to snap to as there is on Serverless, and no tier band: an
    ``n1-standard-8`` worker has eight cores and 30 GiB, and the sizing question is only how to
    divide them. ``accelerators`` is devices per worker (`AcceleratorConfig.accelerator_count`,
    1 on every GPU cluster we build today).
    """
    return UnitShape(
        cores=machine_cores(machine_type),
        memory_bytes=machine_memory_bytes(machine_type),
        accelerators=accelerators,
    )

plan_dataproc_cluster

plan_dataproc_cluster(
    profile,
    families,
    n_cells,
    *,
    machine_type,
    accelerators=0,
    gpu=False,
    device_bytes=None,
    static_gpu_fraction=None,
    target_cells_per_slot=_DEFAULT_TARGET_CELLS_PER_SLOT,
    max_workers=None,
    pin_threads=True,
)

Size a Dataproc cluster and spell it as job properties, in one call (pure).

One pass, where plan_serverless needs two: there is no chicken-and-egg to break, because the executor's shape follows from the worker's machine type and the caller already chose that (the accelerator decides it — a T4 rides an n1-standard-8, an L4 comes bundled in a g2-standard-8). So the shape is known before the slot is sized, and the slot can be clamped to the executor it will actually run in on the first and only attempt.

n_cells is the task count, not the cell count — the bucket count the engine will fan out into (engines.spark_io.default_bucket_count), for the same reason the batch path sizes against it: each bucket holds compute.bucket_target_cells cells that run sequentially inside one pandas frame, so sizing against cells would ask for that many times more workers than the fan-out can keep busy.

max_workers is the operator's explicit ceiling. Left None it is _CLUSTER_MAX_WORKERS — a real clamp rather than the batch path's "saturating count", because the two runtimes have opposite cost shapes: an unused Serverless executor costs nothing, an unused worker costs a VM-hour.

families is every family whose work lands on this cluster, merged the way a shared Ray pool's is (merge_slots): one executor shape has to hold whichever cell arrives.

Source code in src/scale_forecasting/resources/cluster.py

def plan_dataproc_cluster(
    profile: ComputeProfile | None,
    families: Sequence[str],
    n_cells: int,
    *,
    machine_type: str,
    accelerators: int = 0,
    gpu: bool = False,
    device_bytes: int | None = None,
    static_gpu_fraction: float | None = None,
    target_cells_per_slot: int = _DEFAULT_TARGET_CELLS_PER_SLOT,
    max_workers: int | None = None,
    pin_threads: bool = True,
) -> tuple[RuntimeResourcePlan, ClusterTranslation]:
    """Size a Dataproc cluster and spell it as job properties, in one call (pure).

    One pass, where `plan_serverless` needs two: there is no chicken-and-egg to break, because
    the executor's shape follows from the worker's machine type and the caller already chose
    that (the accelerator decides it — a T4 rides an ``n1-standard-8``, an L4 comes bundled in
    a ``g2-standard-8``). So the shape is known before the slot is sized, and the slot can be
    clamped to the executor it will actually run in on the first and only attempt.

    ``n_cells`` is the **task** count, not the cell count — the bucket count the engine will
    fan out into (`engines.spark_io.default_bucket_count`), for the same reason the batch path
    sizes against it: each bucket holds `compute.bucket_target_cells` cells that run
    sequentially inside one pandas frame, so sizing against cells would ask for that many times
    more workers than the fan-out can keep busy.

    ``max_workers`` is the operator's explicit ceiling. Left ``None`` it is
    `_CLUSTER_MAX_WORKERS` — a real clamp rather than the batch path's "saturating count",
    because the two runtimes have opposite cost shapes: an unused Serverless executor costs
    nothing, an unused worker costs a VM-hour.

    ``families`` is every family whose work lands on this cluster, merged the way a shared Ray
    pool's is (`merge_slots`): one executor shape has to hold whichever cell arrives.
    """
    label = "+".join(families) or "cpu"
    unit = cluster_unit(machine_type, accelerators=accelerators)
    cores, _jvm_mb, pool_mb = _cluster_executor(unit)

    slot = merge_slots(
        [
            resource_slot(
                profile,
                family,
                use_gpu=gpu,
                device_bytes=device_bytes,
                static_gpu_fraction=static_gpu_fraction,
                max_cores=cores,
                max_memory_bytes=pool_mb * _MIB if pool_mb is not None else None,
            )
            for family in (families or ["cpu"])
        ],
        family=label,
    )
    # The density the properties will actually spell (`_cluster_density`), not the one the
    # device arithmetic asks for — the fleet and the job have to describe one machine. One
    # executor per worker, so a worker's density *is* an executor's density.
    _cores, _task_cpus, per_unit = _cluster_density(slot, unit)
    ceiling = max(_CLUSTER_MIN_WORKERS, max_workers or _CLUSTER_MAX_WORKERS)
    plan = plan_fleet(
        slot,
        runtime="cluster",
        n_cells=n_cells,
        unit=unit,
        target_cells_per_slot=target_cells_per_slot,
        min_units=_CLUSTER_MIN_WORKERS,
        max_units=ceiling,
        density=per_unit,
    )
    return plan, translate_cluster(plan, pin_threads=pin_threads)

translate_cluster

translate_cluster(plan, *, pin_threads=True)

Spell one RuntimeResourcePlan as Dataproc cluster job properties (pure).

The third algebra over the same three measured numbers. The Serverless inversion does not apply here — a cluster's executor shape is not chosen from a legal table, it is carved out of a machine we already picked — so this is the simplest of the three and differs from translate_serverless in exactly four places:

1. Cores come from the worker, not from the slot. One executor per worker (see _CLUSTER_AM_CORES for why "almost"), so spark.executor.cores is a property of the machine type. The slot still decides spark.task.cpus, and therefore how many cells run at once inside that executor — the number that actually matters.
2. Memory is always stated, measured or not. Everywhere else an unmeasured axis requests nothing, and that rule is right when the platform's own default is shaped for the request we are making. Here it is not: Dataproc bakes spark.executor.memory into the cluster's spark-defaults at create, sized for the default executor shape. Change the shape at job level and leave memory alone and the pairing is stale — a 7-core executor holding a 4-core executor's heap. So the container is sized from the worker (a machine fact, not a measurement) and the split between JVM heap and Python overhead is what measurement refines.
3. The GPU path uses memoryOverhead like everything else. Serverless forbids setting it on a GPU batch and forces the awkward executor.memory inversion; a cluster has no such restriction, so the GPU and CPU paths are the same code.
4. The fleet is workers, not executors. dynamicAllocation still runs (it is what lets a job release executors it is not using) but initialExecutors equals maxExecutors: the workers are already paid for, so there is no ramp worth taking.

What is deliberately not emitted: spark.executor.resource.gpu.amount and spark.task.resource.gpu.amount, the design's GPU-aware scheduling pair. Spark validates an executor resource request against what the NodeManager advertises, and Dataproc does not turn YARN GPU isolation on — no yarn.resource-types, no resource plugin, no discovery script. Emitting the Spark half alone fails every executor at launch. One executor per worker already gives the device a single owner, and spark.task.cpus is what shares it; turning the YARN half on is a cluster-create change that has to be proven live, not an offline one.

Source code in src/scale_forecasting/resources/cluster.py

def translate_cluster(plan: RuntimeResourcePlan, *, pin_threads: bool = True) -> ClusterTranslation:
    """Spell one `RuntimeResourcePlan` as Dataproc **cluster** job properties (pure).

    The third algebra over the same three measured numbers. The Serverless inversion does not
    apply here — a cluster's executor shape is not chosen from a legal table, it is carved out
    of a machine we already picked — so this is the simplest of the three and differs from
    `translate_serverless` in exactly four places:

    1. **Cores come from the worker, not from the slot.** One executor per worker (see
       `_CLUSTER_AM_CORES` for why "almost"), so ``spark.executor.cores`` is a property of
       the machine type. The slot still decides ``spark.task.cpus``, and therefore how many
       cells run at once inside that executor — the number that actually matters.
    2. **Memory is always stated, measured or not.** Everywhere else an unmeasured axis
       requests nothing, and that rule is right when the platform's own default is shaped for
       the request we are making. Here it is not: Dataproc bakes ``spark.executor.memory``
       into the cluster's ``spark-defaults`` at *create*, sized for the default executor
       shape. Change the shape at job level and leave memory alone and the pairing is stale —
       a 7-core executor holding a 4-core executor's heap. So the container is sized from the
       worker (a machine fact, not a measurement) and the *split* between JVM heap and Python
       overhead is what measurement refines.
    3. **The GPU path uses ``memoryOverhead`` like everything else.** Serverless forbids
       setting it on a GPU batch and forces the awkward ``executor.memory`` inversion; a
       cluster has no such restriction, so the GPU and CPU paths are the same code.
    4. **The fleet is workers, not executors.** ``dynamicAllocation`` still runs (it is what
       lets a job release executors it is not using) but ``initialExecutors`` equals
       ``maxExecutors``: the workers are already paid for, so there is no ramp worth taking.

    **What is deliberately not emitted:** ``spark.executor.resource.gpu.amount`` and
    ``spark.task.resource.gpu.amount``, the design's GPU-aware scheduling pair. Spark validates
    an executor resource request against what the NodeManager advertises, and Dataproc does not
    turn YARN GPU isolation on — no ``yarn.resource-types``, no resource plugin, no discovery
    script. Emitting the Spark half alone fails every executor at launch. One executor per
    worker already gives the device a single owner, and ``spark.task.cpus`` is what shares it;
    turning the YARN half on is a cluster-create change that has to be proven live, not an
    offline one.
    """
    slot = plan.slot
    unit = plan.unit
    notes: list[str] = []
    properties: dict[str, str] = {}

    _cores, jvm_mb, pool_mb = _cluster_executor(unit)
    cores, task_cpus, tasks = _cluster_density(slot, unit)
    properties["spark.executor.cores"] = str(cores)

    # Tasks, the native-thread pin, and the memory budgeted for them all derive from this one
    # number, exactly as on the Serverless side — otherwise the executor is over-packed on one
    # axis while under-packed on another.
    if task_cpus > 1:
        properties["spark.task.cpus"] = str(task_cpus)
    if pin_threads:
        for name, value in intraop_env_vars(task_cpus).items():
            properties[f"spark.executorEnv.{name}"] = value
    else:
        notes.append(
            "native thread pools left uncapped so effective_cores can be measured; "
            "executors are deliberately oversubscribed — do not size a real run from this shape"
        )

    if pool_mb is None:
        notes.append("worker machine type unparseable; cluster memory defaults left in place")
    else:
        python_mb = pool_mb
        if slot.memory_bytes:
            wanted = math.ceil(tasks * slot.memory_bytes / _MIB)
            if wanted > python_mb:
                notes.append(
                    f"one cell needs {slot.memory_bytes} bytes and the worker leaves "
                    f"{python_mb}m for {tasks} of them — clamped, expect host memory pressure"
                )
            else:
                python_mb = max(1, wanted)
        properties["spark.executor.memory"] = f"{jvm_mb}m"
        properties["spark.executor.memoryOverhead"] = f"{python_mb}m"

    workers = plan.derived_units or _CLUSTER_MIN_WORKERS
    ideal = (
        math.ceil(plan.n_cells / max(1, plan.slots_per_unit * plan.target_cells_per_slot))
        if plan.n_cells > 0
        else 0
    )
    if ideal > plan.max_units:
        notes.append(
            f"the fan-out wants {ideal} workers; clamped to {plan.max_units} because a "
            "cluster's workers bill whether or not they are busy — raise max_workers to spend it"
        )
    properties["spark.dynamicAllocation.enabled"] = "true"
    properties["spark.dynamicAllocation.minExecutors"] = "1"
    properties["spark.dynamicAllocation.initialExecutors"] = str(workers)
    properties["spark.dynamicAllocation.maxExecutors"] = str(workers)

    if unit.accelerators:
        notes.append(
            "gpu cluster: spark.{executor,task}.resource.gpu.amount withheld — dataproc leaves "
            "yarn gpu isolation off, so a resource request the nodemanager does not advertise "
            "fails every executor at launch; the single executor owns the device instead"
        )

    return ClusterTranslation(
        executor_cores=cores,
        tasks_per_executor=tasks,
        worker_count=workers,
        ideal_workers=ideal,
        properties=properties,
        notes=tuple(notes) + slot.notes,
    )

catalog

The numbers no measurement produced — fallbacks, machine shapes, and two Spark facts.

The bottom of this package, and it is here because the layers above it would otherwise have to import each other sideways. Three groups, each a different kind of "we did not measure this":

• Fallbacks — what a slot is worth when the profile has no basis for it. Every value reproduces what the engines hardcoded before the profiler existed, which is the safety property: enabling measurement can add information, never make an unmeasured run worse.
• Machine shapes — the RAM a GCE machine type implies, because neither Ray nor Vertex will tell us a node's size and the type name is the only handle there is.
• Spark platform facts — the JVM heap floor and the thread-pin variable names. Shared because serverless and cluster are peers: two Spark translators that must agree about the platform, with neither importing the other.

Both fallback directions are chosen, not defaulted. Guessing a node's memory low under-packs it (we pay for capacity we don't use); guessing high over-packs it (the run OOMs). Only one of those is recoverable, and every unknown here leans that way.

device_floor_fraction

device_floor_fraction(schedulable_cores, accelerators)

The smallest GPU fraction worth asking for on a node of this shape (pure).

_MIN_GPU_FRACTION unless the node's cores would have run more cells than that floor permits, in which case it drops far enough to let them.

Why a flat floor was wrong. 0.1 caps a device at ten concurrent cells, and the justification for it — below this a task barely uses the card, so packing overhead dominates — describes a model that is actually using the accelerator. The deep-learning family we ship is not: a NeuralProphet fit peaks at 50–78 KB on a 16 GB card, so the honest fraction is about six millionths and every auto-calibrated run lands on the floor. The fraction stopped being a measurement and became a constant. On an eight-core node that was invisible, because seven usable cores bind before ten cells do; on a sixteen-core node with the same single card the floor binds at ten and five cores sit idle, and binding_axis reports device — sending an operator to change a GPU setting when the real limit is a number in this file (see docs/validation.md).

What replaces it. Cores are the density a node can genuinely sustain — one cell needs a core to run on whatever the card thinks — so the floor is never allowed to sit above the fraction those cores imply. ceil on the per-device share keeps the device bound at or above the core bound rather than one short of it when a node carries several cards. The floor only ever moves down: a pinned fraction an operator chose is theirs, and a node too small to want the relaxation keeps the default it always had.

Only the Ray path passes this today, and that is deliberate. Both Spark translators own a second density derivation of their own (cluster._cluster_density, serverless._serverless_gpu_cores), each live-proven on its platform's executor shapes, and relaxing a floor underneath arithmetic this has not been measured against would be trading a known conservative answer for an unmeasured one. They keep the flat default by not passing an argument. fleet.plan_resources — the generic, runtime-neutral entry point — does pass it.

accelerators <= 0 returns the default unchanged — there is no device axis to bound.

Source code in src/scale_forecasting/resources/catalog.py

def device_floor_fraction(schedulable_cores: int, accelerators: int) -> float:
    """The smallest GPU fraction worth asking for on a node of this shape (pure).

    `_MIN_GPU_FRACTION` unless the node's cores would have run more cells than that floor
    permits, in which case it drops far enough to let them.

    **Why a flat floor was wrong.** ``0.1`` caps a device at ten concurrent cells, and the
    justification for it — below this a task barely uses the card, so packing overhead
    dominates — describes a model that is actually using the accelerator. The deep-learning
    family we ship is not: a NeuralProphet fit peaks at 50–78 KB on a 16 GB card, so the honest
    fraction is about six millionths and *every* auto-calibrated run lands on the floor. The
    fraction stopped being a measurement and became a constant. On an eight-core node that was
    invisible, because seven usable cores bind before ten cells do; on a sixteen-core node with
    the same single card the floor binds at ten and five cores sit idle, and `binding_axis`
    reports ``device`` — sending an operator to change a GPU setting when the real limit is a
    number in this file (see `docs/validation.md`).

    **What replaces it.** Cores are the density a node can genuinely sustain — one cell needs a
    core to run on whatever the card thinks — so the floor is never allowed to sit above the
    fraction those cores imply. ``ceil`` on the per-device share keeps the device bound at or
    above the core bound rather than one short of it when a node carries several cards. The
    floor only ever moves *down*: a pinned fraction an operator chose is theirs, and a node too
    small to want the relaxation keeps the default it always had.

    **Only the Ray path passes this today, and that is deliberate.** Both Spark translators own a
    second density derivation of their own (`cluster._cluster_density`,
    `serverless._serverless_gpu_cores`), each live-proven on its platform's executor shapes, and
    relaxing a floor underneath arithmetic this has not been measured against would be trading a
    known conservative answer for an unmeasured one. They keep the flat default by not passing an
    argument. `fleet.plan_resources` — the generic, runtime-neutral entry point — does pass it.

    ``accelerators <= 0`` returns the default unchanged — there is no device axis to bound.
    """
    if accelerators <= 0:
        return _MIN_GPU_FRACTION
    per_device = max(1, math.ceil(max(1, schedulable_cores) / accelerators))
    return min(_MIN_GPU_FRACTION, 1.0 / per_device)

intraop_env_vars

intraop_env_vars(threads, *, include_omp=True)

{var: str(threads)} for every native thread pool a fit might use (pure).

One definition of the pin, so the two runtimes cannot drift into capping different subsets of these libraries and calling the result the same thing.

include_omp=False is the Ray caller. Ray sets OMP_NUM_THREADS itself, from the task's assigned cores, and its setter is a no-op when the variable already has a value — so passing our own would silently take ownership of a derivation Ray is already doing correctly, and any future divergence between the two would resolve in favour of ours without a word. Leaving that one variable alone keeps a single owner for it. Spark has no such setter and takes all five.

Source code in src/scale_forecasting/resources/catalog.py

def intraop_env_vars(threads: int, *, include_omp: bool = True) -> dict[str, str]:
    """``{var: str(threads)}`` for every native thread pool a fit might use (pure).

    One definition of the pin, so the two runtimes cannot drift into capping different subsets of
    these libraries and calling the result the same thing.

    ``include_omp=False`` is the Ray caller. Ray sets ``OMP_NUM_THREADS`` itself, from the task's
    assigned cores, and its setter is a no-op when the variable already has a value — so passing
    our own would silently take ownership of a derivation Ray is already doing correctly, and any
    future divergence between the two would resolve in favour of ours without a word. Leaving that
    one variable alone keeps a single owner for it. Spark has no such setter and takes all five.
    """
    return {
        name: str(threads) for name in _INTRAOP_ENV_VARS if include_omp or name != "OMP_NUM_THREADS"
    }

machine_cores

machine_cores(machine_type)

vCPUs a GCE machine type implies — n1-standard-8 → 8 (pure; unparseable → 8).

The count suffix of a <family>-<class>-<cores> name — the same shape machine_memory_bytes parses, so the two axes agree about which names they understand — plus _FIXED_MACHINE_SHAPES for accelerator-optimized shapes whose suffix names GPUs (a2-highgpu-1g → 12 vCPUs). Matching the whole shape rather than just a trailing number is what keeps a custom type (n1-custom-8-16384, where the trailing number is megabytes) from being read as a 16384-core machine.

Source code in src/scale_forecasting/resources/catalog.py

def machine_cores(machine_type: str) -> int:
    """vCPUs a GCE machine type implies — ``n1-standard-8`` → 8 (pure; unparseable → 8).

    The count suffix of a ``<family>-<class>-<cores>`` name — the same shape
    `machine_memory_bytes` parses, so the two axes agree about which names they understand —
    plus `_FIXED_MACHINE_SHAPES` for accelerator-optimized shapes whose suffix names GPUs
    (``a2-highgpu-1g`` → 12 vCPUs). Matching the whole shape rather than just a trailing number
    is what keeps a custom type (``n1-custom-8-16384``, where the trailing number is megabytes)
    from being read as a 16384-core machine.
    """
    if machine_type in _FIXED_MACHINE_SHAPES:
        return _FIXED_MACHINE_SHAPES[machine_type][0]
    match = re.match(_MACHINE_TYPE_RE, machine_type)
    return int(match.group(2)) if match else _DEFAULT_MACHINE_CORES

machine_memory_bytes

machine_memory_bytes(machine_type)

RAM a GCE machine type implies, in bytes (pure; unknown → the smallest standard ratio).

Derived as cores x GiB-per-vCPU from _MEMORY_PER_CORE_GIB, keyed on the <family>-<class> prefix — so n1-standard-8 is 30 GiB and n1-highmem-8 is 52 GiB — or taken directly from _FIXED_MACHINE_SHAPES for -Ng GPU shapes (a2-highgpu-1g → 85 GiB). Nameplate RAM, not schedulable RAM: plan_resources applies _SCHEDULABLE_MEMORY_FRACTION on top.

Two different kinds of "we don't know", kept distinct because they warrant different answers. A name whose class is untabulated but whose shape parses (some future n4-standard-8) falls back to the smallest standard ratio — it under-counts memory and therefore under-packs the node, the safe direction. A name that does not parse at all (n1-custom-8-16384, a bare alias) returns 0, meaning unknown: callers treat that as no memory bound rather than as a machine with no memory, so an unrecognised type degrades to today's cores-only packing instead of to one slot.

Source code in src/scale_forecasting/resources/catalog.py

def machine_memory_bytes(machine_type: str) -> int:
    """RAM a GCE machine type implies, in bytes (pure; unknown → the smallest standard ratio).

    Derived as ``cores x GiB-per-vCPU`` from `_MEMORY_PER_CORE_GIB`, keyed on the
    ``<family>-<class>`` prefix — so ``n1-standard-8`` is 30 GiB and ``n1-highmem-8`` is 52
    GiB — or taken directly from `_FIXED_MACHINE_SHAPES` for ``-Ng`` GPU shapes
    (``a2-highgpu-1g`` → 85 GiB). Nameplate RAM, not schedulable RAM: `plan_resources` applies
    `_SCHEDULABLE_MEMORY_FRACTION` on top.

    Two different kinds of "we don't know", kept distinct because they warrant different
    answers. A name whose *class* is untabulated but whose shape parses (some future
    ``n4-standard-8``) falls back to the smallest standard ratio — it under-counts memory
    and therefore under-packs the node, the safe direction. A name that does not parse at
    all (``n1-custom-8-16384``, a bare alias) returns **0**, meaning *unknown*: callers
    treat that as no memory bound rather than as a machine with no memory, so an
    unrecognised type degrades to today's cores-only packing instead of to one slot.
    """
    if machine_type in _FIXED_MACHINE_SHAPES:
        return int(_FIXED_MACHINE_SHAPES[machine_type][1] * _GIB)
    match = re.match(_MACHINE_TYPE_RE, machine_type)
    if match is None:
        return 0
    prefix, cores = match.group(1), int(match.group(2))
    per_core = _MEMORY_PER_CORE_GIB.get(prefix, _DEFAULT_MEMORY_PER_CORE_GIB)
    return int(cores * per_core * _GIB)