pan3793 commented on code in PR #57332:
URL: https://github.com/apache/spark/pull/57332#discussion_r3620667712
##########
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:
addressed, now it fails fast for the 0 MiB case
--
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]
