tgravescs commented on a change in pull request #30204:
URL: https://github.com/apache/spark/pull/30204#discussion_r523108278
##########
File path:
resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala
##########
@@ -302,48 +273,38 @@ private[yarn] class YarnAllocator(
}
// if a ResourceProfile hasn't been seen yet, create the corresponding YARN
Resource for it
- private def createYarnResourceForResourceProfile(
- resourceProfileToTotalExecs: Map[ResourceProfile, Int]): Unit =
synchronized {
- resourceProfileToTotalExecs.foreach { case (rp, num) =>
- if (!rpIdToYarnResource.contains(rp.id)) {
- // Start with the application or default settings
- var heapMem = executorMemory.toLong
- var offHeapMem = executorOffHeapMemory.toLong
- var overheadMem = memoryOverhead.toLong
- var pysparkMem = pysparkWorkerMemory.toLong
- var cores = defaultExecutorCores
- val customResources = new mutable.HashMap[String, String]
- // track the resource profile if not already there
- getOrUpdateRunningExecutorForRPId(rp.id)
- logInfo(s"Resource profile ${rp.id} doesn't exist, adding it")
- val execResources = rp.executorResources
- execResources.foreach { case (r, execReq) =>
- r match {
- case ResourceProfile.MEMORY =>
- heapMem = execReq.amount
- case ResourceProfile.OVERHEAD_MEM =>
- overheadMem = execReq.amount
- case ResourceProfile.PYSPARK_MEM =>
- pysparkMem = execReq.amount
- case ResourceProfile.OFFHEAP_MEM =>
- offHeapMem =
YarnSparkHadoopUtil.executorOffHeapMemorySizeAsMb(sparkConf, execReq)
- case ResourceProfile.CORES =>
- cores = execReq.amount.toInt
- case "gpu" =>
- customResources(YARN_GPU_RESOURCE_CONFIG) =
execReq.amount.toString
- case "fpga" =>
- customResources(YARN_FPGA_RESOURCE_CONFIG) =
execReq.amount.toString
- case rName =>
- customResources(rName) = execReq.amount.toString
- }
+ private def createYarnResourceForResourceProfile(rp: ResourceProfile): Unit
= synchronized {
+ if (!rpIdToYarnResource.contains(rp.id)) {
+ // track the resource profile if not already there
+ getOrUpdateRunningExecutorForRPId(rp.id)
+ logInfo(s"Resource profile ${rp.id} doesn't exist, adding it")
+ val resourcesWithDefaults =
+ ResourceProfile.getResourcesForClusterManager(rp.id,
rp.executorResources,
+ MEMORY_OVERHEAD_FACTOR, sparkConf, isPythonApp,
+ ResourceRequestHelper.resourceNameMapping)
+ val customSparkResources =
+ resourcesWithDefaults.customResources.map { case (name, execReq) =>
+ (name, execReq.amount.toString)
}
- val totalMem = (heapMem + offHeapMem + overheadMem + pysparkMem).toInt
- val resource = Resource.newInstance(totalMem, cores)
- ResourceRequestHelper.setResourceRequests(customResources.toMap,
resource)
- logDebug(s"Created resource capability: $resource")
- rpIdToYarnResource.putIfAbsent(rp.id, resource)
- rpIdToResourceProfile(rp.id) = rp
+ // YARN specified resources are only supported with the default profile
and we only
+ // pick up Spark custom resources for GPU and FPGA with the default
profile
+ // to allow option of them being scheduled off of or not.
+ // We don't currently have a way to specify just YARN custom resources
via the
+ // ResourceProfile so we send along all custom resources defined.
Review comment:
In some ways its a bit weird which is why I wanted to document it but
there is currently no way through resource profiles api to clear out any yarn
specific configs (or k8s templates).
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
