sunchao commented on code in PR #57332:
URL: https://github.com/apache/spark/pull/57332#discussion_r3619154191
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -154,6 +155,26 @@ 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))
+ mem.map(_ / taskSlots)
Review Comment:
[P2] Do not turn a positive PySpark memory budget into no limit
When the task-slot count exceeds the configured memory in MiB, this integer
division returns zero. For example, 64 executor cores, `spark.task.cpus=0.1`,
and 512 MiB produce 640 slots and `512 / 640 == 0`; `setup_memory_limits`
applies `RLIMIT_AS` only when this value is positive, so every worker becomes
unlimited instead of sharing the configured cap. Please fail fast when no
positive MiB share is representable, or otherwise preserve an enforceable limit.
##########
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
Review Comment:
[P2] Derive script OMP threads from the active task profile
A script transformation can be a narrow child of `mapInPandas` or
`mapInArrow` carrying a different resource profile, so DAGScheduler runs this
process inside that profiled task even though this reads the global
`spark.task.cpus`. Moreover, Spark normally puts the global default in the
executor's `OMP_NUM_THREADS`, making the null check skip any per-stage
correction. Please preserve an explicit user override but otherwise derive this
child process's thread count from `TaskContext.get().cpus()`.
--
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]
