pan3793 opened a new pull request, #57332:
URL: https://github.com/apache/spark/pull/57332
### 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:
```bash
# 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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
