You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2021/09/09 02:43:40 UTC
[GitHub] [spark] xwu99 opened a new pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
xwu99 opened a new pull request #33941:
URL: https://github.com/apache/spark/pull/33941
### What changes were proposed in this pull request?
https://issues.apache.org/jira/browse/SPARK-36699
We proposed to optionally change behavior of stage-level scheduling by reusing compatible executors. Two executors binding to different resource profiles are **compatible** only when the executorResources (cores in particular if not defining custom resources) are the same, but taskResources can be different. When the executors are compatible, the tasks can be allocated to any of them even when in the different profiles. Users defining profiles should make sure the different taskResources are properly specified against the same executorResources.
A SparkConf option `spark.dynamicAllocation.reuseExecutors` is defined to change the default behavior which is not reusing executors. When this option is turned on, dynamic allocation will count all compatible executors number to meet init/min/max executor number restrictions.
The first PR will focus on reusing executors with same cores without custom resources.
### Why are the changes needed?
Current stage-level scheduling allocated separated set of executors for different executor profiles. This approach simplified implementation, however is a waste of executor resources when the existing executors have enough resources to run the following tasks.
The typical user scenario is for different stages, user wants to use different core number for the task with same executor resources. For instance in CPU machine learning scenario, to achieve the best performance, given the same executor resources, when in ETL stage, user will allocate 1 core per task and many tasks, and in the following CPU training stage, user will use more cores per task and less tasks. In the existing implementation, two separated profiles and executors are created. Reusing executors will get better CPU resource utilization and better performance.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Unit tests
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
tgravescs commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-984734956
Sorry it looks like I missed your previous comments. Yes I think we should support multiple reuse policies and let the user specify. If we can make it a pluggable api that would be good as well and would let them decide what resources they are comfortable with allowing wasted. Maybe they are ok with memory wasted but not GPU for instance.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 commented on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-938380314
> what exactly is the plan here, I know you wanted to get feedback, but are you going to add in check for all the resources to say they are compatible? Part of this comes down to other things as well. Like memory. I might have large containers with the same number of cores, are they ok to reuse. For instance I might have large containers that I'm using for ML vs ETL. So I think we need to define a policy in more detail.
We can discuss this. Actually current reusing condition is conservative (just use cores) rather than flexible (memory and other resources). And need to consider if it's easy for end user to control the resource sharing (Your case with larger memory and same cores will be reused here, it's hard to consider memory, up to user) and make sense for the real use case. I found some use cases for reusing cores but for other resources such as GPU I didn't have a clear thought right now. I will think over and discuss with you and welcome input.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs edited a comment on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
tgravescs edited a comment on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-931402635
A few other concerns here.
1) how does this work with minimum executors, seems easiest enough to not worry about it for this and just reuse if there.
2) how does this apply to executor monitor idle timeout. we don't timeout executors when more tasks to be run, do we take that into account here and keep compatible ones?
3) Does UI show proper/useful information here
4) I don't think this has taken numBarrierSlotsAvailable into account - see calculateAvailableSlots
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 commented on a change in pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 commented on a change in pull request #33941:
URL: https://github.com/apache/spark/pull/33941#discussion_r724710441
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
##########
@@ -384,9 +386,18 @@ private[spark] class TaskSchedulerImpl(
val execId = shuffledOffers(i).executorId
val host = shuffledOffers(i).host
val taskSetRpID = taskSet.taskSet.resourceProfileId
- // make the resource profile id a hard requirement for now - ie only put tasksets
- // on executors where resource profile exactly matches.
- if (taskSetRpID == shuffledOffers(i).resourceProfileId) {
+
+ val assignTasks = if (reuseExecutors) {
+ val compatibleProfiles = sc.resourceProfileManager.getOtherCompatibleProfileIds(taskSetRpID)
Review comment:
Can cache it in the resourceProfileManager and update when new profiles added.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on pull request #33941: [SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
tgravescs commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-1067987968
> Since ResourceProfiles are only supported with dynamic allocation right now. (SparkException("ResourceProfiles are only supported on YARN and Kubernetes with dynamic allocation enabled."). Maybe a seperate PR to address that.
Yes I think this is fine to leave off and do separate
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on a change in pull request #33941: [SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
tgravescs commented on a change in pull request #33941:
URL: https://github.com/apache/spark/pull/33941#discussion_r823831564
##########
File path: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
##########
@@ -502,14 +511,27 @@ private[spark] class ExecutorAllocationManager(
*/
private def addExecutors(maxNumExecutorsNeeded: Int, rpId: Int): Int = {
val oldNumExecutorsTarget = numExecutorsTargetPerResourceProfileId(rpId)
- // Do not request more executors if it would put our target over the upper bound
- // this is doing a max check per ResourceProfile
- if (oldNumExecutorsTarget >= maxNumExecutors) {
- logDebug("Not adding executors because our current target total " +
- s"is already ${oldNumExecutorsTarget} (limit $maxNumExecutors)")
- numExecutorsToAddPerResourceProfileId(rpId) = 1
- return 0
+
+ if (!reuseExecutors) {
+ // Do not request more executors if it would put our target over the upper bound
+ // this is doing a max check per ResourceProfile
+ if (oldNumExecutorsTarget >= maxNumExecutors) {
+ logDebug("Not adding executors because our current target total " +
+ s"is already ${oldNumExecutorsTarget} (limit $maxNumExecutors)")
+ numExecutorsToAddPerResourceProfileId(rpId) = 1
+ return 0
+ }
+ } else {
+ val numCompatibleExecutors = numExecutorsTargetsCompatibleProfiles(rpId)
+ if (oldNumExecutorsTarget + numCompatibleExecutors >= maxNumExecutors) {
+ logDebug("Not adding executors because our current target total is already " +
+ s"${oldNumExecutorsTarget} (limit: max $maxNumExecutors - " +
+ s"reused $numCompatibleExecutors)")
Review comment:
format differently (limit:...)(reused: ...)
##########
File path: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
##########
@@ -518,11 +540,25 @@ private[spark] class ExecutorAllocationManager(
numExecutorsTarget += numExecutorsToAddPerResourceProfileId(rpId)
// Ensure that our target doesn't exceed what we need at the present moment:
numExecutorsTarget = math.min(numExecutorsTarget, maxNumExecutorsNeeded)
- // Ensure that our target fits within configured bounds:
- numExecutorsTarget = math.max(math.min(numExecutorsTarget, maxNumExecutors), minNumExecutors)
+ numExecutorsTarget = if (!reuseExecutors) {
+ // Ensure that our target fits within configured bounds:
+ math.max(math.min(numExecutorsTarget, maxNumExecutors), minNumExecutors)
+ } else {
+ // Ensure that our target fits within adjusted bounds:
+ val numCompatibleExecutors = numExecutorsTargetsCompatibleProfiles(rpId)
+ val adjustedMinNumExecutors = math.max(0, minNumExecutors - numCompatibleExecutors)
+ val adjustedMaxNumExecutors = math.max(1, maxNumExecutors - numCompatibleExecutors)
Review comment:
this part doesn't make sense to me on initial reading. If we set the min and max and our target should still fit within those and not those adjusted. I get that you are trying to say including the compatible ones keep it in that limit but I think this is hard to read to understand that. this would also get more complicated if the reuse policy could change within an application.
##########
File path: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
##########
@@ -297,11 +299,13 @@ private[spark] class ExecutorAllocationManager(
val numRunningOrPendingTasks = pendingTask + pendingSpeculative + running
val rp = resourceProfileManager.resourceProfileFromId(rpId)
val tasksPerExecutor = rp.maxTasksPerExecutor(conf)
- logDebug(s"max needed for rpId: $rpId numpending: $numRunningOrPendingTasks," +
- s" tasksperexecutor: $tasksPerExecutor")
+
val maxNeeded = math.ceil(numRunningOrPendingTasks * executorAllocationRatio /
tasksPerExecutor).toInt
+ logDebug(s"max needed for rpId: $rpId numpending: $numRunningOrPendingTasks," +
+ s" tasksperexecutor: $tasksPerExecutor = $maxNeeded")
Review comment:
we should change to be more descriptive, even just maxNeeded: $maxNeeded
##########
File path: core/src/main/scala/org/apache/spark/internal/config/package.scala
##########
@@ -1958,6 +1958,12 @@ package object config {
.timeConf(TimeUnit.MILLISECONDS)
.createOptional
+ private[spark] val SCHEDULER_REUSE_COMPATIBLE_EXECUTORS =
+ ConfigBuilder("spark.scheduler.reuseCompatibleExecutors")
+ .version("3.3.0")
Review comment:
needs .doc added to it. I need to think about the name of it.
##########
File path: core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala
##########
@@ -283,8 +283,9 @@ case class SparkListenerApplicationEnd(time: Long) extends SparkListenerEvent
case class SparkListenerLogStart(sparkVersion: String) extends SparkListenerEvent
@DeveloperApi
-@Since("3.1.0")
-case class SparkListenerResourceProfileAdded(resourceProfile: ResourceProfile)
+@Since("3.3.0")
+case class SparkListenerResourceProfileAdded(resourceProfile: ResourceProfile,
Review comment:
I would prefer not to change this interface as it gets used by people. I'd rather create a new one if needed but need to look at how its all used. If its just for environment page to show compatible, I'm not sure its worth it. WE can come back to this once figure out main logic
##########
File path: core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala
##########
@@ -77,13 +85,27 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf,
true
}
+ def updateResourceReusePolicy(resourceNames: Set[String],
+ policy: ResourceProfileCompatiblePolicy): Unit = {
+ writeLock.lock()
+ try {
+ reuseResourceNames = resourceNames
+ reusePolicy = policy
Review comment:
I must be missing where this is used? Or you wanted feedback first?
##########
File path: core/src/main/scala/org/apache/spark/rdd/RDD.scala
##########
@@ -1809,6 +1810,22 @@ abstract class RDD[T: ClassTag](
this
}
+ /**
+ * Specify a ResourceProfile and reuse existing compatible executors to use when calculating
+ * this RDD.
+ * @param reuseResourceNames specify what resource should be checked when reusing executors
+ * @param reusePolicy specify executor reuse policy
Review comment:
I would rather see ResourceProfileCompatiblePolicy as a public interface that user could implement their own policy. We can provide a couple very basic ones, equals and AllGreater or something like that.
Also while its more flexible to do per stage it also I think complicates the allocation strategy for dynamic allocation.
We would also need to know if this policy would allow for using this without dynamic allocation, because theoretically if you are reusing executor with same executor profile but different task requirements, you wouldn't need dynamic allocation.
##########
File path: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
##########
@@ -502,14 +511,27 @@ private[spark] class ExecutorAllocationManager(
*/
private def addExecutors(maxNumExecutorsNeeded: Int, rpId: Int): Int = {
val oldNumExecutorsTarget = numExecutorsTargetPerResourceProfileId(rpId)
- // Do not request more executors if it would put our target over the upper bound
- // this is doing a max check per ResourceProfile
- if (oldNumExecutorsTarget >= maxNumExecutors) {
- logDebug("Not adding executors because our current target total " +
- s"is already ${oldNumExecutorsTarget} (limit $maxNumExecutors)")
- numExecutorsToAddPerResourceProfileId(rpId) = 1
- return 0
+
+ if (!reuseExecutors) {
+ // Do not request more executors if it would put our target over the upper bound
+ // this is doing a max check per ResourceProfile
+ if (oldNumExecutorsTarget >= maxNumExecutors) {
+ logDebug("Not adding executors because our current target total " +
+ s"is already ${oldNumExecutorsTarget} (limit $maxNumExecutors)")
+ numExecutorsToAddPerResourceProfileId(rpId) = 1
+ return 0
+ }
+ } else {
+ val numCompatibleExecutors = numExecutorsTargetsCompatibleProfiles(rpId)
+ if (oldNumExecutorsTarget + numCompatibleExecutors >= maxNumExecutors) {
Review comment:
nit extra space after +
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
##########
@@ -384,9 +386,18 @@ private[spark] class TaskSchedulerImpl(
val execId = shuffledOffers(i).executorId
val host = shuffledOffers(i).host
val taskSetRpID = taskSet.taskSet.resourceProfileId
- // make the resource profile id a hard requirement for now - ie only put tasksets
- // on executors where resource profile exactly matches.
- if (taskSetRpID == shuffledOffers(i).resourceProfileId) {
+
+ val assignTasks = if (reuseExecutors) {
+ val compatibleProfiles = sc.resourceProfileManager.getCompatibleProfileIds(taskSetRpID)
+ taskSetRpID == shuffledOffers(i).resourceProfileId ||
+ compatibleProfiles.contains(shuffledOffers(i).resourceProfileId)
+ } else {
Review comment:
nit indentation
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on a change in pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
tgravescs commented on a change in pull request #33941:
URL: https://github.com/apache/spark/pull/33941#discussion_r719482778
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
##########
@@ -384,9 +386,18 @@ private[spark] class TaskSchedulerImpl(
val execId = shuffledOffers(i).executorId
val host = shuffledOffers(i).host
val taskSetRpID = taskSet.taskSet.resourceProfileId
- // make the resource profile id a hard requirement for now - ie only put tasksets
- // on executors where resource profile exactly matches.
- if (taskSetRpID == shuffledOffers(i).resourceProfileId) {
+
+ val assignTasks = if (reuseExecutors) {
+ val compatibleProfiles = sc.resourceProfileManager.getOtherCompatibleProfileIds(taskSetRpID)
Review comment:
this seems fairly expensive to be calculating every loop through the scheduler. we should fine a way to compute and cache and only have to recompute when new resource profile is added.
##########
File path: core/src/main/scala/org/apache/spark/internal/config/package.scala
##########
@@ -572,6 +572,12 @@ package object config {
.booleanConf
.createWithDefault(false)
+ private[spark] val DYN_ALLOCATION_REUSE_EXECUTORS =
+ ConfigBuilder("spark.dynamicAllocation.reuseExecutors")
Review comment:
I think this is to generic and could confuse users, I would prefer to see something in the name specifically using resource profiles. Really I think I would prefer it to be scheduler config like:
spark.scheduler.reuseCompatibleExecutors
version below needs updating
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
tgravescs commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-920020618
thanks for working on this, I'm very busy right now will likely be end of week or early next week and I'll look
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 commented on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-917760485
@tgravescs
related discussions: https://issues.apache.org/jira/browse/SPARK-31437
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 edited a comment on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 edited a comment on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-917760485
@tgravescs Main code is in place but some redundant code is not cleaned yet. I will finish it if the idea is accepted.
related discussions: https://issues.apache.org/jira/browse/SPARK-31437
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
tgravescs commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-931402635
A few other concerns here.
1) how does this work with minimum executors, seems easiest enough to not worry about it for this and just reuse if there.
2) how does this apply to executor monitor idle timeout. we don't timeout executors when more tasks to be run, do we take that into account here and keep compatible ones?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 commented on a change in pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 commented on a change in pull request #33941:
URL: https://github.com/apache/spark/pull/33941#discussion_r724707608
##########
File path: core/src/main/scala/org/apache/spark/internal/config/package.scala
##########
@@ -572,6 +572,12 @@ package object config {
.booleanConf
.createWithDefault(false)
+ private[spark] val DYN_ALLOCATION_REUSE_EXECUTORS =
+ ConfigBuilder("spark.dynamicAllocation.reuseExecutors")
Review comment:
It will impact both dynamic allocation and scheduling. For end user it make sense to put it under scheduler. Also need to document how it would impact the behavior of spark.dynamicAllocation.init/min/max executors for different resource profiles.
Will update version.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 commented on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-938365699
> A few other concerns here.
>
> 1. how does this work with minimum executors, seems easiest enough to not worry about it for this and just reuse if there.
The idea is resource reuse so min/max will be adjusted to take into account for all compatible executors. (i.e. all compatible executors will share one min/max executors number)
https://github.com/apache/spark/blob/67034f284803bd10a487b5c67eb4c552ace950c3/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala#L558
> 2. how does this apply to executor monitor idle timeout. we don't timeout executors when more tasks to be run, do we take that into account here and keep compatible ones?
Compatible executors will be used as the same as the executor with the original profile. It will also timeout if not more tasks. As mentioned above, it will share the min/max executors number with other compatible profiles.
> 3. Does UI show proper/useful information here
Will check what UI need to be changed. Maybe show some compatible info in the Environment / Resource Profiles
> 4. I don't think this has taken numBarrierSlotsAvailable into account - see calculateAvailableSlots
Didn't notice this, will check.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-915715889
Can one of the admins verify this patch?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 commented on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-1021897818
@tgravescs I will add more tests since several places are changed but you can review first and provide feedback. Some updates since our last discussion:
- There are two ways to enable reuse executors: 1) use config option "spark.scheduler.reuseCompatibleExecutors" 2) use overloaded interface "withResources" to specify what resources should be considered and what is the reuse policy (check code). The second one can enable a stage-level reuse policy change in case we want more flexibility. We can discuss which one is better and then refine the code.
- Cache compatible resource profiles in resource profiler manager when profile added
- Added ResourceProfileCompatiblePolicy and interfaces for customization, there are two predefined policies: EQUAL_RESOURCES, MORE_RESOURCES
- Barrior mode support with changes in calculateAvailableSlots.
- Event logging and json protocol support, WebUI changes, need to polish and add more tests.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 edited a comment on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 edited a comment on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-984263614
@tgravescs
Busy for a while, back to this topic. I wil find some time to address code comments above.
For reuse policy, there are two options in my mind right now:
1) **strict match**: only reuse executors with exact same resources (including all 3rd party resources), there is no resource waste but less user flexibility
2) **reuse larger executor**: if there is a larger executor which has resources larger than or equal to current requirements, eg. if you define less memory in the new stage then you can reuse previous executor with larger memory. or if you define new stage with no GPU, you can reuse previous executor with GPU. in both cases, new stage has less resource requrements. But in this policy user should know there is some resource waste. They need to tradeoff reuse executor or create new ones.
Do we also allow user to select reuse policy at stage level?
I am not sure if all policies can be used in real-world scenerio. How about we implement some policy first to get things working for some scenerio and to leave policies options open to add new policy in the future?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 commented on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-1042542512
@tgravescs Do you have time to check this first? I will add more tests later. Thank!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 commented on a change in pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 commented on a change in pull request #33941:
URL: https://github.com/apache/spark/pull/33941#discussion_r724721774
##########
File path: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
##########
@@ -518,11 +551,32 @@ private[spark] class ExecutorAllocationManager(
numExecutorsTarget += numExecutorsToAddPerResourceProfileId(rpId)
// Ensure that our target doesn't exceed what we need at the present moment:
numExecutorsTarget = math.min(numExecutorsTarget, maxNumExecutorsNeeded)
- // Ensure that our target fits within configured bounds:
- numExecutorsTarget = math.max(math.min(numExecutorsTarget, maxNumExecutors), minNumExecutors)
+ numExecutorsTarget = if (!reuseExecutors) {
Review comment:
Pls let me know if there is better way.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 commented on a change in pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 commented on a change in pull request #33941:
URL: https://github.com/apache/spark/pull/33941#discussion_r724710441
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
##########
@@ -384,9 +386,18 @@ private[spark] class TaskSchedulerImpl(
val execId = shuffledOffers(i).executorId
val host = shuffledOffers(i).host
val taskSetRpID = taskSet.taskSet.resourceProfileId
- // make the resource profile id a hard requirement for now - ie only put tasksets
- // on executors where resource profile exactly matches.
- if (taskSetRpID == shuffledOffers(i).resourceProfileId) {
+
+ val assignTasks = if (reuseExecutors) {
+ val compatibleProfiles = sc.resourceProfileManager.getOtherCompatibleProfileIds(taskSetRpID)
Review comment:
Will cache it in the resourceProfileManager and update when new profiles added.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on pull request #33941: [SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
tgravescs commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-1055556707
sorry been very busy, this is on my list to review still, hopefully in the next week.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on pull request #33941: [SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
tgravescs commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-1064285549
It think it would be best to backup and discuss what exactly we want to target for this PR. These are the things I'm thinking:
- Support a basic reuse policy (perhaps EXEC_CORES_EQUAL since I think that was your original use case) and allow user to specify their own. ie config that perhaps load all policies, like the spark.plugins config. User can give the policy a name and reference it by name.
- I think we can keep the reuse policy specified at the application level rather then stage level to keep logic easier for now. We can always add an interface to specify at stage level later.
- Do we want to allow reusing without dynamic allocation? We would have to throw at runtime if we discovered they tried to specify resource profile that wouldn't be compatible. We could keep this PR simple and not support that at first.
does that sound ok?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 commented on pull request #33941: [SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-1067466661
> It think it would be best to backup and discuss what exactly we want to target for this PR. These are the things I'm thinking:
>
> * Support a basic reuse policy (perhaps EXEC_CORES_EQUAL since I think that was your original use case) and allow user to specify their own. ie config that perhaps load all policies, like the spark.plugins config. User can give the policy a name and reference it by name.
> * I think we can keep the reuse policy specified at the application level rather then stage level to keep logic easier for now. We can always add an interface to specify at stage level later.
For those two I agree to keep it simple at first to address the most common scenarios. I would check the other comments in the coming weeks.
> * Do we want to allow reusing without dynamic allocation? We would have to throw at runtime if we discovered they tried to specify resource profile that wouldn't be compatible. We could keep this PR simple and not support that at first.
Since ResourceProfiles are only supported with dynamic allocation right now. (SparkException("ResourceProfiles are only supported on YARN and Kubernetes with dynamic allocation enabled."). Maybe a seperate PR to address that.
> does that sound ok?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 commented on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-984263614
Busy for a while, back to this topic. I wil find some time to address code comments above.
For reuse policy, there are two options in my mind right now:
1) **strict match**: only reuse executors with exact same resources (including all 3rd party resources), there is no resource waste but less user flexibility
2) **reuse larger executor**: if there is a larger executor which has resources larger than or equal to current requirements, eg. if you define less memory in the new stage then you can reuse previous executor with larger memory. or if you define new stage with no GPU, you can reuse previous executor with GPU. in both cases, new stage has less resource requrements. But in this policy user should know there is some resource waste. They need to tradeoff reuse executor or create new ones.
Do we also allow user to select reuse policy at stage level?
I am not sure if all policies can be used in real-world scenerio. How about we implement some policy first to get things working for some scenerio and to leave policies options open to add new policy in the future?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on a change in pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
tgravescs commented on a change in pull request #33941:
URL: https://github.com/apache/spark/pull/33941#discussion_r719490485
##########
File path: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
##########
@@ -518,11 +551,32 @@ private[spark] class ExecutorAllocationManager(
numExecutorsTarget += numExecutorsToAddPerResourceProfileId(rpId)
// Ensure that our target doesn't exceed what we need at the present moment:
numExecutorsTarget = math.min(numExecutorsTarget, maxNumExecutorsNeeded)
- // Ensure that our target fits within configured bounds:
- numExecutorsTarget = math.max(math.min(numExecutorsTarget, maxNumExecutors), minNumExecutors)
+ numExecutorsTarget = if (!reuseExecutors) {
Review comment:
I need to look in more detail perhaps on how to do this, I don't really like haven't to add conditionals in so many places
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 edited a comment on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 edited a comment on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-938365699
> A few other concerns here.
>
> 1. how does this work with minimum executors, seems easiest enough to not worry about it for this and just reuse if there.
The idea is resource reuse so min/max will be adjusted to take into account for all compatible executors. (i.e. all compatible executors will share one min/max executors number)
https://github.com/apache/spark/blob/67034f284803bd10a487b5c67eb4c552ace950c3/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala#L558
> 2. how does this apply to executor monitor idle timeout. we don't timeout executors when more tasks to be run, do we take that into account here and keep compatible ones?
Compatible executors will be used as the same as the executor with the original profile. It will also timeout if not more tasks. As mentioned above, it will share the min/max executors number with other compatible profiles.
> 3. Does UI show proper/useful information here
Will check what UI need to be changed. Maybe show some compatible info in the Environment / Resource Profiles
> 4. I don't think this has taken numBarrierSlotsAvailable into account - see calculateAvailableSlots
Didn't notice this, will check.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xwu99 edited a comment on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
xwu99 edited a comment on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-1042542512
@tgravescs Do you have time to check this first? I will add more tests later. Thanks!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on pull request #33941: [WIP][SPARK-36699][Core] Reuse compatible executors for stage-level scheduling
Posted by GitBox <gi...@apache.org>.
tgravescs commented on pull request #33941:
URL: https://github.com/apache/spark/pull/33941#issuecomment-1015545919
@xwu99 please let me know when you think this is at a point to review again
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org