sunchao commented on code in PR #57332:
URL: https://github.com/apache/spark/pull/57332#discussion_r3627318835
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -154,6 +155,41 @@ 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 concurrently
running tasks,
+ * which is floor(execCores / taskCpus) rather than the plain core count: a
fractional
+ * `spark.task.cpus` below 1 admits more concurrent tasks than there are
cores, and
+ * dividing by the core count alone would let the workers' aggregate limits
exceed the
+ * executor-wide allocation.
+ */
+ private[spark] def getWorkerMemoryMb(
+ mem: Option[Long],
+ execCores: Int,
+ taskCpus: BigDecimal): Option[Long] = {
+ // The task cpus amount can exceed the announced cores in misconfigured
corners (e.g.
+ // standalone mode, where EXECUTOR_CORES defaults to 1 regardless of the
actual core
+ // count, see SPARK-30299); never split into less than one slot.
+ val taskSlots =
+ math.max(1, ResourceProfile.numTasksBasedOnCores(BigDecimal(execCores),
taskCpus))
Review Comment:
[P2] Account for the actual limiting resource before rejecting PySpark memory
The new zero-share check assumes that `floor(executor cores / task CPUs)` is
the number of concurrently runnable Python tasks, but a custom resource can
impose a much lower limit.
For example:
- Executor: 64 cores, one GPU, and `spark.executor.pyspark.memory=512m`.
- Task: `spark.task.cpus=0.1` and one GPU.
The scheduler can run only one task because the GPU is the limiting
resource, so a 512-MiB worker limit is entirely enforceable. This code instead
calculates 640 CPU slots, rounds `512 / 640` down to zero, and fails every task
as though the configuration were unsatisfiable.
Please derive both the memory split and the fail-fast decision from the
active resource profile's actual maximum concurrent tasks, including
custom-resource limits, and add a GPU-limited regression test.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/BaseScriptTransformationExec.scala:
##########
@@ -84,10 +100,19 @@ trait BaseScriptTransformationExec extends UnaryExecNode {
val path = System.getenv("PATH") + File.pathSeparator +
SparkFiles.getRootDirectory()
builder.environment().put("PATH", path)
- // if OMP_NUM_THREADS is not explicitly set, override it with the value of
"spark.task.cpus"
- if (System.getenv("OMP_NUM_THREADS") == null) {
- builder.environment().put("OMP_NUM_THREADS",
conf.getConfString("spark.task.cpus", "1"))
- }
+ // Derive the script process's OMP_NUM_THREADS from the task's own cpu
amount, which honors
+ // stage-level resource profiles (a script transform can be a child of
mapInPandas/mapInArrow
+ // carrying a different profile), rounded up to an integer >= 1. We check
the config rather
+ // than System.getenv for an explicit user override, because Spark seeds
the executor's
+ // OMP_NUM_THREADS with the global default, so a getenv check would always
suppress the
+ // per-stage value.
+ val ompUserOverride =
+
Option(SparkEnv.get).exists(_.conf.contains("spark.executorEnv.OMP_NUM_THREADS"))
Review Comment:
[P2] Preserve OMP overrides supplied through the inherited environment
The previous `System.getenv("OMP_NUM_THREADS")` check preserved any value
explicitly inherited by the executor. This replacement recognizes only
`spark.executorEnv.OMP_NUM_THREADS`, so an override supplied through a
Kubernetes executor pod template, container image, or exported local-mode
environment is no longer recognized.
For example, an executor pod template can set `OMP_NUM_THREADS=8` without
adding a SparkConf entry. A `0.5`-CPU task now silently replaces that inherited
value with `1`, whereas the previous code preserved `8`.
Please preserve inherited user-provided overrides while distinguishing them
from Spark's automatically generated global default, and cover the actual
script child-process environment in a regression test.
--
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]
