[SPARK-58192][CORE] Support fractional spark.task.cpus

[pan3793]

What changes were proposed in this pull request?

This PR adds support for fractional values of spark.task.cpus (e.g. 0.5, 1.5) and for
fractional stage-level task cpus requests, so that the number of concurrent tasks on an
executor becomes floor(executor cores / task cpus) computed exactly.

Exact decimal accounting. CPU bookkeeping moves from Int to scala.math.BigDecimal:
WorkerOffer.cores, ExecutorData.freeCores, TaskDescription.cpus, and
StatusUpdate.taskCpus all carry the exact decimal value. A new internal CpuAmount object
defines the canonical representation: every value is normalized to a fixed scale of 9
(1e-9 resolution), which keeps BigDecimal on its compact long-backed fast path and makes
arithmetic between normalized values stay at scale 9. Slot math
(ResourceProfile.numTasksBasedOnCores) uses exact FLOOR division, so e.g. 1.0 / 0.1
yields 10 slots (naive Double arithmetic would floor 9.999... to 9), and the result is
clamped to [0, Int.MaxValue] (BigDecimal.intValue() wraps modulo 2^32);
TaskSchedulerImpl.calculateAvailableSlots sums per-executor slots in Long and saturates.

Configuration. A new ConfigBuilder.decimalConf parses the exact decimal string. The
value is validated before normalization: it must be in [1e-9, Int.MaxValue] with at most
9 decimal places, so out-of-range/over-precise values (including extreme exponents like
1e100000000) fail fast with a clear error instead of being silently rounded or triggering
pathological setScale costs.

API.

• TaskContext.cpuAmount(): BigDecimal (new, @Since("4.3.0")) returns the exact amount;
  TaskContext.cpus(): Int is deprecated and now returns the ceiling of the exact amount.
• TaskResourceRequests.cpus(Double) overload (the Int overload delegates to it).
• The amount <= 1.0 || whole assertion in TaskResourceRequest and the
  amount <= 0.5 || whole assertion in ResourceProfile.validate() no longer apply to
  cpus: those rules exist so task amounts map onto discrete resource addresses (GPUs etc.),
  while cpus is a plain quantity drawn from the executor's core pool. Instead, a cpus amount
  must be at least 1e-9 (values that would round to zero at the accounting scale are
  rejected at request construction time).
• PySpark: TaskContext.cpuAmount() -> float (delivered via the task-context JSON sent to
  Python workers), TaskResourceRequests.cpus accepts float and no longer truncates
  fractional amounts when materializing requests to the JVM.

Executor / runtime.

• OMP_NUM_THREADS and the Python worker's cpus default to the ceiling of the exact amount.
• executorRunTime is scaled by the task's cpus amount and clamped at zero (with cpus < 1 the
  scaled elapsed time can be smaller than the subtracted in-run deserialization time).

Kubernetes. In allocation recovery mode (SPARK-55639), SPARK_EXECUTOR_CORES is the
ceiling of spark.task.cpus (the executor --cores argument and registration RPC are
integral). When spark.task.cpus <= 0.5 this cannot express a single-task capacity — the
recovery executor accepts floor(1 / spark.task.cpus) concurrent tasks — so the driver logs
a prominent one-time warning and the config/docs document the limitation.

Misc. ResourceProfile.getTaskCpus is cached (lazy val) and the scheduler resolves task
cpus once per (task set, locality) round instead of per slot probe; MiMa excludes are placed
in v43excludes to match the @Since("4.3.0") annotations so the change can be backported
to branch-4.x.

Why are the changes needed?

spark.task.cpus only accepted whole numbers, so the finest scheduling granularity was one
task per core. Workloads with many lightweight tasks (I/O-bound tasks, PySpark tasks whose
heavy lifting happens in a shared native library or external service) cannot efficiently use
an executor without oversubscribing memory-per-task, and workloads needing "more than 1 but
less than 2" cores per task (e.g. 1.5) had to round up to 2 and waste capacity. Fractional
task cpus let users express both:

# 8 concurrent tasks on a 4-core executor
spark-submit --conf spark.executor.cores=4 --conf spark.task.cpus=0.5 ...

# 2 concurrent tasks on a 4-core executor: floor(4 / 1.5) = 2
spark-submit --conf spark.executor.cores=4 --conf spark.task.cpus=1.5 ...

Does this PR introduce any user-facing change?

Yes.

• spark.task.cpus and TaskResourceRequests.cpus now accept fractional values; whole-number
  configurations behave exactly as before. Values outside [1e-9, Int.MaxValue] or with more
  than 9 decimal places are rejected with a clear error message.
• New API TaskContext.cpuAmount() (Scala and Python); TaskContext.cpus() is deprecated and
  documents its ceiling behavior.
• In K8s allocation recovery mode with spark.task.cpus <= 0.5, recovery executors accept
  more than one concurrent task; a warning is logged and the behavior is documented.

How was this patch tested?

New and updated unit tests:

• SparkContextSuite: config bounds/precision matrix (accepts 1.5, 1e-9, Int.MaxValue,
  trailing zeros; rejects 0, -1, 4e-10, Int.MaxValue + 1, 1e100000000,
  1e1000000000, 10 decimal places).
• ResourceProfileSuite: fractional profiles (0.7/1.5/2.5) are valid, sub-1e-9
  amounts rejected, numTasksBasedOnCores clamp behavior; existing custom-resource
  validation tests unchanged.
• TaskSchedulerImplSuite: fractional scheduling end-to-end (0.5, 0.2, and 1.5 on a
  4-core offer launching exactly 2 tasks), calculateAvailableSlots saturation.
• BasicExecutorFeatureStepSuite / ExecutorPodsAllocatorSuite (K8s): recovery-mode env
  ceiling (0.5 -> 1, 1.5 -> 2, 2.5 -> 3) and the one-time isolation warning.
• PySpark: test_taskcontext.py (cpuAmount() for whole and fractional configs),
  test_resources.py (fractional amounts preserved in TaskResourceRequests).

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Fable 5)

[pan3793]

cc @HyukjinKwon, as you also raised the same idea in #40730 (comment)

I have been wondering if it's better to allow spark.task.cpus to set a floating number like 0.2 so I/O intensive tasks can benefit from that as well as the case like this.

background: we do have observed a large set of such I/O-bound workloads on our cluster, which take a few CPU time but occupy many executor core slots

cc @dongjoon-hyun @sunchao @mridulm @tgravescs @wangyum for review, thank you in advance

[sunchao]

Summary

This change tackles a real scheduler limitation and the central fixed-scale BigDecimal accounting is thoughtfully carried through offers, task descriptions, completion updates, barrier rollback, and status reconstruction. I reviewed exact head 980835a5d6ff18bb64d47dda605b7dc720a44fc3 with separate scheduler, API/Python, runtime/Kubernetes, and adversarial verification passes.

I found four material gaps at the runtime and resource-manager boundaries. They all arise when fractional CPUs increase task concurrency or when a stage-level resource profile differs from the application-wide spark.task.cpus; I left them as inline P2 findings.

Prior state and problem

Before this patch, task CPU demand was integral, so executor core counts doubled as both a resource quantity and, in several subsystems, an implicit upper bound on concurrent tasks or workers. That prevented useful configurations such as five I/O-bound tasks sharing one core or tasks reserving 1.5 cores.

Supporting fractional CPU demand requires more than changing scheduler slot division: every downstream assumption that equates executor cores with concurrent tasks, and every integration that rereads the global configuration instead of the active task profile, must also be updated.

Design approach

The patch introduces a scale-9 decimal CPU representation, parses spark.task.cpus without binary floating-point drift, and performs slot computation with exact floor division. It threads the exact value through scheduler offers, TaskDescription, executor accounting, status updates, TaskContext, and PySpark task-context serialization.

For APIs that require an integer, such as OMP_NUM_THREADS and Kubernetes executor registration, it rounds the exact amount upward. The Kubernetes recovery path additionally documents the unavoidable global spark.task.cpus <= 0.5 isolation limitation.

Correctness / compatibility analysis

The scheduler's normal launch/finish accounting is internally symmetric: the same normalized amount is reserved and returned, barrier rollback preserves it, and slot sums saturate safely. I also found no same-version RPC serialization problem, and the existing MiMa-exclusion pattern for TaskContext additions has project precedent.

The remaining problems are boundary mismatches. PySpark memory is still divided by raw executor cores even though fractional CPUs can create more workers than cores; Python threading and Kubernetes recovery reread the global CPU value instead of the stage profile being launched; and executorRunTime scales the deserialization interval before subtracting it, which can collapse valid runtimes to zero.

Key design decisions

• Fixed-scale decimal accounting avoids cumulative drift and makes floor(cores / cpus) deterministic.
• The exact fractional value remains the scheduler contract; ceiling is appropriate only when projecting it onto an integral interface.
• Stage-level resource profiles must remain authoritative all the way into executor-side child-process settings and resource-manager pod construction.
• Any executor-wide budget must be divided by maximum concurrent task slots, not by physical cores once sub-core tasks are supported.

Implementation sketch

The configuration reader constructs the default resource profile, while stage APIs construct task-specific profiles. TaskSchedulerImpl resolves the profile CPU amount, reserves it from an offer, and sends it in TaskDescription; the executor returns the same amount in terminal status updates.

At execution time, TaskContext.cpuAmount() exposes that resolved value to JVM and Python paths. Separately, Kubernetes maps a requested resource-profile ID to a ResourceProfile and passes it into BasicExecutorFeatureStep when constructing each executor pod. Those existing handoff points provide the exact values needed to address the inline findings without re-reading global configuration.

Behavioral changes worth calling out

With spark.task.cpus < 1, task concurrency can exceed executor cores. That is the intended scheduler behavior, but it also increases the number of simultaneous Python workers and native-library thread pools, so memory caps and thread counts must be derived from task slots and the active profile.

With spark.task.cpus > 1, CPU-weighted runtime metrics remain useful, but deserialization must be removed from the measured execution interval before multiplying by the CPU reservation. Kubernetes recovery executors also need the task CPU amount from the profile for which the pod is being allocated; otherwise they can be unusable for that stage or defeat single-task recovery isolation.

Suggested improvements

• Split configured PySpark executor memory across the maximum concurrent task slots for the active profile, and add a 4-core / 0.5-CPU regression test.
• Set Python worker thread defaults from context.cpuAmount() and test a stage profile that differs from global spark.task.cpus.
• Derive recovery-mode executor cores, and its isolation warning, from the resource profile being allocated.
• Compute CPU-weighted executor runtime from (task interval - in-run deserialization) * task cpus, with a fractional case that currently reports zero.

[sunchao]

[P2] Split PySpark memory across fractional task slots

The fractional concurrency enabled here invalidates getWorkerMemoryMb's assumption above that the worker pool has at most execCores workers. With 4 executor cores, spark.task.cpus=0.5, and 4 GiB of spark.executor.pyspark.memory, Spark can now run 8 Python tasks concurrently, but each worker still receives 4096 / 4 = 1024 MiB. Their aggregate limits can therefore reach 8 GiB even though YARN/Kubernetes adds only 4 GiB to the executor container request, defeating the configured cap and potentially OOMing the executor. Please divide the executor-wide allocation by the maximum concurrent task slots for the active resource profile (or another conservative concurrency bound), and cover this case.

[sunchao]

[P2] Use the task's stage-level CPU amount for OMP_NUM_THREADS

This reads the executor's global spark.task.cpus, but the scheduler may have reserved a different amount from the stage resource profile and already exposes it as context.cpuAmount(). For example, with global CPUs 3, a stage profile requesting 0.5, and a 4-core executor, Spark schedules 8 Python workers while each gets OMP_NUM_THREADS=3, allowing 24 native threads on 4 cores; the inverse mismatch underutilizes larger stage requests. Python worker factories are keyed by the environment map, so using the ceiling of context.cpuAmount() will also keep worker reuse separated correctly by profile.

[sunchao]

[P2] Subtract deserialization before scaling executorRunTime

This currently computes cpus * (deserialization + execution) - deserialization. With cpus=0.2, 100 ms of in-run deserialization, and 100 ms of execution, it produces -60 ms and the clamp reports zero, although the CPU-weighted execution interval is 0.2 * (200 - 100) = 20 ms. Besides under-reporting stage metrics, a zero runtime makes efficient speculation discard the task's process-rate sample. Please remove the in-run deserialization interval before multiplying by the task CPU amount.

[sunchao]

[P2] Derive recovery executor capacity from the profile being launched

ExecutorPodsAllocator resolves the requested profile ID and passes that ResourceProfile into this feature step, but this branch ignores it and rounds the global spark.task.cpus. With global CPUs 1 and a stage profile requesting 1.5, a recovery executor for that profile announces one core and can never launch its 1.5-core tasks. With global CPUs 2 and a stage profile requesting 0.5, it announces two cores and accepts four tasks, defeating recovery isolation. Please derive this value (and the corresponding warning) from the resource profile being allocated.

[sunchao]

One additional issue from the API/configuration pass: stage-level CPU requests can be silently rounded down at the accounting boundary.

[sunchao]

[P2] Reject CPU requests that would be rounded down

This predicate validates only the sign after normalization, so the new public cpus(Double) API accepts values with more precision than the 1e-9 accounting scale and can allocate less CPU than requested. For example, cpus(1.0000000004) is accepted, but getTaskCpus normalizes it to 1.000000000; a 10-core executor then gets 10 slots even though floor(10 / 1.0000000004) = 9. The config path already rejects excess precision, so please validate the unnormalized decimal against the same scale and bounds (or round upward, never downward).

[HyukjinKwon]

I did not take a look at the details but I am supportive of this change

[HyukjinKwon]

cc @tgravescs and @mridulm if you guys find some time to take a look.

cc @jiangxb1987 and @Ngone51