sunchao commented on code in PR #57332:
URL: https://github.com/apache/spark/pull/57332#discussion_r3604937650
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -310,9 +310,11 @@ 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. `spark.task.cpus` may be fractional, so
round up to an integer
Review Comment:
[P2] Split PySpark memory across fractional task slots
The fractional concurrency enabled here invalidates `getWorkerMemoryMb`'s
assumption above that the worker pool has at most `execCores` workers. With 4
executor cores, `spark.task.cpus=0.5`, and 4 GiB of
`spark.executor.pyspark.memory`, Spark can now run 8 Python tasks concurrently,
but each worker still receives `4096 / 4 = 1024` MiB. Their aggregate limits
can therefore reach 8 GiB even though YARN/Kubernetes adds only 4 GiB to the
executor container request, defeating the configured cap and potentially OOMing
the executor. Please divide the executor-wide allocation by the maximum
concurrent task slots for the active resource profile (or another conservative
concurrency bound), and cover this case.
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -950,10 +954,15 @@ 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.
+ // With a fractional cpus below 1 the scaled elapsed time can be
smaller than the in-run
+ // deserialization time, so clamp at zero rather than reporting a
negative run time.
+ task.metrics.setExecutorRunTime(math.max(0L,
TimeUnit.NANOSECONDS.toMillis(
+ // Strip trailing zeros from the scale-9 cpus so this product stays
in BigDecimal's
+ // compact (long-backed) form instead of inflating a long duration
into a BigInteger.
+ (BigDecimal(taskFinishNs - taskStartTimeNs) *
+ CpuAmount.stripTrailingZeros(taskDescription.cpus)).toLong
+ - task.executorDeserializeTimeNs)))
Review Comment:
[P2] Subtract deserialization before scaling executorRunTime
This currently computes `cpus * (deserialization + execution) -
deserialization`. With `cpus=0.2`, 100 ms of in-run deserialization, and 100 ms
of execution, it produces `-60 ms` and the clamp reports zero, although the
CPU-weighted execution interval is `0.2 * (200 - 100) = 20 ms`. Besides
under-reporting stage metrics, a zero runtime makes efficient speculation
discard the task's process-rate sample. Please remove the in-run
deserialization interval before multiplying by the task CPU amount.
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -310,9 +310,11 @@ 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. `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.
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",
+ conf.get(CPUS_PER_TASK).setScale(0,
BigDecimal.RoundingMode.CEILING).intValue.toString)
Review Comment:
[P2] Use the task's stage-level CPU amount for OMP_NUM_THREADS
This reads the executor's global `spark.task.cpus`, but the scheduler may
have reserved a different amount from the stage resource profile and already
exposes it as `context.cpuAmount()`. For example, with global CPUs `3`, a stage
profile requesting `0.5`, and a 4-core executor, Spark schedules 8 Python
workers while each gets `OMP_NUM_THREADS=3`, allowing 24 native threads on 4
cores; the inverse mismatch underutilizes larger stage requests. Python worker
factories are keyed by the environment map, so using the ceiling of
`context.cpuAmount()` will also keep worker reuse separated correctly by
profile.
##########
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
Review Comment:
[P2] Derive recovery executor capacity from the profile being launched
`ExecutorPodsAllocator` resolves the requested profile ID and passes that
`ResourceProfile` into this feature step, but this branch ignores it and rounds
the global `spark.task.cpus`. With global CPUs `1` and a stage profile
requesting `1.5`, a recovery executor for that profile announces one core and
can never launch its 1.5-core tasks. With global CPUs `2` and a stage profile
requesting `0.5`, it announces two cores and accepts four tasks, defeating
recovery isolation. Please derive this value (and the corresponding warning)
from the resource profile being allocated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
