You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ch...@apache.org on 2026/07/26 10:59:52 UTC
(spark) branch branch-4.x updated: [SPARK-58192][CORE] Support fractional spark.task.cpus
This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 4c15d2558902 [SPARK-58192][CORE] Support fractional spark.task.cpus
4c15d2558902 is described below
commit 4c15d2558902b114ab7c66a7f5aeb517e81b1f5e
Author: Cheng Pan <pa...@gmail.com>
AuthorDate: Sun Jul 26 18:59:20 2026 +0800
[SPARK-58192][CORE] Support fractional spark.task.cpus
### 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 is `floor(executor cores / task cpus)` computed exactly.
**Exact decimal accounting.**
- CPU bookkeeping moves from `Int` to `scala.math.BigDecimal`, normalized to a fixed scale of
9 by a new internal `CpuAmount` object (compact, long-backed, drift-free).
- The exact value flows through `WorkerOffer`, `ExecutorData`, `TaskDescription`,
`StatusUpdate`, and `TaskContext`.
- Slot math uses exact `FLOOR` division (`1.0 / 0.1` is 10, not 9), clamps to
`[0, Int.MaxValue]`, and slot sums saturate in `Long`.
**Configuration and API.**
- A new `ConfigBuilder.decimalConf` parses the exact decimal string; `spark.task.cpus` must be
in `[1e-9, Int.MaxValue]` with at most 9 decimal places, validated before normalization so
extreme exponents fail fast.
- `TaskContext.cpuAmount()` (new, Scala and Python) returns the exact amount;
`TaskContext.cpus()` is deprecated and returns its ceiling.
- `TaskResourceRequests.cpus` accepts `Double`/`float`; amounts are rounded to the nearest
`1e-9` and must be in the same range as the config. The addressable-resource assertions
(`<= 1.0 || whole`, `<= 0.5 || whole`) no longer apply to cpus, which is a plain quantity
from the core pool.
- Validation happens at the request entry points and again at profile registration:
`ResourceProfileManager.addResourceProfile` checks the cpus amount under the resource map
key and forces the limiting-resource computation *before* inserting, so a malformed profile
from any live path (raw construction, mutated builder map, Spark Connect) never enters the
registry.
- The `TaskResourceRequest` constructor stays lenient because it deserializes persisted data,
which can carry anything an earlier release accepted (including `0` and `-Infinity`). The
history server additionally parses replayed cpus values defensively: string length and
range checked before parse/normalize, invalid values fall back to the default instead of
failing replay.
**PySpark executor memory** is split by the executor's concurrent task capacity (passed as a
task local property) rather than the raw core count, since fractional cpus can admit more
workers than cores.
- A profile that does not override `pysparkMemory` inherits the default profile's allocation,
matching how its executors are sized.
- With dynamic allocation off -- where default-profile and task-only-profile tasks can share
an executor -- both propagate `ceil(cores / taskCpus)`, so any core-feasible mix of workers
stays within the budget.
- Otherwise the profile's real limiting resource is used, with custom-resource bounds capped
by explicit profile cores when the cores limit is unknown (SPARK-30299).
- A *positive* allocation whose per-worker share rounds to 0 MiB fails fast; `0` still means
"no limit".
**Scheduler.**
- Each stage snapshots the job `Properties` before profile-specific values are written, so
sibling stages with different profiles cannot clobber each other.
- Null job properties are materialized as an empty map.
**Executor / runtime.**
- Python workers derive `OMP_NUM_THREADS` and the integer worker `cpus` from the task's own
cpu amount (stage-level profiles honored) unless `spark.executorEnv.OMP_NUM_THREADS` is set.
- `executorRunTime` subtracts in-run deserialization before weighting by the cpus amount, with
the weight floored at 1 so sub-core tasks do not inflate UI/REST scheduler delay.
- Script-transform child processes get `OMP_NUM_THREADS = ceil(spark.task.cpus)`.
- The pipelined Python UDF writer pool is unbounded: task concurrency is its natural cap, and
a processor-count cap can deadlock barrier stages once tasks outnumber cores.
**Kubernetes.**
- Allocation recovery mode announces `ceil(spark.task.cpus)` cores (global setting only).
- With `spark.task.cpus <= 0.5` a recovery executor accepts `floor(1 / spark.task.cpus)`
tasks instead of one; a one-time warning and docs cover this.
**Docs / misc.**
- `core-migration-guide.md` entries for the PySpark memory behavior changes;
`configuration.md` and `running-on-kubernetes.md` updated.
- `cpuAmount` added to the PySpark API reference.
- MiMa excludes in `v43excludes` to match `Since("4.3.0")`.
**Out of scope** (pre-existing, left to follow-ups):
- K8s recovery-mode cores from stage-level profiles.
- Stage-profile-aware script-transform `OMP_NUM_THREADS`.
- Using the actual registered executor cores for the memory split in standalone (SPARK-30299).
- `shouldCheckExecutorCores` ignoring profile-scoped cores.
- Budgeting multiple concurrently active Python workers within one task (e.g. chained
`mapInPandas`).
### Why are the changes needed?
`spark.task.cpus` only accepted whole numbers, so the finest scheduling granularity was one
task per core. Lightweight tasks (I/O-bound, or PySpark tasks whose heavy lifting happens in
a shared native library or external service) cannot use an executor efficiently, and tasks
needing e.g. `1.5` cores had to round up to 2 and waste capacity:
```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` accept fractional values; whole-number
configurations schedule exactly as before. Out-of-range or over-precise values are rejected
with a clear error.
- New API `TaskContext.cpuAmount()` (Scala and Python); `TaskContext.cpus()` is deprecated.
- In K8s allocation recovery mode with `spark.task.cpus <= 0.5`, recovery executors accept
more than one task (warned and documented).
Fixing places that equated "executor cores" with "concurrent tasks" also changes behavior for
some existing whole-number workloads (covered by `core-migration-guide.md` entries):
- Per-worker PySpark memory follows real concurrency: larger shares for `spark.task.cpus > 1`
and (with dynamic allocation on) for custom-resource-limited profiles; cpus-proportional
shares on shareable executors. Aggregate limits stay within the executor-wide allocation.
- Profiles without an explicit `pysparkMemory` now always carry the default limit (previously
order-dependent).
- An unsatisfiable positive allocation now fails fast instead of silently dropping the limit
(`0` still disables it).
- `executorRunTime` no longer scales the deserialization interval for `cpus > 1`.
- `OMP_NUM_THREADS` honors stage-level profiles.
- Stages carry per-stage snapshots of the job `Properties` instead of sharing one mutable
instance.
### How was this patch tested?
New and updated unit tests:
- `SparkContextSuite`: config bounds/precision matrix.
- `ResourceProfileSuite` / `ResourceProfileManagerSuite`: fractional profiles, slot math,
entry-point and registration validation with registry-unchanged assertions.
- `TaskSchedulerImplSuite`: fractional scheduling end-to-end, slot saturation.
- `TaskDescriptionSuite`: exact fractional encode/decode round-trip.
- `DAGSchedulerSuite`: per-stage property isolation, null job properties, memory-split
propagation across profile shapes (GPU-limited, task-only, ceiling division, capped custom
bounds).
- `ExecutorSuite`: cpu-weighted runtime flooring.
- `AppStatusListenerSuite` / `JsonProtocolSuite` / `KVStoreProtobufSerializerSuite`:
fractional parsing, hostile replayed values, legacy-profile fallbacks.
- `BasePythonRunnerSuite`: worker memory split, fail-fast boundaries, writer-pool concurrency.
- `BasicExecutorFeatureStepSuite` / `ExecutorPodsAllocatorSuite`: K8s recovery-mode ceiling
and one-time warning.
- PySpark `test_taskcontext.py` / `test_worker.py` / `test_resources.py`: `cpuAmount`, worker
env, request round-trips.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Fable 5)
Closes #57332 from pan3793/SPARK-58192.
Authored-by: Cheng Pan <pa...@gmail.com>
Signed-off-by: Cheng Pan <ch...@apache.org>
(cherry picked from commit f8c6ded7b58d681b91a32633b07e5f7ffcc2fb50)
Signed-off-by: Cheng Pan <ch...@apache.org>
---
.../spark/internal/config/ConfigBuilder.scala | 26 +++
.../org/apache/spark/BarrierTaskContext.scala | 2 +-
.../main/scala/org/apache/spark/SparkContext.scala | 5 +-
.../main/scala/org/apache/spark/TaskContext.scala | 15 +-
.../scala/org/apache/spark/TaskContextImpl.scala | 2 +-
.../org/apache/spark/api/python/PythonRunner.scala | 67 +++++--
.../spark/api/python/PythonWorkerUtils.scala | 8 +-
.../scala/org/apache/spark/executor/Executor.scala | 33 +++-
.../org/apache/spark/internal/config/package.scala | 23 ++-
.../org/apache/spark/resource/CpuAmount.scala | 96 ++++++++++
.../apache/spark/resource/ResourceProfile.scala | 81 +++++++--
.../spark/resource/ResourceProfileManager.scala | 18 +-
.../org/apache/spark/resource/ResourceUtils.scala | 22 ++-
.../spark/resource/TaskResourceRequest.scala | 22 ++-
.../spark/resource/TaskResourceRequests.scala | 50 +++++-
.../org/apache/spark/scheduler/DAGScheduler.scala | 71 +++++++-
.../scala/org/apache/spark/scheduler/Task.scala | 2 +-
.../apache/spark/scheduler/TaskDescription.scala | 12 +-
.../apache/spark/scheduler/TaskSchedulerImpl.scala | 53 ++++--
.../apache/spark/scheduler/TaskSetManager.scala | 7 +-
.../org/apache/spark/scheduler/WorkerOffer.scala | 2 +-
.../cluster/CoarseGrainedClusterMessage.scala | 4 +-
.../cluster/CoarseGrainedSchedulerBackend.scala | 8 +-
.../spark/scheduler/cluster/ExecutorData.scala | 7 +-
.../scheduler/local/LocalSchedulerBackend.scala | 6 +-
.../apache/spark/status/AppStatusListener.scala | 34 +++-
.../spark/BarrierStageOnSubmittedSuite.scala | 4 +-
.../scala/org/apache/spark/SparkContextSuite.scala | 30 +++-
.../spark/api/python/BasePythonRunnerSuite.scala | 92 ++++++++++
.../org/apache/spark/executor/ExecutorSuite.scala | 16 +-
.../resource/ResourceProfileManagerSuite.scala | 63 +++++++
.../spark/resource/ResourceProfileSuite.scala | 79 +++++++++
.../CoarseGrainedSchedulerBackendSuite.scala | 10 +-
.../apache/spark/scheduler/DAGSchedulerSuite.scala | 193 +++++++++++++++++++++
.../scheduler/SchedulerIntegrationSuite.scala | 4 +-
.../spark/scheduler/TaskDescriptionSuite.scala | 26 ++-
.../spark/scheduler/TaskSchedulerImplSuite.scala | 60 ++++++-
.../streaming/StreamingShuffleReaderSuite.scala | 2 +-
.../shuffle/streaming/StreamingShuffleSuite.scala | 2 +-
.../streaming/StreamingShuffleWriterSuite.scala | 2 +-
.../spark/status/AppStatusListenerSuite.scala | 68 +++++++-
.../protobuf/KVStoreProtobufSerializerSuite.scala | 6 +-
.../org/apache/spark/util/JsonProtocolSuite.scala | 29 ++++
docs/configuration.md | 13 +-
docs/core-migration-guide.md | 4 +
docs/running-on-kubernetes.md | 5 +
project/MimaExcludes.scala | 10 ++
python/docs/source/reference/pyspark.rst | 2 +
python/pyspark/resource/requests.py | 32 ++--
python/pyspark/resource/tests/test_resources.py | 6 +
python/pyspark/taskcontext.py | 25 ++-
python/pyspark/tests/test_taskcontext.py | 41 +++++
python/pyspark/tests/test_worker.py | 28 +++
python/pyspark/worker_message.py | 4 +
.../scala/org/apache/spark/deploy/k8s/Config.scala | 5 +-
.../k8s/features/BasicExecutorFeatureStep.scala | 6 +-
.../cluster/k8s/ExecutorPodsAllocator.scala | 27 ++-
.../features/BasicExecutorFeatureStepSuite.scala | 21 +++
.../cluster/k8s/ExecutorPodsAllocatorSuite.scala | 27 +++
.../SubexpressionEliminationSuite.scala | 3 +-
.../execution/BaseScriptTransformationExec.scala | 4 +-
.../datasources/SupportsArchiveFormatSuite.scala | 2 +-
.../window/WindowSegmentTreeMemorySuite.scala | 2 +-
63 files changed, 1466 insertions(+), 163 deletions(-)
diff --git a/common/utils/src/main/scala/org/apache/spark/internal/config/ConfigBuilder.scala b/common/utils/src/main/scala/org/apache/spark/internal/config/ConfigBuilder.scala
index a914a9638459..2db7b602201b 100644
--- a/common/utils/src/main/scala/org/apache/spark/internal/config/ConfigBuilder.scala
+++ b/common/utils/src/main/scala/org/apache/spark/internal/config/ConfigBuilder.scala
@@ -345,6 +345,32 @@ private[spark] case class ConfigBuilder(key: String) {
new TypedConfigBuilder(this, toNumber(_, _.toDouble, key, "double"))
}
+ /**
+ * Configure a config entry whose value is a (possibly fractional) decimal number, parsed
+ * directly from its string form into a [[scala.math.BigDecimal]]. Constructing the value from
+ * the string (rather than going through a [[Double]]) preserves the exact decimal value the user
+ * typed (e.g. `0.1` is the exact decimal `0.1`, not the double `0.1000000000000000055...`),
+ * which is required for exact accounting of fractional resources such as a fractional
+ * `spark.task.cpus`.
+ */
+ def decimalConf: TypedConfigBuilder[BigDecimal] = {
+ checkPrependConfig
+ new TypedConfigBuilder(this,
+ toNumber(_, s => BigDecimal(s), key, "decimal"),
+ // strip trailing zeros (so a normalized value like 0.200000000 is written back as 0.2) and
+ // use toPlainString to avoid scientific notation. For extreme exponents (e.g. 1e100000000,
+ // as can be rendered in a checkValue rejection message) toPlainString would materialize
+ // one character per zero, so fall back to scientific notation there.
+ { v: BigDecimal =>
+ val stripped = v.bigDecimal.stripTrailingZeros()
+ if (stripped.scale > 100 || stripped.scale < -100) {
+ stripped.toString
+ } else {
+ stripped.toPlainString
+ }
+ })
+ }
+
def booleanConf: TypedConfigBuilder[Boolean] = {
checkPrependConfig
new TypedConfigBuilder(this, toBoolean(_, key))
diff --git a/core/src/main/scala/org/apache/spark/BarrierTaskContext.scala b/core/src/main/scala/org/apache/spark/BarrierTaskContext.scala
index 49a5e77764a6..cf3ce95c4d1f 100644
--- a/core/src/main/scala/org/apache/spark/BarrierTaskContext.scala
+++ b/core/src/main/scala/org/apache/spark/BarrierTaskContext.scala
@@ -238,7 +238,7 @@ class BarrierTaskContext private[spark] (
taskContext.getMetricsSources(sourceName)
}
- override def cpus(): Int = taskContext.cpus()
+ override def cpuAmount(): BigDecimal = taskContext.cpuAmount()
override def resources(): Map[String, ResourceInformation] = taskContext.resources()
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 3610c676c40f..8cb5eef770a0 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -578,7 +578,10 @@ class SparkContext(config: SparkConf) extends Logging {
// SPARK-28843: limit the OpenMP thread pool to the number of cores assigned to this executor
// this avoids high memory consumption with pandas/numpy because of a large OpenMP thread pool
// see https://github.com/numpy/numpy/issues/10455
- executorEnvs.put("OMP_NUM_THREADS", _conf.get("spark.task.cpus", "1"))
+ // `spark.task.cpus` may be fractional, so round up to an integer number of threads. It is
+ // validated to be > 0, so the ceiling is always >= 1.
+ executorEnvs.put("OMP_NUM_THREADS",
+ _conf.get(CPUS_PER_TASK).setScale(0, BigDecimal.RoundingMode.CEILING).intValue.toString)
}
// We need to register "HeartbeatReceiver" before "createTaskScheduler" because Executor will
diff --git a/core/src/main/scala/org/apache/spark/TaskContext.scala b/core/src/main/scala/org/apache/spark/TaskContext.scala
index a4f68d76128d..a853ea6bf029 100644
--- a/core/src/main/scala/org/apache/spark/TaskContext.scala
+++ b/core/src/main/scala/org/apache/spark/TaskContext.scala
@@ -284,10 +284,21 @@ abstract class TaskContext extends Serializable {
def getLocalProperty(key: String): String
/**
- * CPUs allocated to the task.
+ * CPUs allocated to the task, rounded up to a whole number.
+ *
+ * @note when `spark.task.cpus` is fractional (e.g. `0.2`) this rounds up to the next integer and
+ * so cannot represent the fractional amount; use [[cpuAmount]] for the exact value.
*/
+ @deprecated("use cpuAmount() instead, which supports a fractional spark.task.cpus", "4.3.0")
@Since("3.3.0")
- def cpus(): Int
+ def cpus(): Int = cpuAmount().setScale(0, BigDecimal.RoundingMode.CEILING).intValue
+
+ /**
+ * CPUs allocated to the task as an exact, possibly fractional amount, carried as a
+ * [[scala.math.BigDecimal]] (e.g. `BigDecimal("0.2")` when `spark.task.cpus` is `0.2`).
+ */
+ @Since("4.3.0")
+ def cpuAmount(): BigDecimal
/**
* Resources allocated to the task. The key is the resource name and the value is information
diff --git a/core/src/main/scala/org/apache/spark/TaskContextImpl.scala b/core/src/main/scala/org/apache/spark/TaskContextImpl.scala
index 9bf9ce0d9de2..ec4e08abf9a5 100644
--- a/core/src/main/scala/org/apache/spark/TaskContextImpl.scala
+++ b/core/src/main/scala/org/apache/spark/TaskContextImpl.scala
@@ -57,7 +57,7 @@ private[spark] class TaskContextImpl(
@transient private val metricsSystem: MetricsSystem,
// The default value is only used in tests.
override val taskMetrics: TaskMetrics = TaskMetrics.empty,
- override val cpus: Int = SparkEnv.get.conf.get(config.CPUS_PER_TASK),
+ override val cpuAmount: BigDecimal = SparkEnv.get.conf.get(config.CPUS_PER_TASK),
override val resources: Map[String, ResourceInformation] = Map.empty)
extends TaskContext
with Logging {
diff --git a/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala b/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala
index 697545044e0e..3c300be6a7fc 100644
--- a/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala
+++ b/core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala
@@ -38,7 +38,8 @@ import org.apache.spark.internal.LogKeys.{COUNT, PYTHON_WORKER_IDLE_TIMEOUT, SIZ
import org.apache.spark.internal.config.{BUFFER_SIZE, EXECUTOR_CORES}
import org.apache.spark.internal.config.Python._
import org.apache.spark.rdd.InputFileBlockHolder
-import org.apache.spark.resource.ResourceProfile.{EXECUTOR_CORES_LOCAL_PROPERTY, PYSPARK_MEMORY_LOCAL_PROPERTY}
+import org.apache.spark.resource.ResourceProfile
+import org.apache.spark.resource.ResourceProfile.{EXECUTOR_CORES_LOCAL_PROPERTY, MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY, PYSPARK_MEMORY_LOCAL_PROPERTY}
import org.apache.spark.security.SocketAuthHelper
import org.apache.spark.util._
@@ -125,15 +126,15 @@ private[spark] object BasePythonRunner extends Logging {
* Shared thread pool for pipelined writer tasks. Using a cached thread pool ensures that
* writer threads are reused across tasks, which keeps JIT-compiled code, branch prediction
* history, and CPU caches warm.
- * Bounded by executor cores since each task uses at most one writer thread.
*/
private[python] lazy val pipelinedWriterThreadPool = {
- // Each concurrent task uses at most one writer thread. Bound the pool by available
- // processors, which is the natural upper limit for concurrent tasks on this executor.
- // Using availableProcessors() instead of EXECUTOR_CORES because the latter defaults
- // to 1 in local[*] mode even though multiple tasks run concurrently.
- val maxThreads = Runtime.getRuntime.availableProcessors()
- ThreadUtils.newDaemonCachedThreadPool("python-udf-pipelined-writer", maxThreads)
+ // Each running task uses at most one writer thread, so the pool size is naturally capped
+ // by the scheduler's task concurrency -- which a fractional spark.task.cpus can push above
+ // the host's processor count. Leave the pool unbounded: a cap below the concurrent task
+ // count would queue writers behind running ones, which can deadlock a barrier stage (the
+ // tasks whose workers already started wait at the barrier while the remaining tasks'
+ // writers never run). Idle threads are still reused and reaped after the keep-alive.
+ ThreadUtils.newDaemonCachedThreadPool("python-udf-pipelined-writer")
}
private[spark] lazy val faultHandlerLogDir = Utils.createTempDir(namePrefix = "faulthandler")
@@ -154,6 +155,34 @@ private[spark] object BasePythonRunner extends Logging {
} else None
}
+ /**
+ * Splits the executor-wide pyspark memory allocation evenly across the executor's task slots.
+ * The Python worker pool can grow to the number of tasks that run concurrently on the executor
+ * (`maxConcurrentTasks`, the limiting resource across cores and custom resources), which a
+ * fractional `spark.task.cpus` can push above the core count; dividing by the core count alone
+ * would let the workers' aggregate limits exceed the executor-wide allocation.
+ */
+ private[spark] def getWorkerMemoryMb(
+ mem: Option[Long],
+ maxConcurrentTasks: Int): Option[Long] = {
+ val taskSlots = math.max(1, maxConcurrentTasks)
+ mem.map { m =>
+ val perWorkerMb = m / taskSlots
+ // An explicit spark.executor.pyspark.memory=0 means the limit is disabled
+ // (setup_memory_limits treats a non-positive value as "no limit"). Only a *positive*
+ // budget that rounds down to 0 MiB per worker is a problem: it would silently drop the
+ // configured cap and let the workers' aggregate memory blow past it, so fail fast on
+ // that unsatisfiable case rather than removing the limit.
+ if (m > 0 && perWorkerMb <= 0) {
+ throw new SparkException(
+ s"Cannot honor spark.executor.pyspark.memory=${m}m split across $taskSlots concurrent " +
+ "task slots: each worker would get less than 1 MiB. Increase " +
+ "spark.executor.pyspark.memory, or reduce the executor's concurrent task capacity.")
+ }
+ perWorkerMb
+ }
+ }
+
/**
* Creates a task identifier string for logging following Spark's standard format.
* Format: "task <partition>.<attempt> in stage <stageId> (TID <taskAttemptId>)"
@@ -288,12 +317,6 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
// Authentication helper used when serving method calls via socket from Python side.
private lazy val authHelper = new SocketAuthHelper(conf)
- // each python worker gets an equal part of the allocation. the worker pool will grow to the
- // number of concurrent tasks, which is determined by the number of cores in this executor.
- private def getWorkerMemoryMb(mem: Option[Long], cores: Int): Option[Long] = {
- mem.map(_ / cores)
- }
-
def compute(
inputIterator: Iterator[IN],
partitionIndex: Int,
@@ -310,9 +333,12 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
val memoryMb = Option(context.getLocalProperty(PYSPARK_MEMORY_LOCAL_PROPERTY)).map(_.toLong)
val localdir = env.blockManager.diskBlockManager.localDirs.map(f => f.getPath()).mkString(",")
// If OMP_NUM_THREADS is not explicitly set, override it with the number of task cpus.
- // See SPARK-42613 for details.
+ // See SPARK-42613 for details. Use the task's own cpu amount rather than the global
+ // `spark.task.cpus` so stage-level resource profiles are honored; the amount may be
+ // fractional, and `cpus()` returns its ceiling (always >= 1 since the amount is
+ // validated to be positive).
if (conf.getOption("spark.executorEnv.OMP_NUM_THREADS").isEmpty) {
- envVars.put("OMP_NUM_THREADS", conf.get("spark.task.cpus", "1"))
+ envVars.put("OMP_NUM_THREADS", context.cpus().toString)
}
envVars.put("SPARK_LOCAL_DIRS", localdir) // it's also used in monitor thread
if (reuseWorker) {
@@ -330,7 +356,14 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
// SPARK-30299 this could be wrong with standalone mode when executor
// cores might not be correct because it defaults to all cores on the box.
val execCores = execCoresProp.map(_.toInt).getOrElse(conf.get(EXECUTOR_CORES))
- val workerMemoryMb = getWorkerMemoryMb(memoryMb, execCores)
+ // The Python worker pool grows to the number of tasks that run concurrently on this executor.
+ // Prefer the resource profile's max concurrent tasks (the limiting resource across cores and
+ // custom resources, e.g. GPUs); fall back to the cpu-based slot count when it isn't known
+ // (standalone/local without an explicit spark.executor.cores).
+ val maxConcurrentTasks = Option(context.getLocalProperty(MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY))
+ .map(_.toInt)
+ .getOrElse(ResourceProfile.numTasksBasedOnCores(BigDecimal(execCores), context.cpuAmount()))
+ val workerMemoryMb = BasePythonRunner.getWorkerMemoryMb(memoryMb, maxConcurrentTasks)
if (workerMemoryMb.isDefined) {
envVars.put("PYSPARK_EXECUTOR_MEMORY_MB", workerMemoryMb.get.toString)
}
diff --git a/core/src/main/scala/org/apache/spark/api/python/PythonWorkerUtils.scala b/core/src/main/scala/org/apache/spark/api/python/PythonWorkerUtils.scala
index 14abbcb2b817..8d53cfe1d188 100644
--- a/core/src/main/scala/org/apache/spark/api/python/PythonWorkerUtils.scala
+++ b/core/src/main/scala/org/apache/spark/api/python/PythonWorkerUtils.scala
@@ -30,6 +30,7 @@ import org.apache.spark.{SparkEnv, SparkFiles}
import org.apache.spark.api.python.PythonFunction.PythonAccumulator
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.internal.Logging
+import org.apache.spark.resource.CpuAmount
private[spark] object PythonWorkerUtils extends Logging {
@@ -83,7 +84,12 @@ private[spark] object PythonWorkerUtils extends Logging {
"partitionId" -> context.partitionId(),
"attemptNumber" -> context.attemptNumber(),
"taskAttemptId" -> context.taskAttemptId(),
- "cpus" -> context.cpus(),
+ // The Python worker exposes cpus as an int; round a fractional cpus up to a whole number
+ // (matching the deprecated TaskContext.cpus()).
+ "cpus" -> context.cpuAmount().setScale(0, BigDecimal.RoundingMode.CEILING).intValue,
+ // The exact (possibly fractional) cpus amount as a plain decimal string, read by the
+ // Python TaskContext.cpuAmount().
+ "cpuAmount" -> CpuAmount.toDisplayString(context.cpuAmount()),
"resources" -> context.resources().map { case (k, v) =>
k -> Map(
"name" -> v.name,
diff --git a/core/src/main/scala/org/apache/spark/executor/Executor.scala b/core/src/main/scala/org/apache/spark/executor/Executor.scala
index 0de5c376c710..f288ceacef7e 100644
--- a/core/src/main/scala/org/apache/spark/executor/Executor.scala
+++ b/core/src/main/scala/org/apache/spark/executor/Executor.scala
@@ -48,7 +48,7 @@ import org.apache.spark.internal.config.{EXECUTOR_USER_CLASS_PATH_FIRST => EXECU
import org.apache.spark.internal.plugin.PluginContainer
import org.apache.spark.memory.{SparkOutOfMemoryError, TaskMemoryManager}
import org.apache.spark.metrics.source.JVMCPUSource
-import org.apache.spark.resource.ResourceInformation
+import org.apache.spark.resource.{CpuAmount, ResourceInformation}
import org.apache.spark.rpc.RpcTimeout
import org.apache.spark.scheduler._
import org.apache.spark.serializer.SerializerHelper
@@ -767,8 +767,9 @@ private[spark] class Executor(
t.metrics.setExecutorRunTime(TimeUnit.NANOSECONDS.toMillis(
// SPARK-32898: it's possible that a task is killed when taskStartTimeNs has the initial
// value(=0) still. In this case, the executorRunTime should be considered as 0.
- if (taskStartTimeNs > 0) (System.nanoTime() - taskStartTimeNs) * taskDescription.cpus
- else 0))
+ if (taskStartTimeNs > 0) {
+ Executor.cpuWeightedNanos(System.nanoTime() - taskStartTimeNs, taskDescription.cpus)
+ } else 0))
t.metrics.setJvmGCTime(computeTotalGcTime() - startGCTime)
})
@@ -958,10 +959,14 @@ private[spark] class Executor(
(taskStartTimeNs - deserializeStartTimeNs) + task.executorDeserializeTimeNs))
task.metrics.setExecutorDeserializeCpuTime(
(taskStartCpu - deserializeStartCpuTime) + task.executorDeserializeCpuTime)
- // We need to subtract Task.run()'s deserialization time to avoid double-counting
- task.metrics.setExecutorRunTime(TimeUnit.NANOSECONDS.toMillis(
- (taskFinishNs - taskStartTimeNs) * taskDescription.cpus
- - task.executorDeserializeTimeNs))
+ // We need to subtract Task.run()'s deserialization time to avoid double-counting:
+ // remove the in-run deserialization interval from the elapsed time first, then weight
+ // the remaining run interval by the task's cpu amount. Clamp at zero to guard against
+ // clock anomalies.
+ task.metrics.setExecutorRunTime(math.max(0L, TimeUnit.NANOSECONDS.toMillis(
+ Executor.cpuWeightedNanos(
+ (taskFinishNs - taskStartTimeNs) - task.executorDeserializeTimeNs,
+ taskDescription.cpus))))
task.metrics.setExecutorCpuTime(
(taskFinishCpu - taskStartCpu) - task.executorDeserializeCpuTime)
task.metrics.setJvmGCTime(computeTotalGcTime() - startGCTime)
@@ -1590,6 +1595,20 @@ private[spark] object Executor extends Logging {
val TASK_THREAD_NAME_PREFIX = "Executor task launch worker"
val IDLE_TASK_THREAD_NAME = "Executor task idle worker"
+ /**
+ * Weights a measured task interval by the task's cpu reservation for `executorRunTime`
+ * (SPARK-51666), flooring the weight at 1: a sub-core weight would report a run time shorter
+ * than the elapsed interval, and UI/REST scheduler delay -- computed as wall-clock duration
+ * minus executorRunTime -- would misreport the un-consumed share of the core as scheduler
+ * delay. Relative comparisons for speculation are unaffected because all tasks of a stage
+ * share the same cpu amount. Trailing zeros are stripped from the scale-9 cpus so the
+ * product stays in BigDecimal's compact (long-backed) form instead of inflating a long
+ * duration into a BigInteger.
+ */
+ private[spark] def cpuWeightedNanos(intervalNs: Long, cpus: BigDecimal): Long = {
+ (BigDecimal(intervalNs) * CpuAmount.stripTrailingZeros(cpus.max(BigDecimal(1)))).toLong
+ }
+
// This is reserved for internal use by components that need to read task properties before a
// task is fully deserialized. When possible, the TaskContext.getLocalProperty call should be
// used instead.
diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala
index c8bd6410be27..9ddb0a025ab7 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/package.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala
@@ -28,6 +28,7 @@ import org.apache.spark.metrics.GarbageCollectionMetrics
import org.apache.spark.network.shuffle.Constants
import org.apache.spark.network.shuffledb.DBBackend
import org.apache.spark.network.util.ByteUnit
+import org.apache.spark.resource.CpuAmount
import org.apache.spark.scheduler.{EventLoggingListener, SchedulingMode}
import org.apache.spark.shuffle.sort.io.LocalDiskShuffleDataIO
import org.apache.spark.storage.{DefaultTopologyMapper, RandomBlockReplicationPolicy}
@@ -736,9 +737,19 @@ package object config {
private[spark] val CPUS_PER_TASK =
ConfigBuilder("spark.task.cpus")
.version("0.5.0")
- .intConf
- .checkValue(_ > 0, "Number of cores to allocate for each task should be positive.")
- .createWithDefault(1)
+ .decimalConf
+ // Validate before normalizing: bounding first keeps the setScale in the normalize step
+ // from materializing enormous unscaled values for extreme exponents (e.g. 1e100000000),
+ // and rejecting excess precision is clearer than silently rounding it away.
+ .checkValue(v => v >= CpuAmount.MIN_AMOUNT && v <= CpuAmount.MAX_AMOUNT,
+ "Number of cores to allocate for each task must be a positive value between " +
+ s"${CpuAmount.MIN_AMOUNT} and ${CpuAmount.MAX_AMOUNT} (inclusive).")
+ .checkValue(CpuAmount.stripTrailingZeros(_).scale <= CpuAmount.SCALE,
+ s"Number of cores to allocate for each task supports at most ${CpuAmount.SCALE} " +
+ "decimal places.")
+ // normalize to the fixed CPU accounting scale so every read is a uniform, compact BigDecimal
+ .transform(CpuAmount.normalize)
+ .createWithDefault(BigDecimal(1))
private[spark] val DYN_ALLOCATION_ENABLED =
ConfigBuilder("spark.dynamicAllocation.enabled")
@@ -2566,8 +2577,10 @@ package object config {
"longer time than the threshold. This config helps speculate stage with very few " +
"tasks. Regular speculation configs may also apply if the executor slots are " +
"large enough. E.g. tasks might be re-launched if there are enough successful runs " +
- "even though the threshold hasn't been reached. The number of slots is computed based " +
- "on the conf values of spark.executor.cores and spark.task.cpus minimum 1.")
+ "even though the threshold hasn't been reached. The number of slots is the maximum " +
+ "number of concurrent tasks per executor for the stage's resource profile, computed " +
+ "from the executor cores and the task cpus amount (which may be fractional), or 1 " +
+ "when the executor cores are not known.")
.version("3.0.0")
.timeConf(TimeUnit.MILLISECONDS)
.createOptional
diff --git a/core/src/main/scala/org/apache/spark/resource/CpuAmount.scala b/core/src/main/scala/org/apache/spark/resource/CpuAmount.scala
new file mode 100644
index 000000000000..e2dea8775257
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/resource/CpuAmount.scala
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.resource
+
+/**
+ * The fixed, normalized [[scala.math.BigDecimal]] representation used for all CPU accounting
+ * (cores / cpus per task / free cores / task slots). `spark.task.cpus` can be fractional (e.g.
+ * `0.2`), and a naive `Double` accounting would drift (`1.0 / 0.1 == 9.999...` floors to 9 instead
+ * of 10); carrying the values as an exact `BigDecimal` avoids that.
+ *
+ * Every CPU value is normalized to a fixed scale of [[SCALE]] (9 fractional digits, i.e. `1e-9`
+ * resolution -- far finer than any realistic `spark.task.cpus`). With the integral part bounded by
+ * the 32-bit core range, the value is a "precision 19, scale 9" decimal whose unscaled magnitude
+ * stays below `Long.MAX_VALUE`. That keeps `BigDecimal` on its compact, `long`-backed fast path
+ * (no `BigInteger` allocation) on the scheduler hot path, and -- because `BigDecimal` add/subtract
+ * align to the larger scale -- arithmetic over normalized values stays at scale 9 without any
+ * re-normalization, giving one uniform representation everywhere.
+ */
+private[spark] object CpuAmount {
+
+ /** Number of fractional decimal digits carried by every normalized CPU accounting value. */
+ val SCALE: Int = 9
+
+ /** Smallest positive amount representable at [[SCALE]], i.e. 1e-9. */
+ val MIN_AMOUNT: BigDecimal = BigDecimal(java.math.BigDecimal.valueOf(1, SCALE))
+
+ /**
+ * Largest supported amount. Executor capacity is an `Int` number of cores everywhere (config,
+ * registration RPC, scheduler bookkeeping), so a per-task amount beyond `Int.MaxValue` could
+ * never be satisfied.
+ */
+ val MAX_AMOUNT: BigDecimal = BigDecimal(Int.MaxValue)
+
+ /**
+ * Normalize a value to the fixed [[SCALE]] so it stays in the compact, `long`-backed form.
+ * Constructing values from their exact decimal string (never from a `Double`) and normalizing
+ * here is what guarantees the uniform representation.
+ */
+ def normalize(v: BigDecimal): BigDecimal = v.setScale(SCALE, BigDecimal.RoundingMode.HALF_UP)
+
+ /**
+ * Whether an amount lies in the valid cpus range [[MIN_AMOUNT]] to [[MAX_AMOUNT]]. The
+ * comparison is cheap at any scale, so callers handling untrusted input can (and should)
+ * check the compact parsed value with this before [[normalize]] -- setScale on an extreme
+ * exponent (e.g. 1e10000000) would materialize an enormous unscaled value.
+ */
+ def isInRange(v: BigDecimal): Boolean = v >= MIN_AMOUNT && v <= MAX_AMOUNT
+
+ // Generous bound on the length of a parseable cpus string: sign, 10 integer digits, the
+ // point, 9 decimals, and exponent syntax fit well within it.
+ private val MAX_PARSE_LENGTH = 64
+
+ /**
+ * Safely parses a cpus amount from an untrusted string, e.g. a property read back from a
+ * persisted application environment. The length is bounded before parsing (decimal-string
+ * parsing cost grows superlinearly with the significand, so a crafted million-digit value
+ * -- about 1 KB once compressed in an event log -- could burn seconds before any range
+ * check runs), the range is checked on the compact parsed value before normalizing (see
+ * [[isInRange]]), and any malformed value yields None. Every value the live config parser
+ * accepts fits well within the length bound; an absurdly formatted valid value merely
+ * falls back to the caller's default.
+ */
+ def parseUntrusted(v: String): Option[BigDecimal] = {
+ if (v == null || v.length > MAX_PARSE_LENGTH) {
+ None
+ } else {
+ scala.util.Try(BigDecimal(v)).toOption.filter(isInRange).map(normalize)
+ }
+ }
+
+ /**
+ * The minimal-scale form (trailing zeros removed). Use this for logging/printing (e.g.
+ * `0.200000000` renders as `0.2`) and for products that would otherwise blow the normalized
+ * unscaled value past `Long` range -- most notably `durationNanos * cpus` in the executor CPU
+ * time metric, where a scale-9 `cpus` would inflate the intermediate into a `BigInteger`.
+ */
+ def stripTrailingZeros(v: BigDecimal): BigDecimal = BigDecimal(v.bigDecimal.stripTrailingZeros())
+
+ /** Render a normalized value for logs/messages/UIs with trailing zeros trimmed. */
+ def toDisplayString(v: BigDecimal): String = v.bigDecimal.stripTrailingZeros().toPlainString
+}
diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala
index 971b14265979..e397e80fe573 100644
--- a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala
+++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala
@@ -67,9 +67,11 @@ class ResourceProfile(
* Validate the ResourceProfile
*/
protected def validate(): Unit = {
- // The task.amount in ResourceProfile falls within the range of 0 to 0.5,
- // or it's a whole number
- for ((_, taskReq) <- taskResources) {
+ // The task.amount in ResourceProfile falls within the range of 0 to 0.5, or it's a whole
+ // number, so that it maps onto resource-address slots. CPUs are exempt: they are a plain
+ // quantity drawn from the executor's core pool rather than an addressable resource, so any
+ // positive amount is valid (e.g. 1.5).
+ for ((rName, taskReq) <- taskResources if rName != ResourceProfile.CPUS) {
val taskAmount = taskReq.amount
assert(taskAmount <= 0.5 || taskAmount % 1 == 0,
s"The task resource amount ${taskAmount} must be either <= 0.5, or a whole number.")
@@ -99,8 +101,15 @@ class ResourceProfile(
executorResources.get(ResourceProfile.CORES).map(_.amount.toInt)
}
- private[spark] def getTaskCpus: Option[Int] = {
- taskResources.get(ResourceProfile.CPUS).map(_.amount.toInt)
+ // TaskResourceRequest.amount is a Double (the shared resource-request API). Recover the exact
+ // decimal from it via its string form: BigDecimal(amount.toString) is the exact decimal the
+ // user typed (e.g. 0.1), whereas BigDecimal(amount) would bake in the binary double
+ // (0.1000000000000000055...). Normalize to the fixed CPU accounting scale so the result
+ // matches every other cpus value. Cached because the scheduler reads it on every task-slot
+ // probe; @transient so it is recomputed on deserialization instead of being shipped.
+ @transient private[spark] lazy val getTaskCpus: Option[BigDecimal] = {
+ taskResources.get(ResourceProfile.CPUS)
+ .map(tr => CpuAmount.normalize(BigDecimal(tr.amount.toString)))
}
private[spark] def getPySparkMemory: Option[Long] = {
@@ -195,13 +204,13 @@ class ResourceProfile(
private def calculateTasksAndLimitingResource(sparkConf: SparkConf): Unit = synchronized {
val shouldCheckExecCores = shouldCheckExecutorCores(sparkConf)
var (taskLimit, limitingResource) = if (shouldCheckExecCores) {
- val cpusPerTask = taskResources.get(ResourceProfile.CPUS)
- .map(_.amount).getOrElse(sparkConf.get(CPUS_PER_TASK).toDouble).toInt
+ val cpusPerTask = getTaskCpus.getOrElse(sparkConf.get(CPUS_PER_TASK))
assert(cpusPerTask > 0, "CPUs per task configuration has to be > 0")
val coresPerExecutor = getExecutorCores.getOrElse(sparkConf.get(EXECUTOR_CORES))
_coresLimitKnown = true
ResourceUtils.validateTaskCpusLargeEnough(sparkConf, coresPerExecutor, cpusPerTask)
- val tasksBasedOnCores = coresPerExecutor / cpusPerTask
+ val tasksBasedOnCores = ResourceProfile.numTasksBasedOnCores(
+ CpuAmount.normalize(BigDecimal(coresPerExecutor)), cpusPerTask)
// Note that if the cores per executor aren't set properly this calculation could be off,
// we default it to just be 1 in order to allow checking of the rest of the custom
// resources. We set the limit based on the other resources available.
@@ -301,9 +310,9 @@ private[spark] class TaskResourceProfile(
override val taskResources: Map[String, TaskResourceRequest])
extends ResourceProfile(Map.empty, taskResources) {
- // The task.amount in TaskResourceProfile falls within the range of 0 to 1.0,
- // or it's a whole number, and it has been checked in the TaskResourceRequest.
- // Therefore, we can safely skip this check.
+ // Custom task amounts are checked in the TaskResourceRequest constructor (<= 1.0 or a
+ // whole number) and cpus amounts at the request entry points (TaskResourceRequests / the
+ // spark.task.cpus config parser). Therefore, we can safely skip this check.
override protected def validate(): Unit = {}
override protected[spark] def getCustomExecutorResources()
@@ -401,7 +410,10 @@ object ResourceProfile extends Logging {
private def getDefaultTaskResources(conf: SparkConf): Map[String, TaskResourceRequest] = {
val cpusPerTask = conf.get(CPUS_PER_TASK)
- val treqs = new TaskResourceRequests().cpus(cpusPerTask)
+ // TaskResourceRequest stores the amount as a Double (the custom-resource API contract);
+ // getTaskCpus recovers the exact BigDecimal from it via BigDecimal(amount.toString), so the
+ // BigDecimal -> Double -> BigDecimal round trip stays exact for any realistic value.
+ val treqs = new TaskResourceRequests().cpus(cpusPerTask.toDouble)
ResourceUtils.addTaskResourceRequests(conf, treqs)
treqs.requests
}
@@ -465,10 +477,46 @@ object ResourceProfile extends Logging {
* Get the number of cpus per task if its set in the profile, otherwise return the
* cpus per task for the default profile.
*/
- private[spark] def getTaskCpusOrDefaultForProfile(rp: ResourceProfile, conf: SparkConf): Int = {
+ private[spark] def getTaskCpusOrDefaultForProfile(
+ rp: ResourceProfile, conf: SparkConf): BigDecimal = {
rp.getTaskCpus.getOrElse(conf.get(CPUS_PER_TASK))
}
+ /**
+ * Number of tasks that can run concurrently given the available cores and the number of cpus
+ * required per task. Both arguments are in the internal exact [[scala.math.BigDecimal]]
+ * representation. The division and floor are performed with [[scala.math.BigDecimal]] arithmetic,
+ * which is exact and avoids the floating point rounding errors that would otherwise make values
+ * like `1.0 / 0.1` floor to the wrong slot count.
+ */
+ private[spark] def numTasksBasedOnCores(cores: BigDecimal, cpusPerTask: BigDecimal): Int = {
+ assert(cpusPerTask > 0, "cpusPerTask must be > 0")
+ // floor(cores / cpusPerTask) using exact decimal arithmetic; a scale of 0 with FLOOR rounding
+ // yields the integer quotient (e.g. 1.0 / 0.1 == 10 exactly).
+ val quotient = BigDecimal(
+ cores.bigDecimal.divide(cpusPerTask.bigDecimal, 0, java.math.RoundingMode.FLOOR))
+ // Guard the Int conversion: BigDecimal.intValue() wraps modulo 2^32, so a tiny cpusPerTask
+ // (e.g. 1e-9) could turn a huge quotient into a negative task count. Clamp to
+ // [0, Int.MaxValue] so slot counts and barrier-stage checks always see a sane value.
+ quotient.min(Int.MaxValue).max(0).toInt
+ }
+
+ /**
+ * Ceiling counterpart of [[numTasksBasedOnCores]], used for sizing per-worker memory shares
+ * on executors that profiles can share: dividing a budget by `ceil(cores / cpusPerTask)`
+ * yields a share of at most `budget * cpusPerTask / cores`, so the shares of any set of
+ * co-scheduled tasks whose cpus sum to at most the executor cores never exceed the budget --
+ * even when `cpusPerTask` does not evenly divide the cores, where floor division would
+ * overshoot (e.g. a 3-cpu task on 4 cores would take the whole budget instead of at most
+ * three quarters of it).
+ */
+ private[spark] def numMemoryShareSlotsCeil(cores: BigDecimal, cpusPerTask: BigDecimal): Int = {
+ assert(cpusPerTask > 0, "cpusPerTask must be > 0")
+ val quotient = BigDecimal(
+ cores.bigDecimal.divide(cpusPerTask.bigDecimal, 0, java.math.RoundingMode.CEILING))
+ quotient.min(Int.MaxValue).max(0).toInt
+ }
+
/**
* Get offHeap memory size from [[ExecutorResourceRequest]]
* return 0 if MEMORY_OFFHEAP_ENABLED is false.
@@ -567,4 +615,11 @@ object ResourceProfile extends Logging {
private[spark] val PYSPARK_MEMORY_LOCAL_PROPERTY = "resource.pyspark.memory"
private[spark] val EXECUTOR_CORES_LOCAL_PROPERTY = "resource.executor.cores"
+ // The maximum number of tasks that can run concurrently on an executor of the stage's resource
+ // profile; consumed by PythonRunner to split PySpark worker memory by real concurrency. See
+ // DAGScheduler.addPySparkConfigsToProperties for which bound is propagated: the
+ // cpus-proportional cores split when executors can be shared across profiles (DRA off),
+ // the profile's limiting resource when the cores limit is known, or a cores-capped
+ // custom-resource upper bound otherwise. Unset when no bound is known.
+ private[spark] val MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY = "resource.maxTasksPerExecutor"
}
diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala
index a3d76a92ddd8..fed5129bcf9d 100644
--- a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala
+++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala
@@ -127,6 +127,22 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf,
def addResourceProfile(rp: ResourceProfile): Unit = {
isSupported(rp)
+ // Validate before inserting so a malformed profile never enters the registry, where it
+ // would stay visible to the whole application. The cpus amount is checked under the map
+ // key -- the identity scheduling uses -- with the same rule as the request entry points:
+ // construction stays lenient (deserialization of persisted data must accept anything an
+ // earlier release wrote), so registration is the enforcement point for every live path,
+ // including raw profile construction and Spark Connect.
+ rp.taskResources.get(ResourceProfile.CPUS).foreach { treq =>
+ require(!treq.amount.isNaN && !treq.amount.isInfinity &&
+ CpuAmount.isInRange(CpuAmount.normalize(BigDecimal(treq.amount.toString))),
+ s"The cpus amount ${treq.amount} must be at least 1e-9 and at most ${Int.MaxValue}.")
+ }
+ // Force the computation of maxTasks and limitingResource now so we don't have cost later;
+ // doing it before the insert also surfaces any other malformed shape (e.g. a task
+ // resource without a matching executor resource) before registration instead of after.
+ // The result is cached in the profile, so re-adding an existing profile stays cheap.
+ rp.limitingResource(sparkConf)
var putNewProfile = false
writeLock.lock()
try {
@@ -139,8 +155,6 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf,
}
// do this outside the write lock only when we add a new profile
if (putNewProfile) {
- // force the computation of maxTasks and limitingResource now so we don't have cost later
- rp.limitingResource(sparkConf)
logInfo(log"Added ResourceProfile id: ${MDC(LogKeys.RESOURCE_PROFILE_ID, rp.id)}")
listenerBus.post(SparkListenerResourceProfileAdded(rp))
}
diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceUtils.scala b/core/src/main/scala/org/apache/spark/resource/ResourceUtils.scala
index 8e11923a9e8f..82f26079eeed 100644
--- a/core/src/main/scala/org/apache/spark/resource/ResourceUtils.scala
+++ b/core/src/main/scala/org/apache/spark/resource/ResourceUtils.scala
@@ -407,11 +407,14 @@ private[spark] object ResourceUtils extends Logging {
s"${resourceRequest.id.resourceName}")
}
- def validateTaskCpusLargeEnough(sparkConf: SparkConf, execCores: Int, taskCpus: Int): Boolean = {
- // Number of cores per executor must meet at least one task requirement.
- if (execCores < taskCpus) {
+ def validateTaskCpusLargeEnough(
+ sparkConf: SparkConf, execCores: Int, taskCpus: BigDecimal): Boolean = {
+ // Number of cores per executor must meet at least one task requirement. taskCpus is in the
+ // normalized BigDecimal representation; compare against the executor cores in the same form.
+ val execCpusInternal = CpuAmount.normalize(BigDecimal(execCores))
+ if (execCpusInternal < taskCpus) {
throw new SparkException(s"The number of cores per executor (=$execCores) has to be >= " +
- s"the number of cpus per task = $taskCpus.")
+ s"the number of cpus per task = ${CpuAmount.toDisplayString(taskCpus)}.")
}
true
}
@@ -438,10 +441,12 @@ private[spark] object ResourceUtils extends Logging {
// can't calculate cores limit
return
}
+ // normalized BigDecimal representation of the executor cores (whole number).
+ val coresInternal = CpuAmount.normalize(BigDecimal(cores))
// when executor cores config isn't set, we can't calculate the real limiting resource
// and number of tasks per executor ahead of time, so calculate it now.
if (!coresKnown) {
- val numTasksPerExecCores = cores / taskCpus
+ val numTasksPerExecCores = ResourceProfile.numTasksBasedOnCores(coresInternal, taskCpus)
val numTasksPerExecCustomResource = rp.maxTasksPerExecutor(sparkConf)
if (limitingResource.isEmpty ||
(limitingResource.nonEmpty && numTasksPerExecCores < numTasksPerExecCustomResource)) {
@@ -453,10 +458,11 @@ private[spark] object ResourceUtils extends Logging {
val execReq = rp.getCustomExecutorResources()
if (limitingResource.nonEmpty && !limitingResource.equals(ResourceProfile.CPUS)) {
- if ((taskCpus * maxTaskPerExec) < cores) {
- val resourceNumSlots = Math.floor(cores/taskCpus).toInt
+ if ((taskCpus * maxTaskPerExec) < coresInternal) {
+ val resourceNumSlots = ResourceProfile.numTasksBasedOnCores(coresInternal, taskCpus)
+ val taskCpusDisplay = CpuAmount.toDisplayString(taskCpus)
val message = log"The configuration of cores (exec = ${MDC(NUM_CORES, cores)} " +
- log"task = ${MDC(NUM_TASK_CPUS, taskCpus)}, runnable tasks = " +
+ log"task = ${MDC(NUM_TASK_CPUS, taskCpusDisplay)}, runnable tasks = " +
log"${MDC(NUM_RESOURCE_SLOTS, resourceNumSlots)}) will " +
log"result in wasted resources due to resource ${MDC(RESOURCE, limitingResource)} " +
log"limiting the number of runnable tasks per executor to: " +
diff --git a/core/src/main/scala/org/apache/spark/resource/TaskResourceRequest.scala b/core/src/main/scala/org/apache/spark/resource/TaskResourceRequest.scala
index 2b9b3cdc6811..cdf1a1ec7bce 100644
--- a/core/src/main/scala/org/apache/spark/resource/TaskResourceRequest.scala
+++ b/core/src/main/scala/org/apache/spark/resource/TaskResourceRequest.scala
@@ -28,17 +28,29 @@ import org.apache.spark.annotation.{Since, Stable}
*
* @param resourceName Resource name
* @param amount Amount requesting as a Double to support fractional resource requests.
- * Valid values are less than or equal to 1.0 or whole numbers. This essentially
- * lets you configure X number of tasks to run on a single resource,
- * ie amount equals 0.5 translates into 2 tasks per resource address.
+ * For custom resources, valid values are less than or equal to 1.0 or whole
+ * numbers, since a task's amount must map onto discrete resource addresses -
+ * ie amount equals 0.5 translates into 2 tasks per resource address. CPUs
+ * (resource name "cpus") are a plain quantity drawn from the executor's core
+ * pool rather than an addressable resource, so any amount of at least 1e-9 is
+ * valid, e.g. 1.5; the cpus amount is rounded to the nearest 1e-9, so precision
+ * beyond 9 decimal places is not preserved.
*/
@Stable
@Since("3.1.0")
class TaskResourceRequest(val resourceName: String, val amount: Double)
extends Serializable {
- assert(amount <= 1.0 || amount % 1 == 0,
- s"The resource amount ${amount} must be either <= 1.0, or a whole number.")
+ // Cpus amounts are validated at the request entry points (TaskResourceRequests and the
+ // spark.task.cpus config parser), never here: this constructor also runs when the history
+ // server deserializes persisted data (JsonProtocol event logs, the protobuf KVStore), which
+ // can carry any cpus amount an earlier release accepted -- including 0 and even
+ // -Infinity (the pre-4.3 check was `amount <= 1.0 || whole`, and protobuf round-trips
+ // non-finite doubles) -- and must keep deserializing after an upgrade.
+ if (resourceName != ResourceProfile.CPUS) {
+ assert(amount <= 1.0 || amount % 1 == 0,
+ s"The resource amount ${amount} must be either <= 1.0, or a whole number.")
+ }
override def equals(obj: Any): Boolean = {
obj match {
diff --git a/core/src/main/scala/org/apache/spark/resource/TaskResourceRequests.scala b/core/src/main/scala/org/apache/spark/resource/TaskResourceRequests.scala
index e623a186551b..a67478a71b74 100644
--- a/core/src/main/scala/org/apache/spark/resource/TaskResourceRequests.scala
+++ b/core/src/main/scala/org/apache/spark/resource/TaskResourceRequests.scala
@@ -52,7 +52,34 @@ class TaskResourceRequests() extends Serializable {
*
* @param amount Number of cpus to allocate per Task.
*/
- def cpus(amount: Int): this.type = {
+ def cpus(amount: Int): this.type = cpus(amount.toDouble)
+
+ /**
+ * Specify number of cpus per Task.
+ * This is a convenient API to add [[TaskResourceRequest]] for cpus.
+ *
+ * @param amount Number of cpus to allocate per Task, valid from 1e-9 to Int.MaxValue
+ * after rounding. Fractional values are supported: below 1 (e.g. 0.2) to
+ * let multiple tasks share a CPU core, or above 1 (e.g. 1.5). The value is
+ * rounded to the nearest 1e-9 for internal CPU accounting, so precision
+ * beyond 9 decimal places is not preserved (this also absorbs the binary
+ * noise of computed `Double`s such as `0.1 + 0.2`).
+ */
+ @Since("4.3.0")
+ def cpus(amount: Double): this.type = {
+ // A positive amount below half of the internal accounting scale (1e-9) would silently
+ // round to zero cpus downstream, and one above Int.MaxValue could never be satisfied by
+ // any executor; reject both here where the original value is still visible. The bounds
+ // match spark.task.cpus, but are checked on the value rounded to the accounting scale,
+ // which is what scheduling actually uses: unlike the string-typed config, a Double
+ // carries binary noise (e.g. 0.1 + 0.2), so excess precision rounds instead of being
+ // rejected, and a value within half a step of the range rounds into it (e.g. 5e-10).
+ // Validation lives at this request entry point rather than in the TaskResourceRequest
+ // constructor, which must stay lenient to deserialize historical event logs and
+ // history-server stores written before cpus amounts were validated.
+ require(!amount.isNaN && !amount.isInfinity &&
+ CpuAmount.isInRange(CpuAmount.normalize(BigDecimal(amount.toString))),
+ s"The cpus amount ${amount} must be at least 1e-9 and at most ${Int.MaxValue}.")
val treq = new TaskResourceRequest(CPUS, amount)
_taskResources.put(CPUS, treq)
this
@@ -69,17 +96,28 @@ class TaskResourceRequests() extends Serializable {
* ie amount equals 0.5 translates into 2 tasks per resource address.
*/
def resource(resourceName: String, amount: Double): this.type = {
- val treq = new TaskResourceRequest(resourceName, amount)
- _taskResources.put(resourceName, treq)
- this
+ // Cpus amounts are validated only at request entry points (the TaskResourceRequest
+ // constructor stays lenient to deserialize persisted data), so route them through the
+ // validating method regardless of which entry point the caller picked.
+ if (resourceName == CPUS) {
+ cpus(amount)
+ } else {
+ val treq = new TaskResourceRequest(resourceName, amount)
+ _taskResources.put(resourceName, treq)
+ this
+ }
}
/**
* Add a certain [[TaskResourceRequest]] to the request set.
*/
def addRequest(treq: TaskResourceRequest): this.type = {
- _taskResources.put(treq.resourceName, treq)
- this
+ if (treq.resourceName == CPUS) {
+ cpus(treq.amount)
+ } else {
+ _taskResources.put(treq.resourceName, treq)
+ this
+ }
}
override def toString: String = {
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index 5e608308de29..b87ae41c4bc4 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -46,8 +46,8 @@ import org.apache.spark.network.shuffle.protocol.MergeStatuses
import org.apache.spark.network.util.JavaUtils
import org.apache.spark.partial.{ApproximateActionListener, ApproximateEvaluator, PartialResult}
import org.apache.spark.rdd.{RDD, RDDCheckpointData}
-import org.apache.spark.resource.{ResourceProfile, TaskResourceProfile}
-import org.apache.spark.resource.ResourceProfile.{DEFAULT_RESOURCE_PROFILE_ID, EXECUTOR_CORES_LOCAL_PROPERTY, PYSPARK_MEMORY_LOCAL_PROPERTY}
+import org.apache.spark.resource.{CpuAmount, ResourceProfile, TaskResourceProfile}
+import org.apache.spark.resource.ResourceProfile.{CPUS, DEFAULT_RESOURCE_PROFILE_ID, EXECUTOR_CORES_LOCAL_PROPERTY, MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY, PYSPARK_MEMORY_LOCAL_PROPERTY}
import org.apache.spark.rpc.RpcTimeout
import org.apache.spark.rpc.RpcTimeoutException
import org.apache.spark.storage._
@@ -1649,7 +1649,14 @@ private[spark] class DAGScheduler(
*/
private def addPySparkConfigsToProperties(stage: Stage, properties: Properties): Unit = {
val rp = sc.resourceProfileManager.resourceProfileFromId(stage.resourceProfileId)
+ // A profile that does not override pyspark memory runs its tasks on executors that were
+ // provisioned with the default profile's allocation: a task-only profile declares no
+ // executor resources at all, and getResourcesForClusterManager starts a full profile's
+ // executor request from the default allocation, overriding only what the profile
+ // specifies. Inherit the default value so the worker memory limit matches what the
+ // executor was sized with. An explicit pysparkMemory(0) still disables the limit.
val pysparkMem = rp.getPySparkMemory
+ .orElse(sc.resourceProfileManager.defaultResourceProfile.getPySparkMemory)
// use the getOption on EXECUTOR_CORES.key instead of using the EXECUTOR_CORES config reader
// because the default for this config isn't correct for standalone mode. Here we want
// to know if it was explicitly set or not. The default profile always has it set to either
@@ -1662,6 +1669,56 @@ private[spark] class DAGScheduler(
}
pysparkMem.map(mem => properties.setProperty(PYSPARK_MEMORY_LOCAL_PROPERTY, mem.toString))
execCores.map(cores => properties.setProperty(EXECUTOR_CORES_LOCAL_PROPERTY, cores))
+ // Pass the executor's max concurrent tasks so PySpark splits worker memory by real
+ // concurrency rather than cpu slots alone. Which bound is safe depends on who can share
+ // the executor:
+ // - With dynamic allocation off, default-profile and task-only-profile tasks can run on
+ // one executor concurrently. A custom-resource limit (e.g. a single gpu) bounds only its
+ // own profile's tasks, not the executor's total worker count, so only the
+ // cpus-proportional cores split keeps the aggregate of all co-scheduled workers within
+ // the executor-wide budget: with shares of budget / ceil(cores / taskCpus), each worker
+ // holds at most budget * taskCpus / cores, and the co-scheduled task cpus sum to at
+ // most the executor cores. (Ceiling division keeps this true even when task cpus do not
+ // evenly divide the cores, at the cost of slightly sub-proportional shares there.)
+ // - Otherwise executors serve a single profile and the profile's own limiting resource
+ // (across cores and custom resources) is the real concurrency. When the cores limit is
+ // unknown (standalone without an explicit spark.executor.cores, SPARK-30299), a
+ // custom-resource limit is still a valid upper bound on concurrency, capped by the cores
+ // bound derived from the profile's explicit executor cores when available: an uncapped
+ // overestimate would shrink -- or zero out, tripping the unsatisfiable-share fail-fast --
+ // every worker's memory share. Dividing by an upper bound under-allocates but never
+ // overcommits.
+ // maxTasksPerExecutor() also populates isCoresLimitKnown.
+ val maxTasks = rp.maxTasksPerExecutor(sc.conf)
+ val taskCpus = ResourceProfile.getTaskCpusOrDefaultForProfile(rp, sc.conf)
+ val knownCores = if (rp.isCoresLimitKnown) {
+ Some(rp.getExecutorCores.getOrElse(sc.conf.get(config.EXECUTOR_CORES)))
+ } else {
+ rp.getExecutorCores
+ }
+ val limitedByCustomResource = {
+ val limiting = rp.limitingResource(sc.conf)
+ limiting.nonEmpty && limiting != CPUS
+ }
+ val sharedExecutor = !Utils.isDynamicAllocationEnabled(sc.conf) &&
+ (rp.id == DEFAULT_RESOURCE_PROFILE_ID || rp.isInstanceOf[TaskResourceProfile])
+ val maxTasksToPropagate = if (sharedExecutor) {
+ knownCores.map { c =>
+ ResourceProfile.numMemoryShareSlotsCeil(CpuAmount.normalize(BigDecimal(c)), taskCpus)
+ }
+ } else if (rp.isCoresLimitKnown) {
+ Some(maxTasks)
+ } else if (limitedByCustomResource) {
+ val coresBasedMaxTasks = knownCores.map { c =>
+ ResourceProfile.numTasksBasedOnCores(CpuAmount.normalize(BigDecimal(c)), taskCpus)
+ }
+ Some(coresBasedMaxTasks.fold(maxTasks)(math.min(maxTasks, _)))
+ } else {
+ None
+ }
+ maxTasksToPropagate.foreach { n =>
+ properties.setProperty(MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY, n.toString)
+ }
}
/**
@@ -1865,8 +1922,14 @@ private[spark] class DAGScheduler(
val partitionsToCompute: Seq[Int] = stage.findMissingPartitions()
// Use the scheduling pool, job group, description, etc. from an ActiveJob associated
- // with this Stage
- val properties = jobIdToActiveJob(jobId).properties
+ // with this Stage. Clone it per stage so the profile-specific values written by
+ // addPySparkConfigsToProperties below don't leak -- through the shared, mutable job
+ // Properties instance passed to every task and TaskSet of the job -- into sibling stages
+ // of the same job that use a different resource profile. Job properties may be null when
+ // a job is submitted to the scheduler directly (e.g. in tests); materialize an empty map
+ // since the stage-local resource values are always written.
+ val properties = Option(Utils.cloneProperties(jobIdToActiveJob(jobId).properties))
+ .getOrElse(new Properties())
addPySparkConfigsToProperties(stage, properties)
runningStages += stage
diff --git a/core/src/main/scala/org/apache/spark/scheduler/Task.scala b/core/src/main/scala/org/apache/spark/scheduler/Task.scala
index 0e5e2c9faa05..de3fc25575fe 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/Task.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/Task.scala
@@ -88,7 +88,7 @@ private[spark] abstract class Task[T](
taskAttemptId: Long,
attemptNumber: Int,
metricsSystem: MetricsSystem,
- cpus: Int,
+ cpus: BigDecimal,
resources: Map[String, ResourceInformation],
plugins: Option[PluginContainer]): T = {
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskDescription.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskDescription.scala
index df5f32612bea..7eba116c5680 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskDescription.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskDescription.scala
@@ -27,6 +27,7 @@ import scala.collection.mutable.{HashMap, Map}
import scala.jdk.CollectionConverters._
import org.apache.spark.{JobArtifactSet, JobArtifactState}
+import org.apache.spark.resource.CpuAmount
import org.apache.spark.util.{ByteBufferInputStream, ByteBufferOutputStream, Utils}
/**
@@ -55,7 +56,7 @@ private[spark] class TaskDescription(
val partitionId: Int,
val artifacts: JobArtifactSet,
val properties: Properties,
- val cpus: Int,
+ val cpus: BigDecimal,
// resources is the total resources assigned to the task
// Eg, Map("gpu" -> Map("0" -> ResourceAmountUtils.toInternalResource(0.7))):
// assign 0.7 of the gpu address "0" to this task
@@ -113,8 +114,9 @@ private[spark] object TaskDescription {
dataOut.write(bytes)
}
- // Write cpus.
- dataOut.writeInt(taskDescription.cpus)
+ // Write cpus in its minimal (trailing-zero-trimmed) decimal string form; it is re-normalized
+ // to the fixed CPU scale on read, so fractional values round-trip exactly.
+ dataOut.writeUTF(CpuAmount.toDisplayString(taskDescription.cpus))
// Write resources.
serializeResources(taskDescription.resources, dataOut)
@@ -220,8 +222,8 @@ private[spark] object TaskDescription {
properties.setProperty(key, new String(valueBytes, StandardCharsets.UTF_8))
}
- // Read cpus.
- val cpus = dataIn.readInt()
+ // Read cpus and normalize to the fixed CPU accounting scale.
+ val cpus = CpuAmount.normalize(BigDecimal(dataIn.readUTF()))
// Read resources.
val resources = deserializeResources(dataIn)
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
index 618c8eb45902..d79d69cf65b7 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
@@ -116,7 +116,9 @@ private[spark] class TaskSchedulerImpl(
// Threshold above which we warn user initial TaskSet may be starved
val STARVATION_TIMEOUT_MS = conf.getTimeAsMs("spark.starvation.timeout", "15s")
- // CPUs to request per task
+ // CPUs to request per task, in the internal exact BigDecimal representation so that fractional
+ // values (e.g. 0.2) are accounted exactly. spark.task.cpus is a decimalConf, so conf.get returns
+ // the exact BigDecimal the user configured.
val CPUS_PER_TASK = conf.get(config.CPUS_PER_TASK)
// TaskSetManagers are not thread safe, so any access to one should be synchronized
@@ -403,12 +405,17 @@ private[spark] class TaskSchedulerImpl(
taskSet: TaskSetManager,
maxLocality: TaskLocality,
shuffledOffers: Seq[WorkerOffer],
- availableCpus: Array[Int],
+ availableCpus: Array[BigDecimal],
availableResources: Array[ExecutorResourcesAmounts],
tasks: IndexedSeq[ArrayBuffer[TaskDescription]])
: (Boolean, Option[TaskLocality]) = {
var noDelayScheduleRejects = true
var minLaunchedLocality: Option[TaskLocality] = None
+ // Resolve the task cpus once per (task set, locality) round; the per-offer probes below
+ // would otherwise re-derive it for every slot.
+ val taskSetProf = sc.resourceProfileManager
+ .resourceProfileFromId(taskSet.taskSet.resourceProfileId)
+ val taskCpus = ResourceProfile.getTaskCpusOrDefaultForProfile(taskSetProf, conf)
// nodes and executors that are excluded for the entire application have already been
// filtered out by this point
for (i <- shuffledOffers.indices) {
@@ -419,12 +426,10 @@ private[spark] class TaskSchedulerImpl(
// check whether the task can be scheduled to the executor base on resource profile.
if (sc.resourceProfileManager
.canBeScheduled(taskSetRpID, shuffledOffers(i).resourceProfileId)) {
- val taskResAssignmentsOpt = resourcesMeetTaskRequirements(taskSet, availableCpus(i),
- availableResources(i))
+ val taskResAssignmentsOpt = resourcesMeetTaskRequirements(taskSet, taskCpus,
+ availableCpus(i), availableResources(i))
taskResAssignmentsOpt.foreach { taskResAssignments =>
try {
- val prof = sc.resourceProfileManager.resourceProfileFromId(taskSetRpID)
- val taskCpus = ResourceProfile.getTaskCpusOrDefaultForProfile(prof, conf)
val (taskDescOption, didReject, index) =
taskSet.resourceOffer(execId, host, maxLocality, taskCpus, taskResAssignments)
noDelayScheduleRejects &= !didReject
@@ -442,8 +447,8 @@ private[spark] class TaskSchedulerImpl(
}
minLaunchedLocality = minTaskLocality(minLaunchedLocality, Some(locality))
- availableCpus(i) -= taskCpus
- assert(availableCpus(i) >= 0)
+ availableCpus(i) = availableCpus(i) - taskCpus
+ assert(availableCpus(i).signum >= 0)
availableResources(i).acquire(resources)
}
} catch {
@@ -479,12 +484,14 @@ private[spark] class TaskSchedulerImpl(
*/
private def resourcesMeetTaskRequirements(
taskSet: TaskSetManager,
- availCpus: Int,
+ taskCpus: BigDecimal,
+ availCpus: BigDecimal,
availWorkerResources: ExecutorResourcesAmounts): Option[Map[String, Map[String, Long]]] = {
val rpId = taskSet.taskSet.resourceProfileId
val taskSetProf = sc.resourceProfileManager.resourceProfileFromId(rpId)
- val taskCpus = ResourceProfile.getTaskCpusOrDefaultForProfile(taskSetProf, conf)
- // check if the ResourceProfile has cpus first since that is common case
+ // check if the ResourceProfile has cpus first since that is common case. Both values are in
+ // the internal exact BigDecimal representation, so this comparison is exact regardless of
+ // whether spark.task.cpus is fractional (e.g. 0.2).
if (availCpus < taskCpus) return None
// only look at the resource other than cpus
availWorkerResources.assignAddressesCustomResources(taskSetProf)
@@ -547,8 +554,13 @@ private[spark] class TaskSchedulerImpl(
val shuffledOffers = shuffleOffers(filteredOffers)
// Build a list of tasks to assign to each worker.
// Note the size estimate here might be off with different ResourceProfiles but should be
- // close estimate
- val tasks = shuffledOffers.map(o => new ArrayBuffer[TaskDescription](o.cores / CPUS_PER_TASK))
+ // close estimate. It is only a capacity hint, so cap it: with a tiny fractional
+ // spark.task.cpus the exact slot count can be huge and would preallocate a giant buffer.
+ val tasks = shuffledOffers.map { o =>
+ val sizeHint =
+ math.min(ResourceProfile.numTasksBasedOnCores(o.cores, CPUS_PER_TASK), 1024)
+ new ArrayBuffer[TaskDescription](sizeHint)
+ }
val availableResources = shuffledOffers.map(_.resources).toArray
val availableCpus = shuffledOffers.map(o => o.cores).toArray
val resourceProfileIds = shuffledOffers.map(o => o.resourceProfileId).toArray
@@ -706,7 +718,8 @@ private[spark] class TaskSchedulerImpl(
}
barrierPendingLaunchTasks.foreach { task =>
// revert all assigned resources
- availableCpus(task.assignedOfferIndex) += task.assignedCores
+ availableCpus(task.assignedOfferIndex) =
+ availableCpus(task.assignedOfferIndex) + task.assignedCores
availableResources(task.assignedOfferIndex).release(
task.assignedResources)
// re-add the task to the schedule pending list
@@ -1241,7 +1254,7 @@ private[spark] object TaskSchedulerImpl {
conf: SparkConf,
rpId: Int,
availableRPIds: Array[Int],
- availableCpus: Array[Int],
+ availableCpus: Array[BigDecimal],
availableResources: Array[Map[String, Int]]): Int = {
val resourceProfile = scheduler.sc.resourceProfileManager.resourceProfileFromId(rpId)
val coresKnown = resourceProfile.isCoresLimitKnown
@@ -1257,10 +1270,10 @@ private[spark] object TaskSchedulerImpl {
val cpusPerTask = ResourceProfile.getTaskCpusOrDefaultForProfile(resourceProfile, conf)
val taskLimit = resourceProfile.taskResources.get(limitingResource).map(_.amount).get
- availableCpus.zip(availableResources).zip(availableRPIds)
+ val totalSlots = availableCpus.zip(availableResources).zip(availableRPIds)
.filter { case (_, id) => scheduler.sc.resourceProfileManager.canBeScheduled(rpId, id) }
.map { case ((cpu, resources), _) =>
- val numTasksPerExecCores = cpu / cpusPerTask
+ val numTasksPerExecCores = ResourceProfile.numTasksBasedOnCores(cpu, cpusPerTask)
if (limitedByCpu) {
numTasksPerExecCores
} else {
@@ -1279,7 +1292,11 @@ private[spark] object TaskSchedulerImpl {
resourceLimit
}
}
- }.sum
+ }.foldLeft(0L)(_ + _)
+ // The per-executor slot counts are individually clamped to the Int range, but with a tiny
+ // fractional task cpus amount their sum can still overflow Int; saturate instead of
+ // wrapping so this never returns a negative slot count.
+ totalSlots.min(Int.MaxValue).max(0L).toInt
}
/**
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
index 78e4fc78b880..3a5f60ae1935 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
@@ -33,6 +33,7 @@ import org.apache.spark.errors.SparkCoreErrors
import org.apache.spark.internal.{config, Logging, LogKeys}
import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
+import org.apache.spark.resource.CpuAmount
import org.apache.spark.scheduler.SchedulingMode._
import org.apache.spark.util.{AccumulatorV2, Clock, LongAccumulator, SystemClock, Utils}
import org.apache.spark.util.collection.{OpenHashSet, PercentileHeap}
@@ -462,7 +463,7 @@ private[spark] class TaskSetManager(
execId: String,
host: String,
maxLocality: TaskLocality.TaskLocality,
- taskCpus: Int = sched.CPUS_PER_TASK,
+ taskCpus: BigDecimal = sched.CPUS_PER_TASK,
taskResourceAssignments: Map[String, Map[String, Long]] = Map.empty)
: (Option[TaskDescription], Boolean, Int) =
{
@@ -530,7 +531,7 @@ private[spark] class TaskSetManager(
index: Int,
taskLocality: TaskLocality.Value,
speculative: Boolean,
- taskCpus: Int,
+ taskCpus: BigDecimal,
taskResourceAssignments: Map[String, Map[String, Long]],
launchTime: Long): TaskDescription = {
// Found a task; do some bookkeeping and return a task description
@@ -1545,5 +1546,5 @@ private[scheduler] case class BarrierPendingLaunchTask(
// Used to revert the assigned resources (e.g., cores, custome resources) when the barrier
// task set doesn't launch successfully in a single resourceOffers round.
var assignedOfferIndex: Int = _
- var assignedCores: Int = 0
+ var assignedCores: BigDecimal = CpuAmount.normalize(BigDecimal(0))
}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/WorkerOffer.scala b/core/src/main/scala/org/apache/spark/scheduler/WorkerOffer.scala
index a7d63a8949e1..8af1470e2564 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/WorkerOffer.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/WorkerOffer.scala
@@ -26,7 +26,7 @@ private[spark]
case class WorkerOffer(
executorId: String,
host: String,
- cores: Int,
+ cores: BigDecimal,
// `address` is an optional hostPort string, it provide more useful information than `host`
// when multiple executors are launched on the same host.
address: Option[String] = None,
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala
index 1f452ae7d109..84a6f9b3be2a 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala
@@ -79,7 +79,7 @@ private[spark] object CoarseGrainedClusterMessages {
taskId: Long,
state: TaskState,
data: SerializableBuffer,
- taskCpus: Int,
+ taskCpus: BigDecimal,
resources: Map[String, Map[String, Long]] = Map.empty)
extends CoarseGrainedClusterMessage
@@ -90,7 +90,7 @@ private[spark] object CoarseGrainedClusterMessages {
taskId: Long,
state: TaskState,
data: ByteBuffer,
- taskCpus: Int,
+ taskCpus: BigDecimal,
resources: Map[String, Map[String, Long]]): StatusUpdate = {
StatusUpdate(executorId, taskId, state, new SerializableBuffer(data), taskCpus, resources)
}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
index ad92e22424c7..e4bdcd9fa7a3 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
@@ -35,7 +35,7 @@ import org.apache.spark.internal.LogKeys
import org.apache.spark.internal.LogKeys._
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.Network._
-import org.apache.spark.resource.ResourceProfile
+import org.apache.spark.resource.{CpuAmount, ResourceProfile}
import org.apache.spark.rpc._
import org.apache.spark.scheduler._
import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages._
@@ -232,7 +232,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
case LaunchedExecutor(executorId) =>
executorDataMap.get(executorId).foreach { data =>
- data.freeCores = data.totalCores
+ data.freeCores = CpuAmount.normalize(BigDecimal(data.totalCores))
}
makeOffers(executorId)
@@ -747,7 +747,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
.values.toArray.map { executor =>
(
executor.resourceProfileId,
- executor.totalCores,
+ CpuAmount.normalize(BigDecimal(executor.totalCores)),
executor.resourcesInfo.map { case (name, rInfo) =>
(name, rInfo.totalAddressesAmount)
}
@@ -765,7 +765,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
// this function is for testing only
private[spark] def getExecutorAvailableCpus(
- executorId: String): Option[Int] = synchronized {
+ executorId: String): Option[BigDecimal] = synchronized {
executorDataMap.get(executorId).map(_.freeCores)
}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/ExecutorData.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/ExecutorData.scala
index 07236d4007fa..619cda4dad50 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/cluster/ExecutorData.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/ExecutorData.scala
@@ -26,7 +26,10 @@ import org.apache.spark.scheduler.ExecutorResourceInfo
* @param executorEndpoint The RpcEndpointRef representing this executor
* @param executorAddress The network address of this executor
* @param executorHost The hostname that this executor is running on
- * @param freeCores The current number of cores available for work on the executor
+ * @param freeCores The current number of cores available for work on the executor, in the
+ * internal exact BigDecimal representation so it can be fractional (e.g. when
+ * `spark.task.cpus` is set to a fractional value like 0.2, multiple tasks can
+ * share a single core).
* @param totalCores The total number of cores available to the executor
* @param resourcesInfo The information of the currently available resources on the executor
* @param resourceProfileId The id of the ResourceProfile being used by this executor
@@ -37,7 +40,7 @@ private[cluster] class ExecutorData(
val executorEndpoint: RpcEndpointRef,
val executorAddress: RpcAddress,
override val executorHost: String,
- var freeCores: Int,
+ var freeCores: BigDecimal,
override val totalCores: Int,
override val logUrlMap: Map[String, String],
override val attributes: Map[String, String],
diff --git a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala
index 56c1fccd8439..c3a1a60a0384 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala
@@ -28,7 +28,7 @@ import org.apache.spark.deploy.security.HadoopDelegationTokenManager
import org.apache.spark.executor.{Executor, ExecutorBackend}
import org.apache.spark.internal.{config, Logging, LogKeys}
import org.apache.spark.launcher.{LauncherBackend, SparkAppHandle}
-import org.apache.spark.resource.{ResourceInformation, ResourceProfile}
+import org.apache.spark.resource.{CpuAmount, ResourceInformation, ResourceProfile}
import org.apache.spark.rpc.{RpcCallContext, RpcEndpointRef, RpcEnv, ThreadSafeRpcEndpoint}
import org.apache.spark.scheduler._
import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.{TaskThreadDump, UpdateDelegationTokens}
@@ -57,7 +57,7 @@ private[spark] class LocalEndpoint(
private val totalCores: Int)
extends ThreadSafeRpcEndpoint with Logging {
- private var freeCores = totalCores
+ private var freeCores: BigDecimal = CpuAmount.normalize(BigDecimal(totalCores))
val localExecutorId = SparkContext.DRIVER_IDENTIFIER
val localExecutorHostname = Utils.localCanonicalHostName()
@@ -189,7 +189,7 @@ private[spark] class LocalSchedulerBackend(
// so we expect all executors to be of same ResourceProfile
override def maxNumConcurrentTasks(rp: ResourceProfile): Int = {
val cpusPerTask = ResourceProfile.getTaskCpusOrDefaultForProfile(rp, conf)
- totalCores / cpusPerTask
+ ResourceProfile.numTasksBasedOnCores(CpuAmount.normalize(BigDecimal(totalCores)), cpusPerTask)
}
private def stop(finalState: SparkAppHandle.State): Unit = {
diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala b/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala
index 5674798f0209..ff38d77b2b0e 100644
--- a/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala
+++ b/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala
@@ -23,12 +23,14 @@ import java.util.concurrent.ConcurrentHashMap
import scala.collection.mutable.{ArrayBuffer, HashMap}
import scala.collection.mutable
import scala.jdk.CollectionConverters._
+import scala.util.Try
import org.apache.spark._
import org.apache.spark.executor.{ExecutorMetrics, TaskMetrics}
import org.apache.spark.internal.{Logging, LogKeys}
import org.apache.spark.internal.config.CPUS_PER_TASK
import org.apache.spark.internal.config.Status._
+import org.apache.spark.resource.{CpuAmount, ResourceProfile}
import org.apache.spark.resource.ResourceProfile.CPUS
import org.apache.spark.scheduler._
import org.apache.spark.status.api.v1
@@ -53,7 +55,7 @@ private[spark] class AppStatusListener(
private var sparkVersion = SPARK_VERSION
private var appInfo: v1.ApplicationInfo = null
private var appSummary = new AppSummary(0, 0)
- private var defaultCpusPerTask: Int = 1
+ private var defaultCpusPerTask: BigDecimal = CpuAmount.normalize(BigDecimal(1))
// How often to update live entities. -1 means "never update" when replaying applications,
// meaning only the last write will happen. For live applications, this avoids a few
@@ -183,8 +185,20 @@ private[spark] class AppStatusListener(
details.getOrElse("Classpath Entries", Nil),
Nil)
- defaultCpusPerTask = envInfo.sparkProperties.toMap.get(CPUS_PER_TASK.key).map(_.toInt)
- .getOrElse(defaultCpusPerTask)
+ // Parse the raw property string so the default cpus per task is exact (spark.task.cpus
+ // may be fractional, e.g. 0.2). The string comes from a persisted application environment
+ // and is untrusted -- parseUntrusted bounds the length and range before the costly parse
+ // and normalize steps can run inside the (shared) history server. A legitimate log always
+ // carries a value the live config parser validated, so malformed values keep the previous
+ // default instead of failing the replay.
+ envInfo.sparkProperties.toMap.get(CPUS_PER_TASK.key).foreach { v =>
+ CpuAmount.parseUntrusted(v) match {
+ case Some(d) => defaultCpusPerTask = d
+ case None =>
+ logWarning(log"Ignoring invalid ${MDC(LogKeys.CONFIG, CPUS_PER_TASK.key)} value in " +
+ log"the application environment; keeping the previous default")
+ }
+ }
kvstore.write(new ApplicationEnvironmentInfoWrapper(envInfo))
}
@@ -221,10 +235,20 @@ private[spark] class AppStatusListener(
exec.totalCores = event.executorInfo.totalCores
val rpId = event.executorInfo.resourceProfileId
val liveRP = liveResourceProfiles.get(rpId)
+ // Recover the exact BigDecimal from the Double amount via its string form (see
+ // ResourceProfile.getTaskCpus for why this must go through toString), then normalize.
+ // A replayed profile can carry any amount an earlier release accepted (e.g. a negative
+ // cpus): fall back to the default rather than letting the slot math's positivity
+ // assertion abort the replay.
val cpusPerTask = liveRP.flatMap(_.taskResources.get(CPUS))
- .map(_.amount.toInt).getOrElse(defaultCpusPerTask)
+ .flatMap(tr => Try(BigDecimal(tr.amount.toString)).toOption)
+ .filter(CpuAmount.isInRange)
+ .map(CpuAmount.normalize)
+ .getOrElse(defaultCpusPerTask)
val maxTasksPerExec = liveRP.flatMap(_.maxTasksPerExecutor)
- exec.maxTasks = maxTasksPerExec.getOrElse(event.executorInfo.totalCores / cpusPerTask)
+ exec.maxTasks = maxTasksPerExec.getOrElse(
+ ResourceProfile.numTasksBasedOnCores(
+ CpuAmount.normalize(BigDecimal(event.executorInfo.totalCores)), cpusPerTask))
exec.executorLogs = event.executorInfo.logUrlMap
exec.resources = event.executorInfo.resourcesInfo
exec.attributes = event.executorInfo.attributes
diff --git a/core/src/test/scala/org/apache/spark/BarrierStageOnSubmittedSuite.scala b/core/src/test/scala/org/apache/spark/BarrierStageOnSubmittedSuite.scala
index 13d479e35aae..ec7162c6ffb6 100644
--- a/core/src/test/scala/org/apache/spark/BarrierStageOnSubmittedSuite.scala
+++ b/core/src/test/scala/org/apache/spark/BarrierStageOnSubmittedSuite.scala
@@ -228,7 +228,7 @@ class BarrierStageOnSubmittedSuite extends SparkFunSuite with LocalSparkContext
test("submit a barrier ResultStage that requires more slots than current total under " +
"local-cluster mode") {
val conf = new SparkConf()
- .set(CPUS_PER_TASK, 2)
+ .set(CPUS_PER_TASK, BigDecimal(2))
// Shorten the time interval between two failed checks to make the test fail faster.
.set(BARRIER_MAX_CONCURRENT_TASKS_CHECK_INTERVAL.key, "1s")
// Reduce max check failures allowed to make the test fail faster.
@@ -246,7 +246,7 @@ class BarrierStageOnSubmittedSuite extends SparkFunSuite with LocalSparkContext
test("submit a barrier ShuffleMapStage that requires more slots than current total under " +
"local-cluster mode") {
val conf = new SparkConf()
- .set(CPUS_PER_TASK, 2)
+ .set(CPUS_PER_TASK, BigDecimal(2))
// Shorten the time interval between two failed checks to make the test fail faster.
.set(BARRIER_MAX_CONCURRENT_TASKS_CHECK_INTERVAL.key, "1s")
// Reduce max check failures allowed to make the test fail faster.
diff --git a/core/src/test/scala/org/apache/spark/SparkContextSuite.scala b/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
index 517506b0f26f..709f267de155 100644
--- a/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
@@ -891,7 +891,7 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
("yarn", 2, Option(1))
).foreach { case (master, cpusPerTask, executorCores) =>
val conf = new SparkConf()
- conf.set(CPUS_PER_TASK, cpusPerTask)
+ conf.set(CPUS_PER_TASK, BigDecimal(cpusPerTask))
executorCores.map(executorCores => conf.set(EXECUTOR_CORES, executorCores))
val ex = intercept[SparkException] {
sc = new SparkContext(master, "test", conf)
@@ -1008,7 +1008,7 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
val conf = new SparkConf()
.setMaster("local-cluster[3, 2, 1024]")
.setAppName("test-cluster")
- .set(CPUS_PER_TASK, 2)
+ .set(CPUS_PER_TASK, BigDecimal(2))
.set(WORKER_GPU_ID.amountConf, "3")
.set(WORKER_GPU_ID.discoveryScriptConf, discoveryScript)
.set(TASK_GPU_ID.amountConf, "3")
@@ -1021,7 +1021,7 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
val rdd1 = sc.makeRDD(1 to 10, 3).mapPartitions { it =>
val context = TaskContext.get()
- Iterator(context.cpus())
+ Iterator(context.cpuAmount())
}
val cpus = rdd1.collect()
assert(cpus === Array(2, 2, 2))
@@ -1475,11 +1475,31 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
}
test("SPARK-55757: Improve `spark.task.cpus` validation") {
- val conf = new SparkConf().setAppName("test").setMaster("local").set(CPUS_PER_TASK, 0)
+ val conf = new SparkConf().setAppName("test").setMaster("local")
+ .set(CPUS_PER_TASK, BigDecimal(0))
val m = intercept[SparkIllegalArgumentException] {
sc = new SparkContext(conf)
}.getMessage
- assert(m.contains("Number of cores to allocate for each task should be positive."))
+ assert(m.contains("Number of cores to allocate for each task must be a positive value"))
+ }
+
+ test("SPARK-58192: spark.task.cpus rejects out-of-range and over-precise values") {
+ def read(value: String): BigDecimal = {
+ new SparkConf().set(CPUS_PER_TASK.key, value).get(CPUS_PER_TASK)
+ }
+ assert(read("1.5") == BigDecimal("1.5"))
+ assert(read("0.000000001") == BigDecimal("1E-9"))
+ assert(read("2147483647") == BigDecimal(Int.MaxValue))
+ // trailing zeros beyond 9 decimal places are stripped before the precision check
+ assert(read("1.5000000000000") == BigDecimal("1.5"))
+ // Includes extreme exponents: they must be rejected by the bounds check before the
+ // normalization step could materialize their huge unscaled representation.
+ Seq("0", "-1", "4e-10", "2147483648", "1e100000000", "1e1000000000").foreach { v =>
+ val e = intercept[SparkIllegalArgumentException] { read(v) }
+ assert(e.getMessage.contains("must be a positive value between"), s"for value $v")
+ }
+ val e = intercept[SparkIllegalArgumentException] { read("0.1234567891") }
+ assert(e.getMessage.contains("supports at most 9 decimal places"))
}
test("SPARK-57867: Driver should not reserve off-heap memory in non-local mode") {
diff --git a/core/src/test/scala/org/apache/spark/api/python/BasePythonRunnerSuite.scala b/core/src/test/scala/org/apache/spark/api/python/BasePythonRunnerSuite.scala
new file mode 100644
index 000000000000..5477e8f7488c
--- /dev/null
+++ b/core/src/test/scala/org/apache/spark/api/python/BasePythonRunnerSuite.scala
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.api.python
+
+import java.util.concurrent.{CountDownLatch, TimeUnit}
+
+import org.apache.spark.{SparkException, SparkFunSuite}
+
+class BasePythonRunnerSuite extends SparkFunSuite {
+
+ test("SPARK-58192: pipelined writer pool admits more concurrent writers than processors") {
+ // A fractional spark.task.cpus admits more concurrent tasks -- each holding one writer
+ // thread -- than the host has processors. Writers of a barrier stage block until every
+ // task's worker makes progress, so a pool that queues writers beyond a processor-based
+ // cap can deadlock; all writers must be able to start concurrently.
+ val nWriters = Runtime.getRuntime.availableProcessors() + 2
+ val started = new CountDownLatch(nWriters)
+ val release = new CountDownLatch(1)
+ val futures = (1 to nWriters).map { _ =>
+ BasePythonRunner.pipelinedWriterThreadPool.submit(new Runnable {
+ override def run(): Unit = {
+ started.countDown()
+ release.await(1, TimeUnit.MINUTES)
+ }
+ })
+ }
+ try {
+ assert(started.await(1, TimeUnit.MINUTES),
+ s"expected all $nWriters writers to start concurrently")
+ } finally {
+ release.countDown()
+ futures.foreach(_.get(1, TimeUnit.MINUTES))
+ }
+ }
+
+ test("SPARK-58192: pyspark memory is split across the executor's concurrent task slots") {
+ def workerMemoryMb(maxConcurrentTasks: Int): Option[Long] =
+ BasePythonRunner.getWorkerMemoryMb(Some(4096L), maxConcurrentTasks)
+
+ // One slot per concurrent task; each worker gets an equal share of the executor allocation.
+ assert(workerMemoryMb(4) === Some(1024L))
+ // A fractional spark.task.cpus (e.g. 4 cores / 0.5) admits more concurrent tasks than cores,
+ // so each worker gets a smaller share and the aggregate stays within the allocation.
+ assert(workerMemoryMb(8) === Some(512L))
+ assert(workerMemoryMb(5) === Some(819L))
+ // The concurrency is the limiting resource, not just cpu slots: a GPU that caps the executor
+ // at a single concurrent task means the sole worker keeps the whole allocation, even though
+ // the cpu-slot count (e.g. 64 cores / 0.1 = 640) is far higher.
+ assert(workerMemoryMb(1) === Some(4096L))
+ // Never split into less than one slot.
+ assert(BasePythonRunner.getWorkerMemoryMb(Some(4096L), 0) === Some(4096L))
+ // No pyspark memory configured means no per-worker limit.
+ assert(BasePythonRunner.getWorkerMemoryMb(None, 8) === None)
+ }
+
+ test("SPARK-58192: fail fast when the per-slot pyspark memory share rounds to zero") {
+ // 512 MiB across 640 genuinely concurrent tasks rounds down to 0, which the worker would
+ // treat as "no limit". Fail fast rather than silently dropping the configured cap.
+ val e = intercept[SparkException] {
+ BasePythonRunner.getWorkerMemoryMb(Some(512L), 640)
+ }
+ assert(e.getMessage.contains("spark.executor.pyspark.memory"))
+ assert(e.getMessage.contains("640"))
+ // A share of exactly 1 MiB is still enforceable and must not fail.
+ assert(BasePythonRunner.getWorkerMemoryMb(Some(640L), 640) === Some(1L))
+ // An explicit spark.executor.pyspark.memory=0 means the limit is disabled (the worker
+ // treats 0 as "no limit"), so it must pass through rather than fail fast, even when
+ // concurrency is high enough that a positive budget would round to zero.
+ assert(BasePythonRunner.getWorkerMemoryMb(Some(0L), 640) === Some(0L))
+ assert(BasePythonRunner.getWorkerMemoryMb(Some(0L), 1) === Some(0L))
+ // The smallest positive budget still fails fast when it cannot give every slot at least
+ // 1 MiB; only an explicit budget of 0 disables the limit.
+ intercept[SparkException] {
+ BasePythonRunner.getWorkerMemoryMb(Some(1L), 2)
+ }
+ }
+}
diff --git a/core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala b/core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala
index 8b6925b67de7..4362eb2f816f 100644
--- a/core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala
+++ b/core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala
@@ -50,7 +50,7 @@ import org.apache.spark.internal.config.UI._
import org.apache.spark.memory.{SparkOutOfMemoryError, TestMemoryManager}
import org.apache.spark.metrics.MetricsSystem
import org.apache.spark.rdd.RDD
-import org.apache.spark.resource.ResourceInformation
+import org.apache.spark.resource.{CpuAmount, ResourceInformation}
import org.apache.spark.rpc.{RpcEndpointRef, RpcEnv, RpcTimeout}
import org.apache.spark.scheduler.{DirectTaskResult, FakeTask, ResultTask, Task, TaskDescription}
import org.apache.spark.serializer.{JavaSerializer, SerializerInstance, SerializerManager}
@@ -664,6 +664,20 @@ class ExecutorSuite extends SparkFunSuite
}
}
+ test("SPARK-58192: executorRunTime weights the run interval by cpus, floored at one") {
+ // cpus > 1 scales the interval so CPU-weighted process rates reflect the reservation
+ // (SPARK-51666).
+ assert(Executor.cpuWeightedNanos(1000L, BigDecimal(2)) === 2000L)
+ assert(Executor.cpuWeightedNanos(1000L, CpuAmount.normalize(BigDecimal("1.5"))) === 1500L)
+ assert(Executor.cpuWeightedNanos(1000L, BigDecimal(1)) === 1000L)
+ // A sub-core amount must not shrink the reported run time below the elapsed interval:
+ // UI/REST scheduler delay is wall-clock duration minus executorRunTime, so anything below
+ // the elapsed interval is misreported as scheduler delay.
+ assert(Executor.cpuWeightedNanos(1000L, CpuAmount.normalize(BigDecimal("0.25"))) === 1000L)
+ assert(Executor.cpuWeightedNanos(1000L, CpuAmount.normalize(BigDecimal("0.000000001")))
+ === 1000L)
+ }
+
test(
"SPARK-55093: launchTask should handle TaskRunner construction failures"
) {
diff --git a/core/src/test/scala/org/apache/spark/resource/ResourceProfileManagerSuite.scala b/core/src/test/scala/org/apache/spark/resource/ResourceProfileManagerSuite.scala
index ab57f1c801b0..ce57f053a3e7 100644
--- a/core/src/test/scala/org/apache/spark/resource/ResourceProfileManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/resource/ResourceProfileManagerSuite.scala
@@ -53,6 +53,69 @@ class ResourceProfileManagerSuite extends SparkFunSuite {
s"Executor resources should have 4 cores")
}
+ test("SPARK-58192: malformed profiles are rejected before entering the registry") {
+ val conf = new SparkConf().set(EXECUTOR_CORES, 4)
+ val rpmanager = new ResourceProfileManager(conf, listenerBus)
+
+ // A raw-constructed profile with a NaN cpus amount: the TaskResourceRequest constructor
+ // stays lenient (deserialization of persisted data), so registration is the enforcement
+ // point and must reject it before it becomes visible to the application.
+ val nanProfile = new ResourceProfile(
+ Map.empty,
+ Map(ResourceProfile.CPUS -> new TaskResourceRequest(ResourceProfile.CPUS, Double.NaN)))
+ val e1 = intercept[IllegalArgumentException] {
+ rpmanager.addResourceProfile(nanProfile)
+ }
+ assert(e1.getMessage.contains("must be at least 1e-9"))
+ intercept[SparkException] {
+ rpmanager.resourceProfileFromId(nanProfile.id)
+ }
+
+ // The cpus amount is validated under the map key -- the identity scheduling uses -- so a
+ // sub-scale amount smuggled under the cpus key with a different embedded resource name is
+ // rejected the same way.
+ val mismatchCpus = new TaskResourceProfile(
+ Map(ResourceProfile.CPUS -> new TaskResourceRequest("gpu", 1e-10)))
+ val e2 = intercept[IllegalArgumentException] {
+ rpmanager.addResourceProfile(mismatchCpus)
+ }
+ assert(e2.getMessage.contains("must be at least 1e-9"))
+ intercept[SparkException] {
+ rpmanager.resourceProfileFromId(mismatchCpus.id)
+ }
+
+ // An amount above any possible executor's core count could never be scheduled.
+ val oversized = new TaskResourceProfile(
+ Map(ResourceProfile.CPUS ->
+ new TaskResourceRequest(ResourceProfile.CPUS, 2147483647.5)))
+ val e3 = intercept[IllegalArgumentException] {
+ rpmanager.addResourceProfile(oversized)
+ }
+ assert(e3.getMessage.contains("at most"))
+ intercept[SparkException] {
+ rpmanager.resourceProfileFromId(oversized.id)
+ }
+
+ // The inverse mismatch -- a cpus-named request under a custom key -- fails the forced
+ // limiting-resource computation (no matching executor resource) before insertion.
+ val mismatchCustom = new TaskResourceProfile(
+ Map("gpu" -> new TaskResourceRequest(ResourceProfile.CPUS, 1.5)))
+ intercept[SparkException] {
+ rpmanager.addResourceProfile(mismatchCustom)
+ }
+ intercept[SparkException] {
+ rpmanager.resourceProfileFromId(mismatchCustom.id)
+ }
+
+ // A valid profile built through the public builder still registers.
+ val valid = new ResourceProfileBuilder()
+ .require(new ExecutorResourceRequests().cores(4))
+ .require(new TaskResourceRequests().cpus(0.5))
+ .build()
+ rpmanager.addResourceProfile(valid)
+ assert(rpmanager.resourceProfileFromId(valid.id) === valid)
+ }
+
test("isSupported yarn no dynamic allocation") {
val conf = new SparkConf().setMaster("yarn").set(EXECUTOR_CORES, 4)
conf.set(RESOURCE_PROFILE_MANAGER_TESTING.key, "true")
diff --git a/core/src/test/scala/org/apache/spark/resource/ResourceProfileSuite.scala b/core/src/test/scala/org/apache/spark/resource/ResourceProfileSuite.scala
index 301f97834913..8f92c474a525 100644
--- a/core/src/test/scala/org/apache/spark/resource/ResourceProfileSuite.scala
+++ b/core/src/test/scala/org/apache/spark/resource/ResourceProfileSuite.scala
@@ -408,6 +408,85 @@ class ResourceProfileSuite extends SparkFunSuite with MockitoSugar {
assert(msg.contains("The task resource amount 0.7 must be either <= 0.5, or a whole number"))
}
+ test("SPARK-58192: fractional cpus amounts above 0.5 and above 1 are valid") {
+ // The "<= 0.5 or whole number" rule only applies to addressable custom resources; cpus is
+ // a plain quantity drawn from the core pool, so any amount of at least 1e-9 is valid.
+ Seq(0.7, 1.5, 2.5).foreach { amount =>
+ val ereqs = new ExecutorResourceRequests().cores(4)
+ val treqs = new TaskResourceRequests().cpus(amount)
+ val rp = new ResourceProfileBuilder().require(ereqs).require(treqs).build()
+ assert(rp.getTaskCpus.get == BigDecimal(amount.toString), s"for amount $amount")
+ }
+ }
+
+ test("SPARK-58192: cpus amounts that round to zero at the accounting scale are rejected") {
+ Seq(0.0, -1.0, 1e-10, Double.NaN,
+ Double.PositiveInfinity, Double.NegativeInfinity).foreach { amount =>
+ val e = intercept[IllegalArgumentException] {
+ new TaskResourceRequests().cpus(amount)
+ }
+ assert(e.getMessage.contains("must be at least 1e-9"), s"for amount $amount")
+ }
+ // Above the maximum: no executor can have more than Int.MaxValue cores, so a larger
+ // request could never be satisfied anywhere. The maximum itself stays valid.
+ Seq(2147483647.5, 1e100).foreach { amount =>
+ val e = intercept[IllegalArgumentException] {
+ new TaskResourceRequests().cpus(amount)
+ }
+ assert(e.getMessage.contains("at most"), s"for amount $amount")
+ }
+ assert(new TaskResourceRequests().cpus(Int.MaxValue.toDouble)
+ .requests(ResourceProfile.CPUS).amount === Int.MaxValue.toDouble)
+ // Every live entry point validates, not just cpus(): resource() and addRequest() route
+ // cpus amounts through the same check (spark.task.resource.cpus.amount flows through
+ // resource() as well), while valid amounts still land in the request map.
+ intercept[IllegalArgumentException] {
+ new TaskResourceRequests().resource(ResourceProfile.CPUS, 1e-10)
+ }
+ intercept[IllegalArgumentException] {
+ new TaskResourceRequests().addRequest(new TaskResourceRequest(ResourceProfile.CPUS, 0.0))
+ }
+ assert(new TaskResourceRequests().resource(ResourceProfile.CPUS, 0.5)
+ .requests(ResourceProfile.CPUS).amount === 0.5)
+ // The constructor itself stays fully lenient: it also runs when the history server
+ // deserializes persisted data (event logs, the protobuf KVStore), which can carry any
+ // amount an earlier release accepted -- the pre-4.3 check was `amount <= 1.0 || whole`,
+ // which admitted 0, negatives, and even -Infinity, and protobuf round-trips non-finite
+ // doubles unchanged.
+ assert(new TaskResourceRequest(ResourceProfile.CPUS, 0.0).amount === 0.0)
+ assert(new TaskResourceRequest(ResourceProfile.CPUS, Double.NegativeInfinity)
+ .amount === Double.NegativeInfinity)
+ }
+
+ test("SPARK-58192: numTasksBasedOnCores clamps to [0, Int.MaxValue]") {
+ def numTasks(cores: String, cpusPerTask: String): Int = {
+ ResourceProfile.numTasksBasedOnCores(
+ CpuAmount.normalize(BigDecimal(cores)), CpuAmount.normalize(BigDecimal(cpusPerTask)))
+ }
+ assert(numTasks("4", "1.5") === 2)
+ assert(numTasks("1", "0.1") === 10)
+ // 4 / 1e-9 == 4e9 overflows Int; BigDecimal.intValue() would wrap it to a negative value
+ assert(numTasks("4", "0.000000001") === Int.MaxValue)
+ assert(numTasks("-1", "1") === 0)
+ }
+
+ test("SPARK-58192: numMemoryShareSlotsCeil rounds up and clamps to [0, Int.MaxValue]") {
+ def numSlots(cores: String, cpusPerTask: String): Int = {
+ ResourceProfile.numMemoryShareSlotsCeil(
+ CpuAmount.normalize(BigDecimal(cores)), CpuAmount.normalize(BigDecimal(cpusPerTask)))
+ }
+ // Exact quotients match the floor variant ...
+ assert(numSlots("4", "1") === 4)
+ assert(numSlots("4", "2") === 2)
+ assert(numSlots("1", "0.1") === 10)
+ // ... while non-dividing cpus round up, so a memory budget divided by the result never
+ // exceeds the cpus-proportional share: budget / ceil(4 / 3) is at most budget * 3 / 4.
+ assert(numSlots("4", "3") === 2)
+ assert(numSlots("4", "1.5") === 3)
+ assert(numSlots("4", "0.000000001") === Int.MaxValue)
+ assert(numSlots("-1", "1") === 0)
+ }
+
test("SPARK-45527 fractional TaskResourceRequests in TaskResourceProfile") {
var treqs = new TaskResourceRequests().cpus(1).resource("gpu", 0.1)
new ResourceProfileBuilder().require(treqs).build()
diff --git a/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
index 127e4b2e413c..99b757496379 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
@@ -81,7 +81,7 @@ class CoarseGrainedSchedulerBackendSuite extends SparkFunSuite with LocalSparkCo
test("compute max number of concurrent tasks can be launched when spark.task.cpus > 1") {
val conf = new SparkConf()
- .set(CPUS_PER_TASK, 2)
+ .set(CPUS_PER_TASK, BigDecimal(2))
.setMaster("local-cluster[4, 3, 1024]")
.setAppName("test")
sc = new SparkContext(conf)
@@ -95,7 +95,7 @@ class CoarseGrainedSchedulerBackendSuite extends SparkFunSuite with LocalSparkCo
test("compute max number of concurrent tasks can be launched when some executors are busy") {
val conf = new SparkConf()
- .set(CPUS_PER_TASK, 2)
+ .set(CPUS_PER_TASK, BigDecimal(2))
.setMaster("local-cluster[4, 3, 1024]")
.setAppName("test")
sc = new SparkContext(conf)
@@ -143,7 +143,7 @@ class CoarseGrainedSchedulerBackendSuite extends SparkFunSuite with LocalSparkCo
val discoveryScript = createTempScriptWithExpectedOutput(
dir, "gpuDiscoveryScript", """{"name": "gpu","addresses":["0", "1", "2", "3"]}""")
val conf = new SparkConf()
- .set(CPUS_PER_TASK, 1)
+ .set(CPUS_PER_TASK, BigDecimal(1))
.setMaster("local-cluster[1, 20, 1024]")
.setAppName("test")
.set(WORKER_GPU_ID.amountConf, "4")
@@ -199,7 +199,7 @@ class CoarseGrainedSchedulerBackendSuite extends SparkFunSuite with LocalSparkCo
val discoveryScript = createTempScriptWithExpectedOutput(
dir, "gpuDiscoveryScript", """{"name": "gpu","addresses":["0", "1", "2", "3"]}""")
val conf = new SparkConf()
- .set(CPUS_PER_TASK, taskCpus)
+ .set(CPUS_PER_TASK, BigDecimal(taskCpus))
.setMaster("local-cluster[1, 4, 1024]")
.setAppName("test")
.set(WORKER_GPU_ID.amountConf, "4")
@@ -216,7 +216,7 @@ class CoarseGrainedSchedulerBackendSuite extends SparkFunSuite with LocalSparkCo
val numPartitions = Seq(4 / taskCpus, 4 / taskGpus).min
val ret = sc.parallelize(1 to 20, numPartitions).mapPartitions { _ =>
val tc = TaskContext.get()
- assert(tc.cpus() == taskCpus)
+ assert(tc.cpuAmount() == taskCpus)
val gpus = tc.resources()("gpu").addresses
Iterator.single(gpus)
}.collect()
diff --git a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
index e12348e1be2d..fd6f32919fac 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
@@ -4422,6 +4422,199 @@ class DAGSchedulerSuite extends SparkFunSuite with TempLocalSparkContext with Ti
assert(rpMerged.id == expectedid.get)
}
+ test("SPARK-58192: per-stage resource-profile properties do not mutate the shared job " +
+ "Properties") {
+ // An explicit executor cores makes the default profile's max-concurrency known, so
+ // addPySparkConfigsToProperties writes the per-stage resource local properties.
+ conf.set(config.EXECUTOR_CORES.key, "4")
+ val jobProps = new Properties()
+ val rdd = sc.parallelize(1 to 10, 2).map(x => (x, x))
+ submit(rdd, Array(0, 1), properties = jobProps)
+
+ // The submitted task set carries the profile's max concurrent tasks on its own copy ...
+ val taskSetProps = taskSets.head.properties
+ assert(taskSetProps.getProperty(ResourceProfile.MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY) === "4")
+ // ... while the shared job Properties instance is left untouched, so sibling stages of the
+ // same job that use a different resource profile cannot clobber each other's value (e.g. a
+ // GPU-limited stage and a CPU-limited sibling).
+ assert(jobProps.getProperty(ResourceProfile.MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY) === null)
+ assert(jobProps.getProperty(ResourceProfile.EXECUTOR_CORES_LOCAL_PROPERTY) === null)
+
+ complete(taskSets.head, Seq((Success, 42), (Success, 42)))
+ assert(results === Map(0 -> 42, 1 -> 42))
+ }
+
+ test("SPARK-58192: custom-resource-limited max concurrent tasks is propagated even when the " +
+ "cores limit is unknown") {
+ // No explicit spark.executor.cores and a local master: the profile cannot use cores to bound
+ // concurrency (isCoresLimitKnown is false), but the single gpu proves at most one task runs
+ // at a time, which caps the pyspark memory split regardless of the unknown core count.
+ val ereqs = new ExecutorResourceRequests().cores(64).resource(GPU, 1, "disc")
+ .pysparkMemory("512m")
+ val treqs = new TaskResourceRequests().cpus(0.1).resource(GPU, 1)
+ val rp = new ResourceProfileBuilder().require(ereqs).require(treqs).build()
+
+ val rdd = sc.parallelize(1 to 10, 2).map(x => (x, x)).withResources(rp)
+ submit(rdd, Array(0, 1), properties = new Properties())
+
+ // The single gpu admits at most one concurrent task no matter the core count, so the
+ // worker memory split must use 1 slot, not the cpu-based floor(64 / 0.1) = 640.
+ val taskSetProps = taskSets.head.properties
+ assert(taskSetProps.getProperty(ResourceProfile.MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY) === "1")
+ assert(taskSetProps.getProperty(ResourceProfile.EXECUTOR_CORES_LOCAL_PROPERTY) === "64")
+ assert(taskSetProps.getProperty(ResourceProfile.PYSPARK_MEMORY_LOCAL_PROPERTY) === "512")
+
+ complete(taskSets.head, Seq((Success, 42), (Success, 42)))
+ assert(results === Map(0 -> 42, 1 -> 42))
+ }
+
+ test("SPARK-58192: task-only stage profile inherits the default profile's pyspark memory") {
+ conf.set(config.EXECUTOR_CORES.key, "4")
+ conf.set(config.Python.PYSPARK_EXECUTOR_MEMORY.key, "2g")
+ val shuffleMapRdd = new MyRDD(sc, 2, Nil)
+ val shuffleDep = new ShuffleDependency(shuffleMapRdd, new HashPartitioner(2))
+ val taskRp = new TaskResourceProfile(new TaskResourceRequests().cpus(2).requests)
+ val reduceRdd = new MyRDD(sc, 2, List(shuffleDep), tracker = mapOutputTracker)
+ .withResources(taskRp)
+ submit(reduceRdd, Array(0, 1), properties = new Properties())
+
+ // The default-profile shuffle stage carries the executor-wide pyspark memory (2g = 2048 MiB)
+ // and its cpu-slot concurrency floor(4 / 1) = 4.
+ val parentProps = taskSets(0).properties
+ assert(parentProps.getProperty(ResourceProfile.PYSPARK_MEMORY_LOCAL_PROPERTY) === "2048")
+ assert(parentProps.getProperty(ResourceProfile.MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY) === "4")
+
+ complete(taskSets(0), Seq(
+ (Success, makeMapStatus("hostA", 2)), (Success, makeMapStatus("hostB", 2))))
+
+ // A task-only profile declares no executor resources, so its tasks run on executors created
+ // from the default profile's executor requests: the child stage must inherit the default
+ // profile's pyspark memory rather than dropping the limit, split across its own
+ // floor(4 / 2) = 2 cpu slots.
+ val childProps = taskSets(1).properties
+ assert(childProps.getProperty(ResourceProfile.PYSPARK_MEMORY_LOCAL_PROPERTY) === "2048")
+ assert(childProps.getProperty(ResourceProfile.MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY) === "2")
+
+ complete(taskSets(1), Seq((Success, 42), (Success, 42)))
+ assert(results === Map(0 -> 42, 1 -> 42))
+ }
+
+ test("SPARK-58192: stage-local resource values are written when job properties are null") {
+ // JobSubmitted and MapStageSubmitted permit null job properties; the per-stage snapshot
+ // must materialize an empty map for them since the concurrency bound below is always
+ // written for the default profile.
+ val rdd = sc.parallelize(1 to 10, 2).map(x => (x, x))
+ submit(rdd, Array(0, 1))
+
+ val taskSetProps = taskSets.head.properties
+ assert(taskSetProps !== null)
+ assert(taskSetProps.getProperty(ResourceProfile.MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY) === "1")
+
+ complete(taskSets.head, Seq((Success, 42), (Success, 42)))
+ assert(results === Map(0 -> 42, 1 -> 42))
+ }
+
+ test("SPARK-58192: a full resource profile without pyspark memory inherits the default " +
+ "profile's allocation") {
+ conf.set(config.EXECUTOR_CORES.key, "4")
+ conf.set(config.Python.PYSPARK_EXECUTOR_MEMORY.key, "2g")
+ val ereqs = new ExecutorResourceRequests().cores(8).memory("4g")
+ val treqs = new TaskResourceRequests().cpus(1)
+ val rp = new ResourceProfileBuilder().require(ereqs).require(treqs).build()
+ val rdd = sc.parallelize(1 to 10, 2).map(x => (x, x)).withResources(rp)
+ submit(rdd, Array(0, 1), properties = new Properties())
+
+ // Executors for this profile are provisioned from the default allocation for everything
+ // the profile does not specify -- including the 2g pyspark memory -- so the stage must
+ // carry that value; its slots come from the profile's own 8 cores.
+ val taskSetProps = taskSets.head.properties
+ assert(taskSetProps.getProperty(ResourceProfile.PYSPARK_MEMORY_LOCAL_PROPERTY) === "2048")
+ assert(taskSetProps.getProperty(ResourceProfile.MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY) === "8")
+
+ complete(taskSets.head, Seq((Success, 42), (Success, 42)))
+ assert(results === Map(0 -> 42, 1 -> 42))
+ }
+
+ test("SPARK-58192: custom-resource concurrency bound is capped by explicit profile executor " +
+ "cores when the cores limit is unknown") {
+ // No spark.executor.cores in the conf and a local master, so isCoresLimitKnown is false.
+ // The gpu bound alone admits (8 / 0.01) = 800 concurrent tasks, but the profile's explicit
+ // 4 cores admit only floor(4 / 1) = 4; the memory split must use the tighter bound, since
+ // dividing the budget 800 ways would shrink every worker's share 200x (or round it to
+ // zero, tripping the unsatisfiable-share fail-fast).
+ val ereqs = new ExecutorResourceRequests().cores(4).resource(GPU, 8, "disc")
+ val treqs = new TaskResourceRequests().cpus(1).resource(GPU, 0.01)
+ val rp = new ResourceProfileBuilder().require(ereqs).require(treqs).build()
+ val rdd = sc.parallelize(1 to 10, 2).map(x => (x, x)).withResources(rp)
+ submit(rdd, Array(0, 1), properties = new Properties())
+
+ val taskSetProps = taskSets.head.properties
+ assert(taskSetProps.getProperty(ResourceProfile.MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY) === "4")
+
+ complete(taskSets.head, Seq((Success, 42), (Success, 42)))
+ assert(results === Map(0 -> 42, 1 -> 42))
+ }
+
+ test("SPARK-58192: shareable executors propagate the cpus-proportional bound, not the " +
+ "custom-resource bound") {
+ conf.set(config.EXECUTOR_CORES.key, "2")
+ conf.set("spark.executor.resource.gpu.amount", "1")
+ conf.set("spark.task.resource.gpu.amount", "1")
+ val shuffleMapRdd = new MyRDD(sc, 2, Nil)
+ val shuffleDep = new ShuffleDependency(shuffleMapRdd, new HashPartitioner(2))
+ val taskRp = new TaskResourceProfile(new TaskResourceRequests().cpus(1).requests)
+ val reduceRdd = new MyRDD(sc, 2, List(shuffleDep), tracker = mapOutputTracker)
+ .withResources(taskRp)
+ submit(reduceRdd, Array(0, 1), properties = new Properties())
+
+ // With dynamic allocation off, default-profile and task-only-profile tasks can run on one
+ // executor concurrently, so both stages must get the cpus-proportional floor(2 / 1) = 2:
+ // shares of the executor-wide pyspark memory budget then sum within it for any
+ // core-feasible mix. The gpu-limited bound of 1 would hand every default-profile worker
+ // the whole budget while task-only-profile workers hold their own shares of it.
+ val parentProps = taskSets(0).properties
+ assert(parentProps.getProperty(ResourceProfile.MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY) === "2")
+
+ complete(taskSets(0), Seq(
+ (Success, makeMapStatus("hostA", 2)), (Success, makeMapStatus("hostB", 2))))
+
+ val childProps = taskSets(1).properties
+ assert(childProps.getProperty(ResourceProfile.MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY) === "2")
+
+ complete(taskSets(1), Seq((Success, 42), (Success, 42)))
+ assert(results === Map(0 -> 42, 1 -> 42))
+ }
+
+ test("SPARK-58192: shareable-executor memory shares use ceiling division when task cpus " +
+ "do not evenly divide the cores") {
+ conf.set(config.EXECUTOR_CORES.key, "4")
+ conf.set(config.Python.PYSPARK_EXECUTOR_MEMORY.key, "2g")
+ val shuffleMapRdd = new MyRDD(sc, 2, Nil)
+ val shuffleDep = new ShuffleDependency(shuffleMapRdd, new HashPartitioner(2))
+ val taskRp = new TaskResourceProfile(new TaskResourceRequests().cpus(3).requests)
+ val reduceRdd = new MyRDD(sc, 2, List(shuffleDep), tracker = mapOutputTracker)
+ .withResources(taskRp)
+ submit(reduceRdd, Array(0, 1), properties = new Properties())
+
+ // A 3-cpu task and a 1-cpu default task fit on one 4-core executor concurrently, so their
+ // memory shares must sum within the 2048 MiB budget: the default stage divides by 4
+ // (512 MiB each) and the task-only stage by ceil(4 / 3) = 2 (1024 MiB), for at most
+ // 1536 MiB. Dividing by floor(4 / 3) = 1 would hand the 3-cpu worker the whole 2048 and
+ // let the pair reach 2560.
+ val parentProps = taskSets(0).properties
+ assert(parentProps.getProperty(ResourceProfile.MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY) === "4")
+
+ complete(taskSets(0), Seq(
+ (Success, makeMapStatus("hostA", 2)), (Success, makeMapStatus("hostB", 2))))
+
+ val childProps = taskSets(1).properties
+ assert(childProps.getProperty(ResourceProfile.PYSPARK_MEMORY_LOCAL_PROPERTY) === "2048")
+ assert(childProps.getProperty(ResourceProfile.MAX_TASKS_PER_EXECUTOR_LOCAL_PROPERTY) === "2")
+
+ complete(taskSets(1), Seq((Success, 42), (Success, 42)))
+ assert(results === Map(0 -> 42, 1 -> 42))
+ }
+
test("test 2 resource profiles errors by default") {
import org.apache.spark.resource._
val ereqs = new ExecutorResourceRequests().cores(4)
diff --git a/core/src/test/scala/org/apache/spark/scheduler/SchedulerIntegrationSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/SchedulerIntegrationSuite.scala
index 2cde2cc76ba3..66d9ad4dd686 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/SchedulerIntegrationSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/SchedulerIntegrationSuite.scala
@@ -378,7 +378,7 @@ private[spark] abstract class MockBackend(
val env = SparkEnv.get
/** Accessed by both scheduling and backend thread, so should be protected by this. */
- var freeCores: Int = _
+ var freeCores: BigDecimal = BigDecimal(0)
/**
* Accessed by both scheduling and backend thread, so should be protected by this.
@@ -456,7 +456,7 @@ private[spark] class SingleCoreMockBackend(
)
}
-case class ExecutorTaskStatus(host: String, executorId: String, var freeCores: Int)
+case class ExecutorTaskStatus(host: String, executorId: String, var freeCores: BigDecimal)
class MockRDD(
sc: SparkContext,
diff --git a/core/src/test/scala/org/apache/spark/scheduler/TaskDescriptionSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/TaskDescriptionSuite.scala
index 5aaacd2ec1c3..dcb22f8b0d7d 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/TaskDescriptionSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/TaskDescriptionSuite.scala
@@ -24,7 +24,7 @@ import java.util.Properties
import scala.collection.mutable.HashMap
import org.apache.spark.{JobArtifactSet, SparkFunSuite}
-import org.apache.spark.resource.ResourceAmountUtils
+import org.apache.spark.resource.{CpuAmount, ResourceAmountUtils}
import org.apache.spark.resource.ResourceUtils.GPU
class TaskDescriptionSuite extends SparkFunSuite {
@@ -105,4 +105,28 @@ class TaskDescriptionSuite extends SparkFunSuite {
assert(decodedTaskDescription.serializedTask.equals(taskBuffer))
}
+ test("SPARK-58192: fractional cpus survives the encode/decode round trip exactly") {
+ Seq("0.5", "1.5", "0.000000001", "2").foreach { amount =>
+ val cpus = CpuAmount.normalize(BigDecimal(amount))
+ val taskDescription = new TaskDescription(
+ taskId = 42,
+ attemptNumber = 0,
+ executorId = "testExecutor",
+ name = "task for test",
+ index = 1,
+ partitionId = 1,
+ new JobArtifactSet(None, Map.empty, Map.empty, Map.empty),
+ new Properties(),
+ cpus,
+ Map.empty,
+ ByteBuffer.wrap(Array[Byte](1, 2, 3, 4)))
+ val decoded = TaskDescription.decode(TaskDescription.encode(taskDescription))
+ // The decoded amount is what the executor echoes back in its status update and what the
+ // driver adds back to the executor's free cores, so it must be the identical normalized
+ // value -- same scale included -- not merely numerically close.
+ assert(decoded.cpus === cpus, s"for amount $amount")
+ assert(decoded.cpus.scale === cpus.scale, s"scale for amount $amount")
+ }
+ }
+
}
diff --git a/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala
index c9c9e529405e..377c8afe134d 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala
@@ -35,7 +35,7 @@ import org.scalatestplus.mockito.MockitoSugar
import org.apache.spark._
import org.apache.spark.internal.config
-import org.apache.spark.resource.{ExecutorResourceRequests, ResourceAmountUtils, ResourceProfile, TaskResourceProfile, TaskResourceRequests}
+import org.apache.spark.resource.{CpuAmount, ExecutorResourceRequests, ResourceAmountUtils, ResourceProfile, TaskResourceProfile, TaskResourceRequests}
import org.apache.spark.resource.ResourceAmountUtils.ONE_ENTIRE_RESOURCE
import org.apache.spark.resource.ResourceUtils._
import org.apache.spark.resource.TestResourceIDs._
@@ -240,6 +240,62 @@ class TaskSchedulerImplSuite extends SparkFunSuite with LocalSparkContext
assert(!failedTaskSet)
}
+ test("SPARK-58192: Scheduler correctly accounts for fractional CPUs per task") {
+ // With spark.task.cpus = 0.5, a single core can run 2 tasks concurrently.
+ val taskScheduler = setupSchedulerWithMaster(
+ "local[1]",
+ config.CPUS_PER_TASK.key -> "0.5")
+ val singleCoreOffers =
+ IndexedSeq(WorkerOffer("executor0", "host0", 1))
+ taskScheduler.submitTasks(FakeTask.createTaskSet(5))
+ val taskDescriptions = taskScheduler.resourceOffers(singleCoreOffers).flatten
+ assert(2 === taskDescriptions.length)
+ assert(taskDescriptions.forall(_.executorId == "executor0"))
+ assert(taskDescriptions.forall(td => td.cpus.toDouble == 0.5))
+ assert(!failedTaskSet)
+ }
+
+ test("SPARK-58192: fractional CPUs (0.2) schedule the expected number of tasks") {
+ // 1.0 / 0.2 would floor to 4 with naive Double arithmetic; BigDecimal computes it exactly
+ // as 5, so all 5 task slots on a single core are used.
+ val taskScheduler = setupSchedulerWithMaster(
+ "local[1]",
+ config.CPUS_PER_TASK.key -> "0.2")
+ val singleCoreOffers =
+ IndexedSeq(WorkerOffer("executor0", "host0", 1))
+ taskScheduler.submitTasks(FakeTask.createTaskSet(10))
+ val taskDescriptions = taskScheduler.resourceOffers(singleCoreOffers).flatten
+ assert(5 === taskDescriptions.length)
+ assert(!failedTaskSet)
+ }
+
+ test("SPARK-58192: fractional CPUs above 1 (1.5) schedule the expected number of tasks") {
+ val taskScheduler = setupSchedulerWithMaster(
+ "local[2]",
+ config.CPUS_PER_TASK.key -> "1.5")
+ val offers = IndexedSeq(WorkerOffer("executor0", "host0", 4))
+ taskScheduler.submitTasks(FakeTask.createTaskSet(5))
+ val taskDescriptions = taskScheduler.resourceOffers(offers).flatten
+ // floor(4 / 1.5) == 2 tasks fit on a 4-core executor
+ assert(2 === taskDescriptions.length)
+ assert(taskDescriptions.forall(td => td.cpus.toDouble == 1.5))
+ assert(!failedTaskSet)
+ }
+
+ test("SPARK-58192: calculateAvailableSlots saturates instead of overflowing Int") {
+ val taskScheduler = setupSchedulerWithMaster(
+ "local[1]",
+ config.CPUS_PER_TASK.key -> "0.000000001")
+ // Each 4-core executor contributes floor(4 / 1e-9) clamped to Int.MaxValue; the sum over
+ // two executors must saturate at Int.MaxValue instead of wrapping negative.
+ val rpId = ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID
+ val fourCores = CpuAmount.normalize(BigDecimal(4))
+ val slots = TaskSchedulerImpl.calculateAvailableSlots(taskScheduler, sc.conf,
+ rpId, Array(rpId, rpId), Array(fourCores, fourCores),
+ Array(Map.empty[String, Int], Map.empty[String, Int]))
+ assert(slots === Int.MaxValue)
+ }
+
private def setupTaskSchedulerForLocalityTests(
clock: ManualClock,
conf: SparkConf = new SparkConf()): TaskSchedulerImpl = {
@@ -1125,7 +1181,7 @@ class TaskSchedulerImplSuite extends SparkFunSuite with LocalSparkContext
// as we offer resources to the taskset -- each iteration either schedules
// something, or it terminates that locality level, so the maximum number of
// checks is numCores + numLocalityLevels
- val numCoresOnAllOffers = offers.map(_.cores).sum
+ val numCoresOnAllOffers = offers.map(_.cores).sum.toInt
val numLocalityLevels = TaskLocality.values.size
val maxExcludelistChecks = numCoresOnAllOffers + numLocalityLevels
diff --git a/core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleReaderSuite.scala b/core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleReaderSuite.scala
index b3b09156fe0c..2d8820ab6181 100644
--- a/core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleReaderSuite.scala
+++ b/core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleReaderSuite.scala
@@ -62,7 +62,7 @@ class StreamingShuffleReaderSuite
taskMemoryManager = taskMemoryManager,
localProperties = properties,
metricsSystem = mock[MetricsSystem],
- cpus = 1)
+ cpuAmount = 1)
}
// A minimal DataMessage that is only used to exercise the iterator factory's type dispatch;
diff --git a/core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleSuite.scala b/core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleSuite.scala
index af8e074fb9f1..578af94baa68 100644
--- a/core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleSuite.scala
+++ b/core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleSuite.scala
@@ -78,7 +78,7 @@ class StreamingShuffleSuite
taskMemoryManager = taskMemoryManager,
localProperties = properties,
metricsSystem = metricsSystem,
- cpus = 1
+ cpuAmount = 1
)
}
diff --git a/core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleWriterSuite.scala b/core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleWriterSuite.scala
index 602cdd0bfc48..e421272cc119 100644
--- a/core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleWriterSuite.scala
+++ b/core/src/test/scala/org/apache/spark/shuffle/streaming/StreamingShuffleWriterSuite.scala
@@ -68,7 +68,7 @@ class StreamingShuffleWriterSuite
taskMemoryManager = taskMemoryManager,
localProperties = properties,
metricsSystem = mock[MetricsSystem],
- cpus = 1)
+ cpuAmount = 1)
}
test("getWriter returns a StreamingShuffleWriter") {
diff --git a/core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala b/core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala
index 4d75f5d7a1fc..193c2a9f3349 100644
--- a/core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala
@@ -29,7 +29,7 @@ import org.apache.spark.executor.{ExecutorMetrics, TaskMetrics}
import org.apache.spark.internal.config.History.{HYBRID_STORE_DISK_BACKEND, HybridStoreDiskBackend}
import org.apache.spark.internal.config.Status._
import org.apache.spark.metrics.ExecutorMetricType
-import org.apache.spark.resource.ResourceProfile
+import org.apache.spark.resource.{ResourceProfile, TaskResourceRequest}
import org.apache.spark.scheduler._
import org.apache.spark.scheduler.cluster._
import org.apache.spark.status.ListenerEventsTestHelper._
@@ -107,6 +107,72 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
}
}
+ test("SPARK-58192: executor max tasks with fractional spark.task.cpus") {
+ val listener = new AppStatusListener(store, conf, true)
+ // The default cpus per task reaches the listener as the raw property string, which may be
+ // fractional; the executor's task capacity must use exact decimal slot math - with 4 cores
+ // and 0.5 cpus per task an executor runs 8 concurrent tasks, while integer parsing would
+ // reject the value and integer division would truncate the capacity.
+ listener.onEnvironmentUpdate(SparkListenerEnvironmentUpdate(Map(
+ "JVM Information" -> Seq.empty,
+ "Spark Properties" -> Seq("spark.task.cpus" -> "0.5"),
+ "Hadoop Properties" -> Seq.empty,
+ "System Properties" -> Seq.empty,
+ "Metrics Properties" -> Seq.empty,
+ "Classpath Entries" -> Seq.empty)))
+ listener.onExecutorAdded(SparkListenerExecutorAdded(1L, "1",
+ new ExecutorInfo("1.example.com", 4, Map.empty, Map.empty)))
+
+ check[ExecutorSummaryWrapper]("1") { exec =>
+ assert(exec.info.maxTasks === 8)
+ }
+ }
+
+ test("SPARK-58192: invalid spark.task.cpus in the environment falls back to the default") {
+ val listener = new AppStatusListener(store, conf, true)
+ // A corrupt or crafted event log can carry values the live config parser would have
+ // rejected. An extreme exponent must not materialize an enormous decimal inside the
+ // (shared) history server, and a malformed value must not fail the replay -- both keep
+ // the previous default instead.
+ // The million-digit value must be rejected by the length bound before it is parsed:
+ // decimal parsing cost grows superlinearly with the significand, and the value
+ // compresses to about a kilobyte inside an event log.
+ Seq("1e10000000", "1".repeat(1000000), "abc", "0", "-1").foreach { bad =>
+ listener.onEnvironmentUpdate(SparkListenerEnvironmentUpdate(Map(
+ "JVM Information" -> Seq.empty,
+ "Spark Properties" -> Seq("spark.task.cpus" -> bad),
+ "Hadoop Properties" -> Seq.empty,
+ "System Properties" -> Seq.empty,
+ "Metrics Properties" -> Seq.empty,
+ "Classpath Entries" -> Seq.empty)))
+ }
+ listener.onExecutorAdded(SparkListenerExecutorAdded(1L, "1",
+ new ExecutorInfo("1.example.com", 4, Map.empty, Map.empty)))
+
+ check[ExecutorSummaryWrapper]("1") { exec =>
+ assert(exec.info.maxTasks === 4)
+ }
+ }
+
+ test("SPARK-58192: replayed profile with a historically accepted cpus amount") {
+ val listener = new AppStatusListener(store, conf, true)
+ // Earlier releases accepted e.g. negative cpus amounts into persisted profiles, and the
+ // lenient TaskResourceRequest constructor keeps them deserializable; computing executor
+ // capacity from such a profile must fall back to the default rather than abort the
+ // replay on the slot math's positivity assertion.
+ val legacyProfile = new ResourceProfile(
+ Map.empty,
+ Map(ResourceProfile.CPUS -> new TaskResourceRequest(ResourceProfile.CPUS, -1.0)))
+ listener.onResourceProfileAdded(SparkListenerResourceProfileAdded(legacyProfile))
+ listener.onExecutorAdded(SparkListenerExecutorAdded(1L, "1",
+ new ExecutorInfo("1.example.com", 2, Map.empty, Map.empty, Map.empty,
+ legacyProfile.id)))
+
+ check[ExecutorSummaryWrapper]("1") { exec =>
+ assert(exec.info.maxTasks === 2)
+ }
+ }
+
test("scheduler events") {
val listener = new AppStatusListener(store, conf, true)
diff --git a/core/src/test/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializerSuite.scala b/core/src/test/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializerSuite.scala
index ac568fee1ad4..7bb36a97163b 100644
--- a/core/src/test/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/status/protobuf/KVStoreProtobufSerializerSuite.scala
@@ -316,7 +316,11 @@ class KVStoreProtobufSerializerSuite extends SparkFunSuite {
),
taskResources = Map(
"0" -> new TaskResourceRequest(resourceName = "exec1", amount = 1),
- "1" -> new TaskResourceRequest(resourceName = "exec2", amount = 1)
+ "1" -> new TaskResourceRequest(resourceName = "exec2", amount = 1),
+ // A pre-4.3 store can carry any cpus amount the old `<= 1.0 || whole` check
+ // accepted, including -Infinity; deserialization must restore it unchanged.
+ "cpus" -> new TaskResourceRequest(resourceName = "cpus",
+ amount = Double.NegativeInfinity)
)
))
)
diff --git a/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala b/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala
index 0dbd8d792137..f6d2298a2aba 100644
--- a/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala
+++ b/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala
@@ -203,6 +203,35 @@ class JsonProtocolSuite extends SparkFunSuite {
testEvent(resourceProfileAdded, resourceProfileJsonString)
}
+ test("SPARK-58192: resource profiles with historical zero cpus amounts deserialize") {
+ // Event logs written before cpus amounts were validated can carry a cpus amount of 0;
+ // deserialization must accept them so the history server can replay such applications.
+ val legacyJson =
+ """
+ |{
+ | "Event":"SparkListenerResourceProfileAdded",
+ | "Resource Profile Id":7,
+ | "Executor Resource Requests":{
+ | "cores":{
+ | "Resource Name":"cores",
+ | "Amount":2,
+ | "Discovery Script":"",
+ | "Vendor":""
+ | }
+ | },
+ | "Task Resource Requests":{
+ | "cpus":{
+ | "Resource Name":"cpus",
+ | "Amount":0.0
+ | }
+ | }
+ |}
+ """.stripMargin
+ val event = JsonProtocol.sparkEventFromJson(legacyJson)
+ .asInstanceOf[SparkListenerResourceProfileAdded]
+ assert(event.resourceProfile.taskResources("cpus").amount === 0.0)
+ }
+
test("Dependent Classes") {
val logUrlMap = Map("stderr" -> "mystderr", "stdout" -> "mystdout")
val attributes = Map("ContainerId" -> "ct1", "User" -> "spark")
diff --git a/docs/configuration.md b/docs/configuration.md
index 4999b08908de..ba66aba66128 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -3132,9 +3132,11 @@ Apart from these, the following properties are also available, and may be useful
slots on a single executor and the task is taking longer time than the threshold. This config
helps speculate stage with very few tasks. Regular speculation configs may also apply if the
executor slots are large enough. E.g. tasks might be re-launched if there are enough successful
- runs even though the threshold hasn't been reached. The number of slots is computed based on
- the conf values of spark.executor.cores and spark.task.cpus minimum 1.
- Default unit is bytes, unless otherwise specified.
+ runs even though the threshold hasn't been reached. The number of slots is the maximum
+ number of concurrent tasks per executor for the stage's resource profile, computed from
+ the executor cores and the task cpus amount (which may be fractional), or 1 when the
+ executor cores are not known.
+ Default unit is milliseconds, unless otherwise specified.
</td>
<td>3.0.0</td>
</tr>
@@ -3177,7 +3179,10 @@ Apart from these, the following properties are also available, and may be useful
<td><code>spark.task.cpus</code></td>
<td>1</td>
<td>
- Number of cores to allocate for each task.
+ Number of cores to allocate for each task. This can also be set to a fractional value,
+ either below 1 (e.g. <code>0.2</code>) to allow multiple tasks to share a CPU core, or
+ above 1 (e.g. <code>1.5</code>). In either case the number of tasks that can run
+ concurrently on an executor is <code>floor(executor cores / spark.task.cpus)</code>.
</td>
<td>0.5.0</td>
</tr>
diff --git a/docs/core-migration-guide.md b/docs/core-migration-guide.md
index f91abaa9ddca..cb4309968ebc 100644
--- a/docs/core-migration-guide.md
+++ b/docs/core-migration-guide.md
@@ -38,6 +38,10 @@ license: |
- Since Spark 4.3, the Spark Master REST API rejects a submission whose request body exceeds `spark.master.rest.maxRequestBodySize` (default `100m`) with HTTP 413. To allow larger request bodies, increase `spark.master.rest.maxRequestBodySize`.
+- Since Spark 4.3, `spark.task.cpus` accepts fractional values, and the executor-wide `spark.executor.pyspark.memory` allocation is split across the executor's concurrent task capacity instead of its raw core count. Each Python worker's memory limit can therefore change for existing configurations: with `spark.task.cpus` greater than 1 each worker receives a proportionally larger share, and with dynamic allocation enabled, when a custom resource (e.g. GPUs) limits concurrency the fewer c [...]
+
+- Since Spark 4.3, a positive `spark.executor.pyspark.memory` allocation that is too small to give each concurrent task slot at least 1 MiB fails the Python task with an error instead of silently running the workers without any memory limit. Setting `spark.executor.pyspark.memory=0` still disables the limit. To restore a working memory limit, increase `spark.executor.pyspark.memory` or reduce the executor's concurrent task capacity.
+
## Upgrading from Core 4.1 to 4.2
- Since Spark 4.2, Spark Master REST API uses Java 21 virtual threads by default when running on Java 21 or later. To restore the legacy behavior, you can set `spark.master.rest.virtualThread.enabled` to `false`.
diff --git a/docs/running-on-kubernetes.md b/docs/running-on-kubernetes.md
index e2b88fd9f2a4..855bbf9c3db0 100644
--- a/docs/running-on-kubernetes.md
+++ b/docs/running-on-kubernetes.md
@@ -731,6 +731,11 @@ See the [configuration page](configuration.html) for information on Spark config
In other words, the recovery-mode executors replace the OOM-terminated executors to
survive from the resource-hungry tasks for the remaining tasks and stages.
If set to <code>false</code>, Spark will not use the recovery-mode executors.
+ Recovery-mode executors always derive their announced cores from the global
+ <code>spark.task.cpus</code>, not from stage-level resource profiles. Note that when
+ <code>spark.task.cpus</code> is 0.5 or less, a recovery-mode executor announces a single
+ CPU core and therefore accepts <code>floor(1 / spark.task.cpus)</code> concurrent tasks
+ instead of only one.
</td>
<td>4.2.0</td>
</tr>
diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala
index 643e4dca984e..984ce2f37ce5 100644
--- a/project/MimaExcludes.scala
+++ b/project/MimaExcludes.scala
@@ -35,6 +35,16 @@ object MimaExcludes {
// Exclude rules for 4.3.x from 4.2.0 (add 4.3-specific filters below as needed).
lazy val v43excludes: Seq[Problem => Boolean] = v42excludes ++ Seq(
+ // [SPARK-58192][CORE] Support fractional spark.task.cpus. The existing TaskContext.cpus(): Int
+ // and TaskResourceRequests.cpus(Int) are retained (cpus() deprecated in favor of the new
+ // TaskContext.cpuAmount(): BigDecimal), but the internal StatusUpdate message's taskCpus field
+ // is now a BigDecimal.
+ ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.TaskContext.cpuAmount"),
+ ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages#StatusUpdate.taskCpus"),
+ ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages#StatusUpdate.copy$default$5"),
+ ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages#StatusUpdate.copy"),
+ ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages#StatusUpdate.this"),
+ ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages#StatusUpdate.apply"),
// [SPARK-57910] ALS reuses the shared HasIntermediateStorageLevel param; the param and its
// getter are declared final in the shared trait (same name, type, default and validation).
ProblemFilters.exclude[FinalMethodProblem](
diff --git a/python/docs/source/reference/pyspark.rst b/python/docs/source/reference/pyspark.rst
index 798cda64d4d0..21a8f6dee99c 100644
--- a/python/docs/source/reference/pyspark.rst
+++ b/python/docs/source/reference/pyspark.rst
@@ -274,6 +274,7 @@ Management
StorageLevel.MEMORY_ONLY_2
StorageLevel.OFF_HEAP
TaskContext.attemptNumber
+ TaskContext.cpuAmount
TaskContext.cpus
TaskContext.get
TaskContext.getLocalProperty
@@ -286,6 +287,7 @@ Management
BarrierTaskContext.allGather
BarrierTaskContext.attemptNumber
BarrierTaskContext.barrier
+ BarrierTaskContext.cpuAmount
BarrierTaskContext.cpus
BarrierTaskContext.get
BarrierTaskContext.getLocalProperty
diff --git a/python/pyspark/resource/requests.py b/python/pyspark/resource/requests.py
index 324a5af361fc..999faf6b8314 100644
--- a/python/pyspark/resource/requests.py
+++ b/python/pyspark/resource/requests.py
@@ -375,8 +375,11 @@ class TaskResourceRequest:
A task resource request. This is used in conjunction with the
:class:`pyspark.resource.ResourceProfile` to programmatically specify the resources
needed for an RDD that will be applied at the stage level. The amount is specified
- as a float to allow for saying you want more than 1 task per resource. Valid values
- are less than or equal to 0.5 or whole numbers.
+ as a float to allow for saying you want more than 1 task per resource. For custom
+ resources, valid values are less than or equal to 0.5 or whole numbers, since a task's
+ amount must map onto discrete resource addresses. CPUs (resource name "cpus") are a
+ plain quantity drawn from the executor's core pool rather than an addressable
+ resource, so any positive amount is valid, e.g. 1.5.
Use :class:`pyspark.resource.TaskResourceRequests` class as a convenience API.
Parameters
@@ -385,9 +388,11 @@ class TaskResourceRequest:
Name of the resource
amount : float
Amount requesting as a float to support fractional resource requests.
- Valid values are less than or equal to 0.5 or whole numbers. This essentially
- lets you configure X number of tasks to run on a single resource,
- ie amount equals 0.5 translates into 2 tasks per resource address.
+ For custom resources, valid values are less than or equal to 0.5 or whole
+ numbers; this essentially lets you configure X number of tasks to run on a
+ single resource, ie amount equals 0.5 translates into 2 tasks per resource
+ address. For cpus, any positive amount is valid, including fractional ones
+ below 1 (e.g. 0.5) or above 1 (e.g. 1.5).
.. versionadded:: 3.1.0
@@ -474,29 +479,36 @@ class TaskResourceRequests:
if _requests is not None:
for k, v in _requests.items():
if k == self._CPUS:
- self._java_task_resource_requests.cpus(int(v.amount))
+ # Pass the amount as a float so the fractional part survives; py4j
+ # then selects the Double overload of TaskResourceRequests.cpus.
+ self._java_task_resource_requests.cpus(float(v.amount))
else:
self._java_task_resource_requests.resource(v.resourceName, v.amount)
else:
self._java_task_resource_requests = None
self._task_resources: Dict[str, TaskResourceRequest] = {}
- def cpus(self, amount: int) -> "TaskResourceRequests":
+ def cpus(self, amount: float) -> "TaskResourceRequests":
"""
Specify number of cpus per Task.
This is a convenient API to add :class:`TaskResourceRequest` for cpus.
+ .. versionchanged:: 4.3.0
+ Accepts fractional amounts.
+
Parameters
----------
- amount : int
- Number of cpus to allocate per Task.
+ amount : float
+ Number of cpus to allocate per Task, valid from 1e-9 to 2147483647 after
+ rounding. Fractional values are supported: below 1 (e.g. 0.5) to let multiple
+ tasks share a CPU core, or above 1 (e.g. 1.5).
Returns
-------
:class:`TaskResourceRequests`
"""
if self._java_task_resource_requests is not None:
- self._java_task_resource_requests.cpus(amount)
+ self._java_task_resource_requests.cpus(float(amount))
else:
self._task_resources[self._CPUS] = TaskResourceRequest(self._CPUS, amount)
return self
diff --git a/python/pyspark/resource/tests/test_resources.py b/python/pyspark/resource/tests/test_resources.py
index e62a56d8d475..8e9b304774ac 100644
--- a/python/pyspark/resource/tests/test_resources.py
+++ b/python/pyspark/resource/tests/test_resources.py
@@ -26,6 +26,12 @@ from pyspark.testing.utils import (
class ResourceProfileTests(unittest.TestCase):
+ def test_fractional_cpus_request(self):
+ # SPARK-58192: fractional cpus amounts are preserved instead of being truncated to int
+ for amount in [0.5, 1.5]:
+ treqs = TaskResourceRequests().cpus(amount)
+ self.assertEqual(treqs.requests["cpus"].amount, amount)
+
def test_profile_before_sc(self):
rpb = ResourceProfileBuilder()
ereqs = ExecutorResourceRequests().cores(2).memory("6g").memoryOverhead("1g")
diff --git a/python/pyspark/taskcontext.py b/python/pyspark/taskcontext.py
index 50cba601321b..796e3b5ef5f6 100644
--- a/python/pyspark/taskcontext.py
+++ b/python/pyspark/taskcontext.py
@@ -128,6 +128,7 @@ class TaskContext:
_taskAttemptId: Optional[int] = None
_localProperties: Optional[Dict[str, str]] = None
_cpus: Optional[int] = None
+ _cpuAmount: Optional[float] = None
_resources: Optional[Dict[str, "ResourceInformation"]] = None
def __new__(cls: Type["TaskContext"], **kwargs: Any) -> "TaskContext":
@@ -191,6 +192,7 @@ class TaskContext:
attemptNumber=json["attemptNumber"],
taskAttemptId=json["taskAttemptId"],
cpus=json["cpus"],
+ cpuAmount=float(json["cpuAmount"]) if "cpuAmount" in json else None,
resources={
k: ResourceInformation(v["name"], v["addresses"])
for k, v in json["resources"].items()
@@ -263,7 +265,9 @@ class TaskContext:
def cpus(self) -> int:
"""
- CPUs allocated to the task.
+ CPUs allocated to the task, rounded up to a whole number when the exact
+ allocation is fractional. Use :meth:`TaskContext.cpuAmount` to get the exact,
+ possibly fractional, amount.
Returns
-------
@@ -272,6 +276,25 @@ class TaskContext:
"""
return cast(int, self._cpus)
+ def cpuAmount(self) -> float:
+ """
+ The exact amount of CPUs allocated to the task. This can be fractional when
+ ``spark.task.cpus`` or the task resource profile requests a fractional amount
+ (e.g. 0.5).
+
+ .. versionadded:: 4.3.0
+
+ Returns
+ -------
+ float
+ the exact, possibly fractional, amount of CPUs.
+
+ See Also
+ --------
+ TaskContext.cpus
+ """
+ return cast(float, self._cpuAmount)
+
def resources(self) -> Dict[str, "ResourceInformation"]:
"""
Resources allocated to the task. The key is the resource name and the value is information
diff --git a/python/pyspark/tests/test_taskcontext.py b/python/pyspark/tests/test_taskcontext.py
index 13f3e111bd25..2f39ff53bc9e 100644
--- a/python/pyspark/tests/test_taskcontext.py
+++ b/python/pyspark/tests/test_taskcontext.py
@@ -351,6 +351,12 @@ class TaskContextTestsWithResources(unittest.TestCase):
cpus = rdd.map(lambda x: TaskContext.get().cpus()).take(1)[0]
self.assertEqual(cpus, 2)
+ def test_cpu_amount(self):
+ """SPARK-58192: the exact cpu amount is available."""
+ rdd = self.sc.parallelize(range(10))
+ cpu_amount = rdd.map(lambda x: TaskContext.get().cpuAmount()).take(1)[0]
+ self.assertEqual(cpu_amount, 2.0)
+
def test_resources(self):
"""Test that multiple resources are all available (SPARK-54929)."""
rdd = self.sc.parallelize(range(10))
@@ -369,6 +375,41 @@ class TaskContextTestsWithResources(unittest.TestCase):
self.sc.stop()
+class TaskContextTestsWithFractionalCpus(unittest.TestCase):
+ def setUp(self):
+ class_name = self.__class__.__name__
+ conf = SparkConf().set("spark.test.home", SPARK_HOME)
+ conf = conf.set("spark.task.cpus", "0.5")
+ self.sc = SparkContext("local-cluster[1,2,1024]", class_name, conf=conf)
+
+ def test_fractional_cpu_amount(self):
+ """SPARK-58192: cpuAmount() exposes the exact fractional amount; cpus() rounds up."""
+ rdd = self.sc.parallelize(range(4), 2)
+ cpu_amount, cpus = rdd.map(
+ lambda x: (TaskContext.get().cpuAmount(), TaskContext.get().cpus())
+ ).take(1)[0]
+ self.assertEqual(cpu_amount, 0.5)
+ self.assertEqual(cpus, 1)
+
+ def test_omp_num_threads_follows_task_cpus(self):
+ """SPARK-58192: OMP_NUM_THREADS defaults to the ceiling of the task's own cpu
+ amount, honoring stage-level resource profiles over the global spark.task.cpus."""
+ from pyspark.resource import ResourceProfileBuilder, TaskResourceRequests
+
+ rdd = self.sc.parallelize(range(2), 1)
+ omp = rdd.map(lambda _: os.environ["OMP_NUM_THREADS"]).take(1)[0]
+ self.assertEqual(omp, "1")
+
+ treqs = TaskResourceRequests().cpus(2)
+ rp = ResourceProfileBuilder().require(treqs).build
+ rdd = self.sc.parallelize(range(2), 1).withResources(rp)
+ omp = rdd.map(lambda _: os.environ["OMP_NUM_THREADS"]).take(1)[0]
+ self.assertEqual(omp, "2")
+
+ def tearDown(self):
+ self.sc.stop()
+
+
if __name__ == "__main__":
from pyspark.testing import main
diff --git a/python/pyspark/tests/test_worker.py b/python/pyspark/tests/test_worker.py
index b9ba8223a1cd..14a40dc23389 100644
--- a/python/pyspark/tests/test_worker.py
+++ b/python/pyspark/tests/test_worker.py
@@ -223,6 +223,34 @@ class WorkerMemoryTest(unittest.TestCase):
self.sc.stop()
+@unittest.skipIf(
+ not has_resource_module or sys.platform != "linux",
+ "Memory limit feature in Python worker is dependent on "
+ "Python's 'resource' module on Linux; however, not found or not on Linux.",
+)
+class WorkerMemoryFractionalTaskCpusTest(unittest.TestCase):
+ def setUp(self):
+ class_name = self.__class__.__name__
+ conf = SparkConf().set("spark.executor.pyspark.memory", "2g")
+ conf = conf.set("spark.task.cpus", "0.5")
+ self.sc = SparkContext("local[4]", class_name, conf=conf)
+
+ def test_memory_limit_split_across_fractional_task_slots(self):
+ # SPARK-58192: with spark.task.cpus=0.5 each executor core fits two concurrent
+ # tasks, so each Python worker gets half of the executor-wide pyspark memory.
+ rdd = self.sc.parallelize(range(1), 1)
+
+ def getrlimit():
+ return resource.getrlimit(resource.RLIMIT_AS)
+
+ [(soft_limit, hard_limit)] = rdd.map(lambda _: getrlimit()).collect()
+ self.assertEqual(soft_limit, 1024 * 1024 * 1024)
+ self.assertEqual(hard_limit, 1024 * 1024 * 1024)
+
+ def tearDown(self):
+ self.sc.stop()
+
+
class WorkerSegfaultTest(ReusedPySparkTestCase):
@classmethod
def conf(cls):
diff --git a/python/pyspark/worker_message.py b/python/pyspark/worker_message.py
index c037e728088b..5be806882c97 100644
--- a/python/pyspark/worker_message.py
+++ b/python/pyspark/worker_message.py
@@ -43,6 +43,7 @@ class TaskContextInfo:
attempt_number: int
task_attempt_id: int
cpus: int
+ cpu_amount: float
resources: dict[str, ResourceInfo]
local_properties: dict[str, str]
@@ -58,6 +59,7 @@ class TaskContextInfo:
attempt_number=task_context_json["attemptNumber"],
task_attempt_id=task_context_json["taskAttemptId"],
cpus=task_context_json["cpus"],
+ cpu_amount=float(task_context_json["cpuAmount"]),
resources={
k: cls.ResourceInfo(name=v["name"], addresses=v["addresses"])
for k, v in task_context_json["resources"].items()
@@ -75,6 +77,7 @@ class TaskContextInfo:
attemptNumber=self.attempt_number,
taskAttemptId=self.task_attempt_id,
cpus=self.cpus,
+ cpuAmount=self.cpu_amount,
resources={
k: ResourceInformation(v.name, v.addresses) for k, v in self.resources.items()
},
@@ -87,6 +90,7 @@ class TaskContextInfo:
attemptNumber=self.attempt_number,
taskAttemptId=self.task_attempt_id,
cpus=self.cpus,
+ cpuAmount=self.cpu_amount,
resources={
k: ResourceInformation(v.name, v.addresses) for k, v in self.resources.items()
},
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
index 00a02f3cba6f..7fd21dfdcb0f 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala
@@ -598,7 +598,10 @@ private[spark] object Config extends Logging {
"allocate the recovery-mode executors which accept only a single task per executor JVM. " +
"In other words, the recovery-mode executors replace the OOM-terminated executors to " +
"survive from the resource-hungry tasks for the remaining tasks and stages. " +
- "If set to `false`, Spark will not use the recovery-mode executors.")
+ "If set to `false`, Spark will not use the recovery-mode executors. " +
+ "Note that when spark.task.cpus is 0.5 or less, a recovery-mode executor announces a " +
+ "single CPU core and therefore accepts floor(1 / spark.task.cpus) concurrent tasks " +
+ "instead of only one.")
.version("4.2.0")
.booleanConf
.createOptional
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStep.scala
index 4209b2f513c2..9eb9303fa0dd 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStep.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStep.scala
@@ -168,7 +168,11 @@ private[spark] class BasicExecutorFeatureStep(
ENV_DRIVER_URL -> driverUrl,
ENV_EXECUTOR_CORES -> {
if (kubernetesConf.get(KUBERNETES_ALLOCATION_RECOVERY_MODE_ENABLED).getOrElse(false)) {
- kubernetesConf.get("spark.task.cpus", "1")
+ // `spark.task.cpus` may be fractional, so round up to an integer (at least 1).
+ // When it is 0.5 or less, the single announced core fits more than one task and
+ // the single-task-per-recovery-executor guarantee no longer holds;
+ // ExecutorPodsAllocator logs a warning for that case.
+ math.max(1, kubernetesConf.get("spark.task.cpus", "1.0").toDouble.ceil.toInt).toString
} else {
execResources.cores.get.toString
}
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala
index fb1206d6d102..2523cd64d3d7 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala
@@ -18,7 +18,7 @@ package org.apache.spark.scheduler.cluster.k8s
import java.time.Instant
import java.util.concurrent.{ConcurrentHashMap, TimeUnit}
-import java.util.concurrent.atomic.AtomicInteger
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
import scala.collection.mutable
import scala.jdk.CollectionConverters._
@@ -34,7 +34,7 @@ import org.apache.spark.deploy.k8s.KubernetesConf
import org.apache.spark.deploy.k8s.KubernetesUtils.addOwnerReference
import org.apache.spark.internal.{Logging, LogKeys}
import org.apache.spark.internal.config._
-import org.apache.spark.resource.ResourceProfile
+import org.apache.spark.resource.{CpuAmount, ResourceProfile}
import org.apache.spark.scheduler.cluster.SchedulerBackendUtils.DEFAULT_NUMBER_EXECUTORS
import org.apache.spark.util.{Clock, Utils}
@@ -135,6 +135,7 @@ class ExecutorPodsAllocator(
def start(applicationId: String, schedulerBackend: KubernetesClusterSchedulerBackend): Unit = {
appId = applicationId
+ warnIfRecoveryModeCannotIsolateSingleTask()
if (shouldWaitForDriverReadiness) {
driverPod.foreach { pod =>
// Wait until the driver pod is ready before starting executors, as the headless service
@@ -470,6 +471,28 @@ class ExecutorPodsAllocator(
def setRecoveryMode(): Unit = {
conf.setIfMissing(KUBERNETES_ALLOCATION_RECOVERY_MODE_ENABLED, true)
+ warnIfRecoveryModeCannotIsolateSingleTask()
+ }
+
+ // Recovery mode's single-task guarantee is enforced by announcing SPARK_EXECUTOR_CORES =
+ // ceil(spark.task.cpus), which must be a whole number. When the task cpus amount is 0.5 or
+ // less, the single announced core fits more than one task, so the guarantee cannot hold;
+ // log a prominent warning (at most once) instead of silently overcommitting recovery
+ // executors.
+ private val recoveryModeCpusWarned = new AtomicBoolean(false)
+
+ private def warnIfRecoveryModeCannotIsolateSingleTask(): Unit = {
+ val taskCpus = conf.get(CPUS_PER_TASK)
+ if (conf.get(KUBERNETES_ALLOCATION_RECOVERY_MODE_ENABLED).getOrElse(false) &&
+ taskCpus <= BigDecimal(0.5) && recoveryModeCpusWarned.compareAndSet(false, true)) {
+ val slots = ResourceProfile.numTasksBasedOnCores(
+ CpuAmount.normalize(BigDecimal(1)), taskCpus)
+ logWarning(log"Recovery mode is enabled while spark.task.cpus = " +
+ log"${MDC(LogKeys.NUM_TASK_CPUS, CpuAmount.toDisplayString(taskCpus))} is <= 0.5. A " +
+ log"recovery-mode executor announces a single core, so it accepts up to " +
+ log"${MDC(LogKeys.NUM_SLOTS, slots)} concurrent tasks instead of only one. Set " +
+ log"spark.task.cpus above 0.5 to restore single-task recovery executors.")
+ }
}
protected def requestNewExecutors(
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStepSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStepSuite.scala
index e1261cadf3ab..ba5aa35985da 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStepSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/BasicExecutorFeatureStepSuite.scala
@@ -716,6 +716,27 @@ class BasicExecutorFeatureStepSuite extends SparkFunSuite with BeforeAndAfter {
assert(cores === "2")
}
+ test("SPARK-58192: recovery-mode ENV_EXECUTOR_CORES rounds fractional spark.task.cpus up") {
+ val rpb = new ResourceProfileBuilder()
+ val ereq = new ExecutorResourceRequests()
+ val treq = new TaskResourceRequests()
+ ereq.cores(4)
+ treq.cpus(0.5)
+ rpb.require(ereq).require(treq)
+ val rp = rpb.build()
+
+ baseConf.set(KUBERNETES_ALLOCATION_RECOVERY_MODE_ENABLED, true)
+ Seq("0.5" -> "1", "1.5" -> "2", "2.5" -> "3").foreach { case (taskCpus, expectedCores) =>
+ baseConf.set("spark.task.cpus", taskCpus)
+ val secMgr = new SecurityManager(baseConf)
+ val step = new BasicExecutorFeatureStep(newExecutorConf(), secMgr, rp)
+ val executor = step.configurePod(SparkPod.initialPod())
+ val cores =
+ executor.container.getEnv.asScala.find(_.getName == ENV_EXECUTOR_CORES).get.getValue
+ assert(cores === expectedCores, s"for spark.task.cpus = $taskCpus")
+ }
+ }
+
// There is always exactly one controller reference, and it points to the driver pod.
private def checkOwnerReferences(executor: Pod, driverPodUid: String): Unit = {
assert(executor.getMetadata.getOwnerReferences.size() === 1)
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala
index 3522c49682ab..a3e0b6625ce4 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala
@@ -156,6 +156,33 @@ class ExecutorPodsAllocatorSuite extends SparkFunSuite with BeforeAndAfter {
when(persistentVolumeClaimList.getItems).thenReturn(Seq.empty[PersistentVolumeClaim].asJava)
}
+ test("SPARK-58192: warn when recovery mode cannot isolate a single task") {
+ val confWithFractionalCpus = conf.clone.set(CPUS_PER_TASK, BigDecimal(0.5))
+ val allocator = new ExecutorPodsAllocator(confWithFractionalCpus, secMgr,
+ executorBuilder, kubernetesClient, snapshotsStore, waitForExecutorPodsClock)
+ val logAppender = new LogAppender("recovery mode fractional cpus")
+ withLogAppender(logAppender) {
+ allocator.setRecoveryMode()
+ // the warning is logged at most once
+ allocator.setRecoveryMode()
+ }
+ val warnings = logAppender.loggingEvents
+ .map(_.getMessage.getFormattedMessage)
+ .filter(_.contains("instead of only one"))
+ assert(warnings.size === 1)
+ assert(warnings.head.contains("2 concurrent tasks"))
+
+ // no warning when a recovery-mode executor's single announced core fits exactly one task
+ val allocator2 = new ExecutorPodsAllocator(conf.clone, secMgr,
+ executorBuilder, kubernetesClient, snapshotsStore, waitForExecutorPodsClock)
+ val logAppender2 = new LogAppender("recovery mode whole cpus")
+ withLogAppender(logAppender2) {
+ allocator2.setRecoveryMode()
+ }
+ assert(!logAppender2.loggingEvents.exists(
+ _.getMessage.getFormattedMessage.contains("instead of only one")))
+ }
+
test("SPARK-49447: Prevent small values less than 100 for batch delay") {
val m = intercept[IllegalArgumentException] {
val conf = new SparkConf().set(KUBERNETES_ALLOCATION_BATCH_DELAY.key, "1")
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/SubexpressionEliminationSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/SubexpressionEliminationSuite.scala
index 7971be98f0cd..4f5525158af9 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/SubexpressionEliminationSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/SubexpressionEliminationSuite.scala
@@ -426,7 +426,8 @@ class SubexpressionEliminationSuite extends SparkFunSuite with ExpressionEvalHel
test("SPARK-38333: PlanExpression expression should skip addExprTree function in Executor") {
try {
// suppose we are in executor
- val context1 = new TaskContextImpl(0, 0, 0, 0, 0, 1, null, new Properties, null, cpus = 0)
+ val context1 =
+ new TaskContextImpl(0, 0, 0, 0, 0, 1, null, new Properties, null, cpuAmount = 0)
TaskContext.setTaskContext(context1)
val equivalence = new EquivalentExpressions
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala
index 7450032aa8a1..2d8fec806f7f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala
@@ -85,8 +85,10 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
SparkFiles.getRootDirectory()
builder.environment().put("PATH", path)
// if OMP_NUM_THREADS is not explicitly set, override it with the value of "spark.task.cpus"
+ // which may be fractional, so round up to an integer (at least 1) of threads.
if (System.getenv("OMP_NUM_THREADS") == null) {
- builder.environment().put("OMP_NUM_THREADS", conf.getConfString("spark.task.cpus", "1"))
+ val taskCpus = conf.getConfString("spark.task.cpus", "1.0").toDouble
+ builder.environment().put("OMP_NUM_THREADS", math.max(1, taskCpus.ceil.toInt).toString)
}
val proc = builder.start()
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala
index 5905f0ab8d60..a676eea5ea45 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/SupportsArchiveFormatSuite.scala
@@ -341,7 +341,7 @@ class SupportsArchiveFormatSuite extends SparkFunSuite {
taskMemoryManager = null,
localProperties = new Properties,
metricsSystem = null,
- cpus = 1)
+ cpuAmount = 1)
TaskContext.setTaskContext(ctx)
try {
val it = SupportsArchiveFormat.readArchiveEntries(
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/window/WindowSegmentTreeMemorySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/WindowSegmentTreeMemorySuite.scala
index e60253042c17..74c6c001174e 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/window/WindowSegmentTreeMemorySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/window/WindowSegmentTreeMemorySuite.scala
@@ -81,7 +81,7 @@ class WindowSegmentTreeMemorySuite extends SparkFunSuite with LocalSparkContext
localProperties = new Properties,
metricsSystem = null.asInstanceOf[MetricsSystem],
taskMetrics = TaskMetrics.empty,
- cpus = 1)
+ cpuAmount = 1)
TaskContext.setTaskContext(tc)
try body(tmm, mm)
finally {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org