From 3a7d0605a7f1bc05061d4f4c5a6fdae5ae6c66f5 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Wed, 1 Sep 2021 21:11:10 +0800 Subject: [PATCH 01/26] Add some utilities for profiles --- .../spark/resource/ResourceProfile.scala | 5 + .../resource/ResourceProfileManager.scala | 14 +++ .../ExecutorAllocationManagerSuite.scala | 92 +++++++++++++++++++ 3 files changed, 111 insertions(+) diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala index 339870195044c..f48b754cb6c68 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala @@ -241,6 +241,11 @@ class ResourceProfile( rp.taskResources == taskResources && rp.executorResources == executorResources } + // check that executor resources are equal, but the task resources could be different + private[spark] def resourcesCompatible(rp: ResourceProfile): Boolean = { + rp.executorResources == executorResources + } + override def hashCode(): Int = Seq(taskResources, executorResources).hashCode() override def toString(): String = { diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala index 2858443c7cd33..6a0674f92448f 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala @@ -127,4 +127,18 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, readLock.unlock() } } + + /* + * Get all compatible profiles + */ + def getCompatibleProfiles(rp: ResourceProfile): Map[Int, ResourceProfile] = { + readLock.lock() + try { + resourceProfileIdToResourceProfile.filter { case (_, rpEntry) => + rpEntry.resourcesCompatible(rp) + }.toMap + } finally { + readLock.unlock() + } + } } diff --git a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala index 2fb5140d38b2c..3ec9674b05edd 100644 --- a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala +++ b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala @@ -286,6 +286,72 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite { assert(numExecutorsTarget(manager, rprof1.id) === 2) } + test("reuse executors") { + // scalastyle:off println + + val manager = createManager(createConfReuseExecutors(1, 10, 1)) + println("--- SparkListenerStageSubmitted defaultProfile ---") + post(SparkListenerStageSubmitted(createStageInfo(0, 1000, rp = defaultProfile))) + val rp1 = new ResourceProfileBuilder() + val execReqs = new ExecutorResourceRequests().cores(4) + val taskReqs = new TaskResourceRequests().cpus(2) + rp1.require(execReqs).require(taskReqs) + val rprof1 = rp1.build + rpManager.addResourceProfile(rprof1) + println("--- SparkListenerStageSubmitted rprof1 ---") + post(SparkListenerStageSubmitted(createStageInfo(1, 1000, rp = rprof1))) + val updatesNeeded = + new mutable.HashMap[ResourceProfile, ExecutorAllocationManager.TargetNumUpdates] + + // Keep adding until the limit is reached + assert(numExecutorsTargetForDefaultProfileId(manager) === 1) + assert(numExecutorsToAddForDefaultProfile(manager) === 1) + assert(addExecutorsToTargetForDefaultProfile(manager, updatesNeeded) === 1) + assert(numExecutorsToAdd(manager, rprof1) === 1) + assert(numExecutorsTarget(manager, rprof1.id) === 1) + println("--- addExecutorsToTarget ---") + assert(addExecutorsToTarget(manager, updatesNeeded, rprof1) === 1) + println("--- doUpdateRequest ---") + doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) + assert(numExecutorsTargetForDefaultProfileId(manager) === 2) + assert(numExecutorsToAddForDefaultProfile(manager) === 2) + assert(addExecutorsToTargetForDefaultProfile(manager, updatesNeeded) === 2) + assert(numExecutorsToAdd(manager, rprof1) === 2) + assert(numExecutorsTarget(manager, rprof1.id) === 2) + println("--- addExecutorsToTarget ---") + assert(addExecutorsToTarget(manager, updatesNeeded, rprof1) === 2) + println("--- doUpdateRequest ---") + doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) + assert(numExecutorsTargetForDefaultProfileId(manager) === 4) + assert(numExecutorsToAddForDefaultProfile(manager) === 4) + assert(addExecutorsToTargetForDefaultProfile(manager, updatesNeeded) === 4) + assert(numExecutorsToAdd(manager, rprof1) === 4) + assert(numExecutorsTarget(manager, rprof1.id) === 4) + assert(addExecutorsToTarget(manager, updatesNeeded, rprof1) === 4) + doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) + assert(numExecutorsTargetForDefaultProfileId(manager) === 8) + assert(numExecutorsToAddForDefaultProfile(manager) === 8) + // reached the limit of 10 + assert(addExecutorsToTargetForDefaultProfile(manager, updatesNeeded) === 2) + assert(numExecutorsToAdd(manager, rprof1) === 8) + assert(numExecutorsTarget(manager, rprof1.id) === 8) + assert(addExecutorsToTarget(manager, updatesNeeded, rprof1) === 2) + doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) + assert(numExecutorsTargetForDefaultProfileId(manager) === 10) + assert(numExecutorsToAddForDefaultProfile(manager) === 1) + assert(addExecutorsToTargetForDefaultProfile(manager, updatesNeeded) === 0) + assert(numExecutorsToAdd(manager, rprof1) === 1) + assert(numExecutorsTarget(manager, rprof1.id) === 10) + assert(addExecutorsToTarget(manager, updatesNeeded, rprof1) === 0) + doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) + assert(numExecutorsTargetForDefaultProfileId(manager) === 10) + assert(numExecutorsToAddForDefaultProfile(manager) === 1) + assert(numExecutorsToAdd(manager, rprof1) === 1) + assert(numExecutorsTarget(manager, rprof1.id) === 10) + + // scalastyle:on println + } + test("remove executors multiple profiles") { val manager = createManager(createConf(5, 10, 5)) val rp1 = new ResourceProfileBuilder() @@ -1718,6 +1784,32 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite { sparkConf } + private def createConfReuseExecutors( + minExecutors: Int = 1, + maxExecutors: Int = 5, + initialExecutors: Int = 1, + decommissioningEnabled: Boolean = false): SparkConf = { + val sparkConf = new SparkConf() + .set(config.DYN_ALLOCATION_ENABLED, true) + .set(config.DYN_ALLOCATION_MIN_EXECUTORS, minExecutors) + .set(config.DYN_ALLOCATION_MAX_EXECUTORS, maxExecutors) + .set(config.DYN_ALLOCATION_INITIAL_EXECUTORS, initialExecutors) + .set(config.DYN_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT.key, + s"${schedulerBacklogTimeout.toString}s") + .set(config.DYN_ALLOCATION_SUSTAINED_SCHEDULER_BACKLOG_TIMEOUT.key, + s"${sustainedSchedulerBacklogTimeout.toString}s") + .set(config.DYN_ALLOCATION_EXECUTOR_IDLE_TIMEOUT.key, s"${executorIdleTimeout.toString}s") + .set(config.SHUFFLE_SERVICE_ENABLED, true) + .set(config.DYN_ALLOCATION_TESTING, true) + // SPARK-22864/SPARK-32287: effectively disable the allocation schedule for the tests so that + // we won't result in the race condition between thread "spark-dynamic-executor-allocation" + // and thread "pool-1-thread-1-ScalaTest-running". + .set(TEST_DYNAMIC_ALLOCATION_SCHEDULE_ENABLED, false) + .set(DECOMMISSION_ENABLED, decommissioningEnabled) + .set(config.EXECUTOR_CORES, 4) + sparkConf + } + private def createManager( conf: SparkConf, clock: Clock = new SystemClock()): ExecutorAllocationManager = { From 96fa10389ef3af4308b2fe7b89d3422f2e80af25 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Thu, 2 Sep 2021 18:42:31 +0800 Subject: [PATCH 02/26] Add utilities for compatible executors --- .../spark/ExecutorAllocationManager.scala | 39 +++++++++++-- .../spark/internal/config/package.scala | 6 ++ .../spark/resource/ResourceProfile.scala | 3 +- .../resource/ResourceProfileManager.scala | 43 ++++++++++++++- .../ExecutorAllocationManagerSuite.scala | 55 ++++++++++++++++++- 5 files changed, 139 insertions(+), 7 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala index c4b619300b583..973306a5c1c7d 100644 --- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala +++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala @@ -110,6 +110,8 @@ private[spark] class ExecutorAllocationManager( import ExecutorAllocationManager._ + private val reuseExecutors = conf.get(DYN_ALLOCATION_REUSE_EXECUTORS) + // Lower and upper bounds on the number of executors. private val minNumExecutors = conf.get(DYN_ALLOCATION_MIN_EXECUTORS) private val maxNumExecutors = conf.get(DYN_ALLOCATION_MAX_EXECUTORS) @@ -297,10 +299,15 @@ 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") +// logDebug(s"max needed for rpId: $rpId numpending: $numRunningOrPendingTasks," + +// s" tasksperexecutor: $tasksPerExecutor") + val numOtherCompatibleExecutors = executorCountWithCompatibleResourceProfile(rpId) val maxNeeded = math.ceil(numRunningOrPendingTasks * executorAllocationRatio / - tasksPerExecutor).toInt + tasksPerExecutor).toInt - numOtherCompatibleExecutors + + logDebug(s"$numOtherCompatibleExecutors compatible executors and " + + s"$maxNeeded max needed for rpId: $rpId numpending: $numRunningOrPendingTasks," + + s" tasksperexecutor: $tasksPerExecutor") val maxNeededWithSpeculationLocalityOffset = if (tasksPerExecutor > 1 && maxNeeded == 1 && pendingSpeculative > 0) { @@ -490,6 +497,14 @@ private[spark] class ExecutorAllocationManager( numExecutorsTargetPerResourceProfileId(rpId) - oldNumExecutorsTarget } + private def executorCountWithCompatibleResourceProfile(rpId: Int): Int = { + val compatibleProfileIds = resourceProfileManager.getOtherCompatibleProfileIds(rpId) + + logDebug("compatibleProfileIds for rpId " + rpId + ": " + compatibleProfileIds.mkString(" ")) + + compatibleProfileIds.map { executorMonitor.executorCountWithResourceProfile(_) }.sum + } + /** * Update the target number of executors and figure out how many to add. * If the cap on the number of executors is reached, give up and reset the @@ -502,6 +517,11 @@ private[spark] class ExecutorAllocationManager( */ private def addExecutors(maxNumExecutorsNeeded: Int, rpId: Int): Int = { val oldNumExecutorsTarget = numExecutorsTargetPerResourceProfileId(rpId) + logDebug(s"rpId " + rpId + ": executorCount, " + + executorMonitor.executorCountWithResourceProfile(rpId) + + ", compatible executorCount: " + + executorCountWithCompatibleResourceProfile(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) { @@ -512,8 +532,19 @@ private[spark] class ExecutorAllocationManager( } // There's no point in wasting time ramping up to the number of executors we already have, so // make sure our target is at least as much as our current allocation: + var numExecutorsTarget = math.max(numExecutorsTargetPerResourceProfileId(rpId), - executorMonitor.executorCountWithResourceProfile(rpId)) + executorMonitor.executorCountWithResourceProfile(rpId)) + +// var numExecutorsTarget = if (reuseExecutors) { +// logDebug("Using number of compatible executors as target") +// math.max(numExecutorsTargetPerResourceProfileId(rpId), +// executorCountWithCompatibleResourceProfile(rpId)) +// } else { +// math.max(numExecutorsTargetPerResourceProfileId(rpId), +// executorMonitor.executorCountWithResourceProfile(rpId)) +// } + // Boost our target with the number to add for this round: numExecutorsTarget += numExecutorsToAddPerResourceProfileId(rpId) // Ensure that our target doesn't exceed what we need at the present moment: diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index a942ba5401ab2..bd7e714fd729e 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/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") + .version("1.2.0") + .booleanConf + .createWithDefault(false) + private[spark] val DYN_ALLOCATION_TESTING = ConfigBuilder("spark.dynamicAllocation.testing") .version("1.2.0") diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala index f48b754cb6c68..fe2e14e4507c3 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala @@ -241,9 +241,10 @@ class ResourceProfile( rp.taskResources == taskResources && rp.executorResources == executorResources } + // TODO: define what a compatible executor is, right now is equal cores // check that executor resources are equal, but the task resources could be different private[spark] def resourcesCompatible(rp: ResourceProfile): Boolean = { - rp.executorResources == executorResources + rp.executorResources("cores") == executorResources("cores") } override def hashCode(): Int = Seq(taskResources, executorResources).hashCode() diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala index 6a0674f92448f..54b873e02f8d7 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala @@ -129,7 +129,7 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, } /* - * Get all compatible profiles + * Get all compatible profiles including itself */ def getCompatibleProfiles(rp: ResourceProfile): Map[Int, ResourceProfile] = { readLock.lock() @@ -141,4 +141,45 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, readLock.unlock() } } + + /* + * Get all compatible profile IDs excluding itself + */ + def getOtherCompatibleProfileIds(rpId: Int): Set[Int] = { + val resourceProfile = resourceProfileFromId(rpId) + readLock.lock() + try { + resourceProfileIdToResourceProfile.filter { case (id: Int, rpEntry: ResourceProfile) => + id != rpId && rpEntry.resourcesCompatible(resourceProfile) + }.map(_._1).toSet + } finally { + readLock.unlock() + } + } + + /* + * Get all compatible profile count including itself + */ +// def getCompatibleProfileCount(rpId: Int): Int = { +// val resourceProfile = resourceProfileFromId(rpId) +// readLock.lock() +// try { +// resourceProfileIdToResourceProfile.count { case (_, rpEntry) => +// rpEntry.resourcesCompatible(resourceProfile) +// } +// } finally { +// readLock.unlock() +// } +// } + + def dumpResourceProfile(): Unit = { + readLock.lock() + try { + resourceProfileIdToResourceProfile.foreach { case (id: Int, rpEntry: ResourceProfile) => + println("profile " + id + ": " + rpEntry) + } + } finally { + readLock.unlock() + } + } } diff --git a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala index 3ec9674b05edd..63c4c3757b449 100644 --- a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala +++ b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala @@ -119,7 +119,7 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite { } test("add executors default profile") { - val manager = createManager(createConf(1, 10, 1)) + val manager = createManager(createConfReuseExecutors(1, 10, 1)) post(SparkListenerStageSubmitted(createStageInfo(0, 1000))) val updatesNeeded = @@ -290,6 +290,12 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite { // scalastyle:off println val manager = createManager(createConfReuseExecutors(1, 10, 1)) +// rpManager.dumpResourceProfile() + +// onExecutorAddedDefaultProfile(manager, "first") +// onExecutorAddedDefaultProfile(manager, "second") +// onExecutorAddedDefaultProfile(manager, "third") + println("--- SparkListenerStageSubmitted defaultProfile ---") post(SparkListenerStageSubmitted(createStageInfo(0, 1000, rp = defaultProfile))) val rp1 = new ResourceProfileBuilder() @@ -298,8 +304,15 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite { rp1.require(execReqs).require(taskReqs) val rprof1 = rp1.build rpManager.addResourceProfile(rprof1) +// rpManager.dumpResourceProfile() println("--- SparkListenerStageSubmitted rprof1 ---") post(SparkListenerStageSubmitted(createStageInfo(1, 1000, rp = rprof1))) + +// onExecutorAdded(manager, "firstrp1", rprof1) +// onExecutorAdded(manager, "secondrp1", rprof1) +// onExecutorAdded(manager, "thirdrp1", rprof1) +// onExecutorAdded(manager, "fourthrp1", rprof1) + val updatesNeeded = new mutable.HashMap[ResourceProfile, ExecutorAllocationManager.TargetNumUpdates] @@ -349,6 +362,45 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite { assert(numExecutorsToAdd(manager, rprof1) === 1) assert(numExecutorsTarget(manager, rprof1.id) === 10) + println("--- Register previously requested executors ---") + + onExecutorAddedDefaultProfile(manager, "first") + onExecutorAddedDefaultProfile(manager, "second") + logDebug(numExecutorsTargetForDefaultProfileId(manager).toString) + logDebug(numExecutorsToAddForDefaultProfile(manager).toString) + onExecutorAddedDefaultProfile(manager, "third") + onExecutorAddedDefaultProfile(manager, "fourth") + + logDebug("numExecutorsTargetForDefaultProfileId: " + + numExecutorsTargetForDefaultProfileId(manager).toString) + logDebug("numExecutorsToAddForDefaultProfile: " + + numExecutorsToAddForDefaultProfile(manager).toString) + logDebug("addExecutorsToTargetForDefaultProfile: " + + addExecutorsToTargetForDefaultProfile(manager, updatesNeeded).toString) + doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) + + onExecutorAdded(manager, "firstrp1", rprof1) + onExecutorAdded(manager, "secondrp1", rprof1) + + logDebug("numExecutorsTargetForDefaultProfileId: " + + numExecutorsTargetForDefaultProfileId(manager).toString) + logDebug("numExecutorsToAddForDefaultProfile: " + + numExecutorsToAddForDefaultProfile(manager).toString) + logDebug("addExecutorsToTargetForDefaultProfile: " + + addExecutorsToTargetForDefaultProfile(manager, updatesNeeded).toString) + doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) + + onExecutorAdded(manager, "thirdrp1", rprof1) + onExecutorAdded(manager, "fourthrp1", rprof1) + + logDebug("numExecutorsTargetForDefaultProfileId: " + + numExecutorsTargetForDefaultProfileId(manager).toString) + logDebug("numExecutorsToAddForDefaultProfile: " + + numExecutorsToAddForDefaultProfile(manager).toString) + logDebug("addExecutorsToTargetForDefaultProfile: " + + addExecutorsToTargetForDefaultProfile(manager, updatesNeeded).toString) + doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) + // scalastyle:on println } @@ -1791,6 +1843,7 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite { decommissioningEnabled: Boolean = false): SparkConf = { val sparkConf = new SparkConf() .set(config.DYN_ALLOCATION_ENABLED, true) + .set(config.DYN_ALLOCATION_REUSE_EXECUTORS, true) .set(config.DYN_ALLOCATION_MIN_EXECUTORS, minExecutors) .set(config.DYN_ALLOCATION_MAX_EXECUTORS, maxExecutors) .set(config.DYN_ALLOCATION_INITIAL_EXECUTORS, initialExecutors) From 8f9d0fecbbfdc4ba3b599f4dc3c13cf6bc8441e6 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Fri, 3 Sep 2021 11:03:53 +0800 Subject: [PATCH 03/26] update TaskSchedulerImpl for reuse executors --- .../spark/ExecutorAllocationManager.scala | 12 +++++++++--- .../spark/scheduler/TaskSchedulerImpl.scala | 17 ++++++++++++++--- 2 files changed, 23 insertions(+), 6 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala index 973306a5c1c7d..e44d3691c57c5 100644 --- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala +++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala @@ -301,10 +301,16 @@ private[spark] class ExecutorAllocationManager( val tasksPerExecutor = rp.maxTasksPerExecutor(conf) // logDebug(s"max needed for rpId: $rpId numpending: $numRunningOrPendingTasks," + // s" tasksperexecutor: $tasksPerExecutor") - val numOtherCompatibleExecutors = executorCountWithCompatibleResourceProfile(rpId) - val maxNeeded = math.ceil(numRunningOrPendingTasks * executorAllocationRatio / - tasksPerExecutor).toInt - numOtherCompatibleExecutors + val numOtherCompatibleExecutors = executorCountWithCompatibleResourceProfile(rpId) + // if reusing executors, should subtract number of compatible executors + val maxNeeded = if (reuseExecutors) { + math.ceil(numRunningOrPendingTasks * executorAllocationRatio / + tasksPerExecutor).toInt - numOtherCompatibleExecutors + } else { + math.ceil(numRunningOrPendingTasks * executorAllocationRatio / + tasksPerExecutor).toInt + } logDebug(s"$numOtherCompatibleExecutors compatible executors and " + s"$maxNeeded max needed for rpId: $rpId numpending: $numRunningOrPendingTasks," + s" tasksperexecutor: $tasksPerExecutor") diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala index 55db73ab2a045..bfc60568b39f4 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala @@ -103,6 +103,8 @@ private[spark] class TaskSchedulerImpl( // of tasks that are very short. val MIN_TIME_TO_SPECULATION = conf.get(SPECULATION_MIN_THRESHOLD) + private val reuseExecutors = conf.get(DYN_ALLOCATION_REUSE_EXECUTORS) + private val speculationScheduler = ThreadUtils.newDaemonSingleThreadScheduledExecutor("task-scheduler-speculation") @@ -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) + taskSetRpID == shuffledOffers(i).resourceProfileId || + compatibleProfiles.contains(shuffledOffers(i).resourceProfileId) + } else { + // make the resource profile id a hard requirement for now - ie only put tasksets + // on executors where resource profile exactly matches. + taskSetRpID == shuffledOffers(i).resourceProfileId + } + + if (assignTasks) { val taskResAssignmentsOpt = resourcesMeetTaskRequirements(taskSet, availableCpus(i), availableResources(i)) taskResAssignmentsOpt.foreach { taskResAssignments => From 5a4067a88535fd73a34c3aab70baf2703c09ca1f Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Fri, 3 Sep 2021 15:48:41 +0800 Subject: [PATCH 04/26] Add test for reusable executors for multiple ResourceProfiles --- .../scheduler/TaskSchedulerImplSuite.scala | 58 +++++++++++++++++++ 1 file changed, 58 insertions(+) diff --git a/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala index 85ea4f582e37c..439e465ec64b3 100644 --- a/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala +++ b/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala @@ -1835,6 +1835,64 @@ class TaskSchedulerImplSuite extends SparkFunSuite with LocalSparkContext with B assert(2 == taskDescriptions.head.resources(GPU).addresses.size) } + test("Scheduler works with reusable executors for multiple ResourceProfiles") { + val taskCpus = 1 + val executorCpus = 4 + + val taskScheduler = setupScheduler(numCores = executorCpus, + config.EXECUTOR_CORES.key -> executorCpus.toString, + config.CPUS_PER_TASK.key -> taskCpus.toString) + + val ereqs = new ExecutorResourceRequests().cores(4) + val treqs = new TaskResourceRequests().cpus(2) + val rp = new ResourceProfile(ereqs.requests, treqs.requests) + taskScheduler.sc.resourceProfileManager.addResourceProfile(rp) + val taskSet = FakeTask.createTaskSet(3) + val rpTaskSet = FakeTask.createTaskSet(5, stageId = 1, stageAttemptId = 0, + priority = 0, rpId = rp.id) + + val resourcesDefaultProf = Map(GPU -> ArrayBuffer("0", "1", "2", "3")) + val resources = Map(GPU -> ArrayBuffer("4", "5", "6", "7", "8", "9")) + + val workerOffers = + IndexedSeq(new WorkerOffer("executor0", "host0", 2, None, resourcesDefaultProf), + new WorkerOffer("executor1", "host1", 6, None, resources, rp.id)) + taskScheduler.submitTasks(taskSet) + taskScheduler.submitTasks(rpTaskSet) + // should have 2 for default profile and 2 for additional resource profile + var taskDescriptions = taskScheduler.resourceOffers(workerOffers).flatten + + logDebug(taskDescriptions.mkString("\n")) + +// assert(5 === taskDescriptions.length) +// var has2Gpus = 0 +// var has1Gpu = 0 +// for (tDesc <- taskDescriptions) { +// assert(tDesc.resources.contains(GPU)) +// if (tDesc.resources(GPU).addresses.size == 2) { +// has2Gpus += 1 +// } +// if (tDesc.resources(GPU).addresses.size == 1) { +// has1Gpu += 1 +// } +// } +// assert(has2Gpus == 3) +// assert(has1Gpu == 2) +// +// val resources3 = Map(GPU -> ArrayBuffer("14", "15", "16", "17", "18", "19")) +// +// // clear the first 2 worker offers so they don't have any room and add a third +// // for the resource profile +// val workerOffers3 = IndexedSeq( +// new WorkerOffer("executor0", "host0", 0, None, Map.empty), +// new WorkerOffer("executor1", "host1", 0, None, Map.empty, rp.id), +// new WorkerOffer("executor2", "host2", 6, None, resources3, rp.id)) +// taskDescriptions = taskScheduler.resourceOffers(workerOffers3).flatten +// assert(2 === taskDescriptions.length) +// assert(taskDescriptions.head.resources.contains(GPU)) +// assert(2 == taskDescriptions.head.resources(GPU).addresses.size) + } + private def setupSchedulerForDecommissionTests(clock: Clock, numTasks: Int): TaskSchedulerImpl = { // one task per host val numHosts = numTasks From 05ad5e91625ae4a89d8f7547b9be25698f567e60 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Mon, 6 Sep 2021 19:11:05 +0800 Subject: [PATCH 05/26] Add test Scheduler works with reusable executors for multiple ResourceProfiles --- .../scheduler/TaskSchedulerImplSuite.scala | 77 ++++++++++++------- 1 file changed, 51 insertions(+), 26 deletions(-) diff --git a/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala index 439e465ec64b3..7f13b5e5ba719 100644 --- a/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala +++ b/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala @@ -1836,33 +1836,58 @@ class TaskSchedulerImplSuite extends SparkFunSuite with LocalSparkContext with B } test("Scheduler works with reusable executors for multiple ResourceProfiles") { - val taskCpus = 1 - val executorCpus = 4 - - val taskScheduler = setupScheduler(numCores = executorCpus, - config.EXECUTOR_CORES.key -> executorCpus.toString, - config.CPUS_PER_TASK.key -> taskCpus.toString) - - val ereqs = new ExecutorResourceRequests().cores(4) - val treqs = new TaskResourceRequests().cpus(2) - val rp = new ResourceProfile(ereqs.requests, treqs.requests) - taskScheduler.sc.resourceProfileManager.addResourceProfile(rp) - val taskSet = FakeTask.createTaskSet(3) - val rpTaskSet = FakeTask.createTaskSet(5, stageId = 1, stageAttemptId = 0, - priority = 0, rpId = rp.id) - - val resourcesDefaultProf = Map(GPU -> ArrayBuffer("0", "1", "2", "3")) - val resources = Map(GPU -> ArrayBuffer("4", "5", "6", "7", "8", "9")) - - val workerOffers = - IndexedSeq(new WorkerOffer("executor0", "host0", 2, None, resourcesDefaultProf), - new WorkerOffer("executor1", "host1", 6, None, resources, rp.id)) + val defaultExecutorCpus = 4 + val defaultTaskCpus = 1 + + val taskScheduler = setupScheduler(numCores = defaultExecutorCpus, + config.EXECUTOR_CORES.key -> defaultExecutorCpus.toString, + config.CPUS_PER_TASK.key -> defaultTaskCpus.toString, + config.DYN_ALLOCATION_REUSE_EXECUTORS.key -> true.toString) + + val ereqs1 = new ExecutorResourceRequests().cores(4) + val treqs1 = new TaskResourceRequests().cpus(2) + val rpCompatible = new ResourceProfile(ereqs1.requests, treqs1.requests) + taskScheduler.sc.resourceProfileManager.addResourceProfile(rpCompatible) + + val ereqs2 = new ExecutorResourceRequests().cores(6) + val treqs2 = new TaskResourceRequests().cpus(2) + val rpIncompatible = new ResourceProfile(ereqs2.requests, treqs2.requests) + taskScheduler.sc.resourceProfileManager.addResourceProfile(rpIncompatible) + + val taskSet = FakeTask.createTaskSet(8) + val rpTaskSetCompatible = FakeTask.createTaskSet(4, stageId = 1, stageAttemptId = 0, + priority = 0, rpId = rpCompatible.id) + val rpTaskSetInCompatible = FakeTask.createTaskSet(4, stageId = 2, stageAttemptId = 0, + priority = 0, rpId = rpIncompatible.id) + + val workerOffersCompatible = + IndexedSeq(new WorkerOffer("executor0", "host0", 4, None), + new WorkerOffer("executor1", "host1", 4, None, Map.empty, rpCompatible.id)) + + val workerOffersIncompatible = + IndexedSeq(new WorkerOffer("executor0", "host0", 4, None), + new WorkerOffer("executor1", "host1", 6, None, Map.empty, rpIncompatible.id)) + + logDebug("8 tasks will be allocated to executor0 and executor1") taskScheduler.submitTasks(taskSet) - taskScheduler.submitTasks(rpTaskSet) - // should have 2 for default profile and 2 for additional resource profile - var taskDescriptions = taskScheduler.resourceOffers(workerOffers).flatten - - logDebug(taskDescriptions.mkString("\n")) + var taskDescriptions = taskScheduler.resourceOffers(workerOffersCompatible).flatten + logDebug("4 tasks will be allocated executor0 since executor1 is incompatible") + taskScheduler.submitTasks(taskSet) + taskDescriptions = taskScheduler.resourceOffers(workerOffersIncompatible).flatten + +// assert(8 === taskDescriptions.length) +// assert(taskDescriptions.exists(_.executorId == "executor0") +// && taskDescriptions.exists(_.executorId == "executor1")) + + logDebug("4 tasks will be allocated to both executor0 and executor1") + taskScheduler.submitTasks(rpTaskSetCompatible) + taskDescriptions = taskScheduler.resourceOffers(workerOffersCompatible).flatten + logDebug("3 tasks will be allocated executor1 only since executor0 is incompatible") + taskScheduler.submitTasks(rpTaskSetInCompatible) + taskDescriptions = taskScheduler.resourceOffers(workerOffersIncompatible).flatten +// assert(4 === taskDescriptions.length) +// assert(taskDescriptions.exists(_.executorId == "executor0") +// && taskDescriptions.exists(_.executorId == "executor1")) // assert(5 === taskDescriptions.length) // var has2Gpus = 0 From 9b5b604e79ef19a820b2eb10cd9816fd40d7a684 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Tue, 7 Sep 2021 16:05:20 +0800 Subject: [PATCH 06/26] revise resourcesCompatible to consider custom resources --- .../org/apache/spark/resource/ResourceProfile.scala | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala index fe2e14e4507c3..031a750706a98 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala @@ -29,6 +29,7 @@ import org.apache.spark.annotation.{Evolving, Since} import org.apache.spark.internal.Logging import org.apache.spark.internal.config._ import org.apache.spark.internal.config.Python.PYSPARK_EXECUTOR_MEMORY +import org.apache.spark.resource.ResourceProfile.hasCustomExecutorResources import org.apache.spark.util.Utils /** @@ -241,10 +242,11 @@ class ResourceProfile( rp.taskResources == taskResources && rp.executorResources == executorResources } - // TODO: define what a compatible executor is, right now is equal cores - // check that executor resources are equal, but the task resources could be different + // check that executor cores resources are equal, but the task resources could be different + // custom resources are not supported right now private[spark] def resourcesCompatible(rp: ResourceProfile): Boolean = { - rp.executorResources("cores") == executorResources("cores") + !hasCustomExecutorResources(this) && !hasCustomExecutorResources(rp) && + rp.executorResources("cores") == executorResources("cores") } override def hashCode(): Int = Seq(taskResources, executorResources).hashCode() @@ -379,6 +381,11 @@ object ResourceProfile extends Logging { } } + private[spark] def hasCustomExecutorResources(rp: ResourceProfile): Boolean = { + rp.executorResources.keys.exists( + k => !ResourceProfile.allSupportedExecutorResources.contains(k)) + } + private[spark] def getCustomTaskResources( rp: ResourceProfile): Map[String, TaskResourceRequest] = { rp.taskResources.filterKeys(k => !k.equals(ResourceProfile.CPUS)).toMap From 9b270880201828f62d3f5611a7fee331a374b9ce Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Tue, 7 Sep 2021 16:06:47 +0800 Subject: [PATCH 07/26] update ExecutorAllocationManager & test --- .../spark/ExecutorAllocationManager.scala | 70 ++++++------ .../resource/ResourceProfileManager.scala | 22 ++-- .../ExecutorAllocationManagerSuite.scala | 108 +++++++----------- 3 files changed, 84 insertions(+), 116 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala index e44d3691c57c5..b7f992db7a54c 100644 --- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala +++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala @@ -299,21 +299,10 @@ 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 numOtherCompatibleExecutors = executorCountWithCompatibleResourceProfile(rpId) - // if reusing executors, should subtract number of compatible executors - val maxNeeded = if (reuseExecutors) { - math.ceil(numRunningOrPendingTasks * executorAllocationRatio / - tasksPerExecutor).toInt - numOtherCompatibleExecutors - } else { - math.ceil(numRunningOrPendingTasks * executorAllocationRatio / - tasksPerExecutor).toInt - } - logDebug(s"$numOtherCompatibleExecutors compatible executors and " + - s"$maxNeeded max needed for rpId: $rpId numpending: $numRunningOrPendingTasks," + + logDebug(s"max needed for rpId: $rpId numpending: $numRunningOrPendingTasks," + s" tasksperexecutor: $tasksPerExecutor") + val maxNeeded = math.ceil(numRunningOrPendingTasks * executorAllocationRatio / + tasksPerExecutor).toInt val maxNeededWithSpeculationLocalityOffset = if (tasksPerExecutor > 1 && maxNeeded == 1 && pendingSpeculative > 0) { @@ -503,12 +492,20 @@ private[spark] class ExecutorAllocationManager( numExecutorsTargetPerResourceProfileId(rpId) - oldNumExecutorsTarget } - private def executorCountWithCompatibleResourceProfile(rpId: Int): Int = { +// private def executorCountWithCompatibleResourceProfile(rpId: Int): Int = { +// val compatibleProfileIds = resourceProfileManager.getOtherCompatibleProfileIds(rpId) +// +// logDebug("compatibleProfileIds for rpId " + rpId + ": " + compatibleProfileIds.mkString(" ")) +// +// compatibleProfileIds.map { executorMonitor.executorCountWithResourceProfile(_) }.sum +// } + + private def numExecutorsTargetsCompatibleProfiles(rpId: Int): Int = { val compatibleProfileIds = resourceProfileManager.getOtherCompatibleProfileIds(rpId) - logDebug("compatibleProfileIds for rpId " + rpId + ": " + compatibleProfileIds.mkString(" ")) +// logDebug("compatibleProfileIds for rpId " + rpId + ": " + compatibleProfileIds.mkString(" ")) - compatibleProfileIds.map { executorMonitor.executorCountWithResourceProfile(_) }.sum + compatibleProfileIds.map { numExecutorsTargetPerResourceProfileId(_) }.sum } /** @@ -523,11 +520,6 @@ private[spark] class ExecutorAllocationManager( */ private def addExecutors(maxNumExecutorsNeeded: Int, rpId: Int): Int = { val oldNumExecutorsTarget = numExecutorsTargetPerResourceProfileId(rpId) - logDebug(s"rpId " + rpId + ": executorCount, " + - executorMonitor.executorCountWithResourceProfile(rpId) + - ", compatible executorCount: " + - executorCountWithCompatibleResourceProfile(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) { @@ -538,28 +530,34 @@ private[spark] class ExecutorAllocationManager( } // There's no point in wasting time ramping up to the number of executors we already have, so // make sure our target is at least as much as our current allocation: - var numExecutorsTarget = math.max(numExecutorsTargetPerResourceProfileId(rpId), - executorMonitor.executorCountWithResourceProfile(rpId)) - -// var numExecutorsTarget = if (reuseExecutors) { -// logDebug("Using number of compatible executors as target") -// math.max(numExecutorsTargetPerResourceProfileId(rpId), -// executorCountWithCompatibleResourceProfile(rpId)) -// } else { -// math.max(numExecutorsTargetPerResourceProfileId(rpId), -// executorMonitor.executorCountWithResourceProfile(rpId)) -// } - + executorMonitor.executorCountWithResourceProfile(rpId)) // Boost our target with the number to add for this round: 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) + // Adjust min and max num of executors due to the compatible executors + val adjustedMaxNumExecutors = if (reuseExecutors) { + math.max(1, maxNumExecutors - numExecutorsTargetsCompatibleProfiles(rpId)) + } else { + maxNumExecutors + } + val adjustedMinNumExecutors = if (reuseExecutors) { + math.max(1, minNumExecutors - numExecutorsTargetsCompatibleProfiles(rpId)) + } else { + minNumExecutors + } + + // Ensure that our target fits within adjusted bounds: + numExecutorsTarget = math.max(math.min(numExecutorsTarget, adjustedMaxNumExecutors), + adjustedMinNumExecutors) + val delta = numExecutorsTarget - oldNumExecutorsTarget numExecutorsTargetPerResourceProfileId(rpId) = numExecutorsTarget + logDebug("new numExecutorsTargetPerResourceProfileId: " + + numExecutorsTargetPerResourceProfileId.mkString(", ")) + // If our target has not changed, do not send a message // to the cluster manager and reset our exponential growth if (delta == 0) { diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala index 54b873e02f8d7..22bb6b9e5ee77 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala @@ -131,16 +131,16 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, /* * Get all compatible profiles including itself */ - def getCompatibleProfiles(rp: ResourceProfile): Map[Int, ResourceProfile] = { - readLock.lock() - try { - resourceProfileIdToResourceProfile.filter { case (_, rpEntry) => - rpEntry.resourcesCompatible(rp) - }.toMap - } finally { - readLock.unlock() - } - } +// def getCompatibleProfiles(rp: ResourceProfile): Map[Int, ResourceProfile] = { +// readLock.lock() +// try { +// resourceProfileIdToResourceProfile.filter { case (_, rpEntry) => +// rpEntry.resourcesCompatible(rp) +// }.toMap +// } finally { +// readLock.unlock() +// } +// } /* * Get all compatible profile IDs excluding itself @@ -176,7 +176,7 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, readLock.lock() try { resourceProfileIdToResourceProfile.foreach { case (id: Int, rpEntry: ResourceProfile) => - println("profile " + id + ": " + rpEntry) + logDebug("rpId " + id + ": " + rpEntry) } } finally { readLock.unlock() diff --git a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala index 63c4c3757b449..87564e1f3960a 100644 --- a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala +++ b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala @@ -216,6 +216,7 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite { assert(numExecutorsTarget(manager, rprof1.id) === 8) assert(addExecutorsToTarget(manager, updatesNeeded, rprof1) === 2) doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) + assert(numExecutorsTargetForDefaultProfileId(manager) === 10) assert(numExecutorsToAddForDefaultProfile(manager) === 1) assert(addExecutorsToTargetForDefaultProfile(manager, updatesNeeded) === 0) @@ -286,17 +287,8 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite { assert(numExecutorsTarget(manager, rprof1.id) === 2) } - test("reuse executors") { - // scalastyle:off println - + test("reuse executors multiple profiles") { val manager = createManager(createConfReuseExecutors(1, 10, 1)) -// rpManager.dumpResourceProfile() - -// onExecutorAddedDefaultProfile(manager, "first") -// onExecutorAddedDefaultProfile(manager, "second") -// onExecutorAddedDefaultProfile(manager, "third") - - println("--- SparkListenerStageSubmitted defaultProfile ---") post(SparkListenerStageSubmitted(createStageInfo(0, 1000, rp = defaultProfile))) val rp1 = new ResourceProfileBuilder() val execReqs = new ExecutorResourceRequests().cores(4) @@ -304,15 +296,8 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite { rp1.require(execReqs).require(taskReqs) val rprof1 = rp1.build rpManager.addResourceProfile(rprof1) -// rpManager.dumpResourceProfile() - println("--- SparkListenerStageSubmitted rprof1 ---") + rpManager.dumpResourceProfile() post(SparkListenerStageSubmitted(createStageInfo(1, 1000, rp = rprof1))) - -// onExecutorAdded(manager, "firstrp1", rprof1) -// onExecutorAdded(manager, "secondrp1", rprof1) -// onExecutorAdded(manager, "thirdrp1", rprof1) -// onExecutorAdded(manager, "fourthrp1", rprof1) - val updatesNeeded = new mutable.HashMap[ResourceProfile, ExecutorAllocationManager.TargetNumUpdates] @@ -322,86 +307,71 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite { assert(addExecutorsToTargetForDefaultProfile(manager, updatesNeeded) === 1) assert(numExecutorsToAdd(manager, rprof1) === 1) assert(numExecutorsTarget(manager, rprof1.id) === 1) - println("--- addExecutorsToTarget ---") assert(addExecutorsToTarget(manager, updatesNeeded, rprof1) === 1) - println("--- doUpdateRequest ---") doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) + assert(numExecutorsTargetForDefaultProfileId(manager) === 2) assert(numExecutorsToAddForDefaultProfile(manager) === 2) assert(addExecutorsToTargetForDefaultProfile(manager, updatesNeeded) === 2) assert(numExecutorsToAdd(manager, rprof1) === 2) assert(numExecutorsTarget(manager, rprof1.id) === 2) - println("--- addExecutorsToTarget ---") assert(addExecutorsToTarget(manager, updatesNeeded, rprof1) === 2) - println("--- doUpdateRequest ---") doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) + assert(numExecutorsTargetForDefaultProfileId(manager) === 4) assert(numExecutorsToAddForDefaultProfile(manager) === 4) - assert(addExecutorsToTargetForDefaultProfile(manager, updatesNeeded) === 4) + assert(addExecutorsToTargetForDefaultProfile(manager, updatesNeeded) === 2) assert(numExecutorsToAdd(manager, rprof1) === 4) assert(numExecutorsTarget(manager, rprof1.id) === 4) - assert(addExecutorsToTarget(manager, updatesNeeded, rprof1) === 4) - doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) - assert(numExecutorsTargetForDefaultProfileId(manager) === 8) - assert(numExecutorsToAddForDefaultProfile(manager) === 8) - // reached the limit of 10 - assert(addExecutorsToTargetForDefaultProfile(manager, updatesNeeded) === 2) - assert(numExecutorsToAdd(manager, rprof1) === 8) - assert(numExecutorsTarget(manager, rprof1.id) === 8) - assert(addExecutorsToTarget(manager, updatesNeeded, rprof1) === 2) + // reached the limit of 10 (compatible executors are shared), no more updates + // numExecutorsTargetPerResourceProfileId: 0 -> 6, 2 -> 4 + assert(addExecutorsToTarget(manager, updatesNeeded, rprof1) === 0) doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) - assert(numExecutorsTargetForDefaultProfileId(manager) === 10) + + assert(numExecutorsTargetForDefaultProfileId(manager) === 6) assert(numExecutorsToAddForDefaultProfile(manager) === 1) assert(addExecutorsToTargetForDefaultProfile(manager, updatesNeeded) === 0) assert(numExecutorsToAdd(manager, rprof1) === 1) - assert(numExecutorsTarget(manager, rprof1.id) === 10) + assert(numExecutorsTarget(manager, rprof1.id) === 4) assert(addExecutorsToTarget(manager, updatesNeeded, rprof1) === 0) doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) - assert(numExecutorsTargetForDefaultProfileId(manager) === 10) - assert(numExecutorsToAddForDefaultProfile(manager) === 1) - assert(numExecutorsToAdd(manager, rprof1) === 1) - assert(numExecutorsTarget(manager, rprof1.id) === 10) - - println("--- Register previously requested executors ---") + // Register previously requested executors onExecutorAddedDefaultProfile(manager, "first") + onExecutorAdded(manager, "firstrp1", rprof1) + assert(numExecutorsTargetForDefaultProfileId(manager) === 6) + assert(numExecutorsTarget(manager, rprof1.id) === 4) onExecutorAddedDefaultProfile(manager, "second") - logDebug(numExecutorsTargetForDefaultProfileId(manager).toString) - logDebug(numExecutorsToAddForDefaultProfile(manager).toString) onExecutorAddedDefaultProfile(manager, "third") onExecutorAddedDefaultProfile(manager, "fourth") - - logDebug("numExecutorsTargetForDefaultProfileId: " + - numExecutorsTargetForDefaultProfileId(manager).toString) - logDebug("numExecutorsToAddForDefaultProfile: " + - numExecutorsToAddForDefaultProfile(manager).toString) - logDebug("addExecutorsToTargetForDefaultProfile: " + - addExecutorsToTargetForDefaultProfile(manager, updatesNeeded).toString) - doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) - - onExecutorAdded(manager, "firstrp1", rprof1) onExecutorAdded(manager, "secondrp1", rprof1) - - logDebug("numExecutorsTargetForDefaultProfileId: " + - numExecutorsTargetForDefaultProfileId(manager).toString) - logDebug("numExecutorsToAddForDefaultProfile: " + - numExecutorsToAddForDefaultProfile(manager).toString) - logDebug("addExecutorsToTargetForDefaultProfile: " + - addExecutorsToTargetForDefaultProfile(manager, updatesNeeded).toString) - doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) - onExecutorAdded(manager, "thirdrp1", rprof1) onExecutorAdded(manager, "fourthrp1", rprof1) + assert(numExecutorsTargetForDefaultProfileId(manager) === 6) + assert(numExecutorsTarget(manager, rprof1.id) === 4) + onExecutorAddedDefaultProfile(manager, "first") // duplicates should not count + onExecutorAddedDefaultProfile(manager, "second") + onExecutorAdded(manager, "firstrp1", rprof1) + onExecutorAdded(manager, "secondrp1", rprof1) + assert(numExecutorsTargetForDefaultProfileId(manager) === 6) + assert(numExecutorsTarget(manager, rprof1.id) === 4) - logDebug("numExecutorsTargetForDefaultProfileId: " + - numExecutorsTargetForDefaultProfileId(manager).toString) - logDebug("numExecutorsToAddForDefaultProfile: " + - numExecutorsToAddForDefaultProfile(manager).toString) - logDebug("addExecutorsToTargetForDefaultProfile: " + - addExecutorsToTargetForDefaultProfile(manager, updatesNeeded).toString) + // Try adding again + // This should still fail because the number pending + running is still at the limit + assert(addExecutorsToTargetForDefaultProfile(manager, updatesNeeded) === 0) + assert(addExecutorsToTarget(manager, updatesNeeded, rprof1) === 0) doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) - - // scalastyle:on println + assert(numExecutorsTargetForDefaultProfileId(manager) === 6) + assert(numExecutorsToAddForDefaultProfile(manager) === 1) + assert(numExecutorsToAdd(manager, rprof1) === 1) + assert(numExecutorsTarget(manager, rprof1.id) === 4) + assert(addExecutorsToTargetForDefaultProfile(manager, updatesNeeded) === 0) + assert(addExecutorsToTarget(manager, updatesNeeded, rprof1) === 0) + doUpdateRequest(manager, updatesNeeded.toMap, clock.getTimeMillis()) + assert(numExecutorsTargetForDefaultProfileId(manager) === 6) + assert(numExecutorsToAddForDefaultProfile(manager) === 1) + assert(numExecutorsToAdd(manager, rprof1) === 1) + assert(numExecutorsTarget(manager, rprof1.id) === 4) } test("remove executors multiple profiles") { From 540c8e2287d8f5e3714c6f5599a3fe86804b6f50 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Wed, 15 Sep 2021 20:16:20 +0800 Subject: [PATCH 08/26] fix adjustedMinNumExecutors & adjustedMaxNumExecutors --- .../spark/ExecutorAllocationManager.scala | 63 ++++++++++++------- .../ExecutorAllocationManagerSuite.scala | 2 +- 2 files changed, 42 insertions(+), 23 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala index b7f992db7a54c..3a19c33297fcc 100644 --- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala +++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala @@ -299,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") + val maxNeededWithSpeculationLocalityOffset = if (tasksPerExecutor > 1 && maxNeeded == 1 && pendingSpeculative > 0) { // If we have pending speculative tasks and only need a single executor, allocate one more @@ -520,14 +522,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)") + numExecutorsToAddPerResourceProfileId(rpId) = 1 + return 0 + } } + // There's no point in wasting time ramping up to the number of executors we already have, so // make sure our target is at least as much as our current allocation: var numExecutorsTarget = math.max(numExecutorsTargetPerResourceProfileId(rpId), @@ -536,21 +551,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) - // Adjust min and max num of executors due to the compatible executors - val adjustedMaxNumExecutors = if (reuseExecutors) { - math.max(1, maxNumExecutors - numExecutorsTargetsCompatibleProfiles(rpId)) - } else { - maxNumExecutors - } - val adjustedMinNumExecutors = if (reuseExecutors) { - math.max(1, minNumExecutors - numExecutorsTargetsCompatibleProfiles(rpId)) + numExecutorsTarget = if (!reuseExecutors) { + // Ensure that our target fits within configured bounds: + math.max(math.min(numExecutorsTarget, maxNumExecutors), minNumExecutors) } else { - minNumExecutors - } + // 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) + + // scalastyle:off println + println(rpId, minNumExecutors, maxNumExecutors, numCompatibleExecutors) + // scalastyle:on println - // Ensure that our target fits within adjusted bounds: - numExecutorsTarget = math.max(math.min(numExecutorsTarget, adjustedMaxNumExecutors), - adjustedMinNumExecutors) + // assert(adjustedMinNumExecutors >= 0, s"$adjustedMinNumExecutors") + // assert(adjustedMaxNumExecutors > 0, s"$adjustedMaxNumExecutors") + + math.max(math.min(numExecutorsTarget, adjustedMaxNumExecutors), + adjustedMinNumExecutors) + } val delta = numExecutorsTarget - oldNumExecutorsTarget numExecutorsTargetPerResourceProfileId(rpId) = numExecutorsTarget diff --git a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala index 87564e1f3960a..02e8bce841854 100644 --- a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala +++ b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala @@ -119,7 +119,7 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite { } test("add executors default profile") { - val manager = createManager(createConfReuseExecutors(1, 10, 1)) + val manager = createManager(createConf(1, 10, 1)) post(SparkListenerStageSubmitted(createStageInfo(0, 1000))) val updatesNeeded = From b28b295717d58426288db61465d8fe128feefd18 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Thu, 16 Sep 2021 21:55:22 +0800 Subject: [PATCH 09/26] update initialNumExecutors for compatible executors --- .../org/apache/spark/ExecutorAllocationManager.scala | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala index 3a19c33297fcc..d3d4263f68afb 100644 --- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala +++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala @@ -766,7 +766,13 @@ private[spark] class ExecutorAllocationManager( if (!numExecutorsTargetPerResourceProfileId.contains(profId)) { numExecutorsTargetPerResourceProfileId.put(profId, initialNumExecutors) - if (initialNumExecutors > 0) { + val adjustedinitialNumExecutors = if (reuseExecutors) { + val numCompatibleExecutors = numExecutorsTargetsCompatibleProfiles(profId) + math.max(0, initialNumExecutors - numCompatibleExecutors) + } else { + initialNumExecutors + } + if (adjustedinitialNumExecutors > 0) { logDebug(s"requesting executors, rpId: $profId, initial number is $initialNumExecutors") // we need to trigger a schedule since we add an initial number here. client.requestTotalExecutors( From c1355ea5017913fa82947d0cbaa3c64b2c70a3aa Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Tue, 11 Jan 2022 11:18:57 +0800 Subject: [PATCH 10/26] rename option to spark.scheduler.reuseCompatibleExecutors --- .../org/apache/spark/ExecutorAllocationManager.scala | 2 +- .../org/apache/spark/internal/config/package.scala | 12 ++++++------ .../apache/spark/scheduler/TaskSchedulerImpl.scala | 2 +- .../spark/ExecutorAllocationManagerSuite.scala | 2 +- .../spark/scheduler/TaskSchedulerImplSuite.scala | 2 +- 5 files changed, 10 insertions(+), 10 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala index d3d4263f68afb..34c3117d8b8bb 100644 --- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala +++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala @@ -110,7 +110,7 @@ private[spark] class ExecutorAllocationManager( import ExecutorAllocationManager._ - private val reuseExecutors = conf.get(DYN_ALLOCATION_REUSE_EXECUTORS) + private val reuseExecutors = conf.get(SCHEDULER_REUSE_COMPATIBLE_EXECUTORS) // Lower and upper bounds on the number of executors. private val minNumExecutors = conf.get(DYN_ALLOCATION_MIN_EXECUTORS) diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index bd7e714fd729e..8e91faaac1d3c 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -572,12 +572,6 @@ package object config { .booleanConf .createWithDefault(false) - private[spark] val DYN_ALLOCATION_REUSE_EXECUTORS = - ConfigBuilder("spark.dynamicAllocation.reuseExecutors") - .version("1.2.0") - .booleanConf - .createWithDefault(false) - private[spark] val DYN_ALLOCATION_TESTING = ConfigBuilder("spark.dynamicAllocation.testing") .version("1.2.0") @@ -1964,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") + .booleanConf + .createWithDefault(false) + private[spark] val SPECULATION_ENABLED = ConfigBuilder("spark.speculation") .version("0.6.0") diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala index bfc60568b39f4..7b73bce47d98c 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala @@ -103,7 +103,7 @@ private[spark] class TaskSchedulerImpl( // of tasks that are very short. val MIN_TIME_TO_SPECULATION = conf.get(SPECULATION_MIN_THRESHOLD) - private val reuseExecutors = conf.get(DYN_ALLOCATION_REUSE_EXECUTORS) + private val reuseExecutors = conf.get(SCHEDULER_REUSE_COMPATIBLE_EXECUTORS) private val speculationScheduler = ThreadUtils.newDaemonSingleThreadScheduledExecutor("task-scheduler-speculation") diff --git a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala index 02e8bce841854..5810d01af9b06 100644 --- a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala +++ b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala @@ -1813,7 +1813,7 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite { decommissioningEnabled: Boolean = false): SparkConf = { val sparkConf = new SparkConf() .set(config.DYN_ALLOCATION_ENABLED, true) - .set(config.DYN_ALLOCATION_REUSE_EXECUTORS, true) + .set(config.SCHEDULER_REUSE_COMPATIBLE_EXECUTORS, true) .set(config.DYN_ALLOCATION_MIN_EXECUTORS, minExecutors) .set(config.DYN_ALLOCATION_MAX_EXECUTORS, maxExecutors) .set(config.DYN_ALLOCATION_INITIAL_EXECUTORS, initialExecutors) diff --git a/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala index 7f13b5e5ba719..ffb70e2c072be 100644 --- a/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala +++ b/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala @@ -1842,7 +1842,7 @@ class TaskSchedulerImplSuite extends SparkFunSuite with LocalSparkContext with B val taskScheduler = setupScheduler(numCores = defaultExecutorCpus, config.EXECUTOR_CORES.key -> defaultExecutorCpus.toString, config.CPUS_PER_TASK.key -> defaultTaskCpus.toString, - config.DYN_ALLOCATION_REUSE_EXECUTORS.key -> true.toString) + config.SCHEDULER_REUSE_COMPATIBLE_EXECUTORS.key -> true.toString) val ereqs1 = new ExecutorResourceRequests().cores(4) val treqs1 = new TaskResourceRequests().cpus(2) From 5af3390e5abd300574b84f836c1b3da649771979 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Tue, 11 Jan 2022 21:47:13 +0800 Subject: [PATCH 11/26] add cache compatible profiles --- .../spark/resource/ResourceProfile.scala | 18 +++++++++++ .../resource/ResourceProfileManager.scala | 31 ++++++++++++++++--- 2 files changed, 45 insertions(+), 4 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala index 031a750706a98..559e6b5c03b81 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala @@ -249,6 +249,24 @@ class ResourceProfile( rp.executorResources("cores") == executorResources("cores") } + private[spark] def resourcesCompatible(rp: ResourceProfile, + isCompatibleFunc: (ResourceProfile, ResourceProfile) => Boolean = isCompatibleExecutorsDefault) + : Boolean = { + isCompatibleFunc(this, rp) + } + + private def isCompatibleExecutorsDefault(prevRP: ResourceProfile, + curRP: ResourceProfile ): Boolean = { + !hasCustomExecutorResources(prevRP) && !hasCustomExecutorResources(curRP) && + prevRP.executorResources("cores") == curRP.executorResources("cores") + } + + private def isCompatibleExecutorsLarger(prevRP: ResourceProfile, + curRP: ResourceProfile ): Boolean = { + !hasCustomExecutorResources(prevRP) && !hasCustomExecutorResources(curRP) && + prevRP.executorResources("cores") == curRP.executorResources("cores") + } + override def hashCode(): Int = Seq(taskResources, executorResources).hashCode() override def toString(): String = { diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala index 22bb6b9e5ee77..17b11d053ed62 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala @@ -39,6 +39,7 @@ import org.apache.spark.util.Utils.isTesting private[spark] class ResourceProfileManager(sparkConf: SparkConf, listenerBus: LiveListenerBus) extends Logging { private val resourceProfileIdToResourceProfile = new HashMap[Int, ResourceProfile]() + private val resourceProfileIdToCompatibleResourceProfileIds = new HashMap[Int, Set[Int]]() private val (readLock, writeLock) = { val lock = new ReentrantReadWriteLock() @@ -84,6 +85,7 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, try { if (!resourceProfileIdToResourceProfile.contains(rp.id)) { val prev = resourceProfileIdToResourceProfile.put(rp.id, rp) + computeCompatibleProfileIds() if (prev.isEmpty) putNewProfile = true } } finally { @@ -146,13 +148,34 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, * Get all compatible profile IDs excluding itself */ def getOtherCompatibleProfileIds(rpId: Int): Set[Int] = { - val resourceProfile = resourceProfileFromId(rpId) +// val resourceProfile = resourceProfileFromId(rpId) +// readLock.lock() +// try { +// resourceProfileIdToResourceProfile.filter { case (id: Int, rpEntry: ResourceProfile) => +// id != rpId && rpEntry.resourcesCompatible(resourceProfile) +// }.map(_._1).toSet +// } finally { +// readLock.unlock() +// } + resourceProfileIdToCompatibleResourceProfileIds.getOrElse(rpId, null) + } + + def computeCompatibleProfileIds(): Unit = { readLock.lock() try { - resourceProfileIdToResourceProfile.filter { case (id: Int, rpEntry: ResourceProfile) => - id != rpId && rpEntry.resourcesCompatible(resourceProfile) - }.map(_._1).toSet + resourceProfileIdToResourceProfile.foreach { + case (id: Int, rpEntry: ResourceProfile) => + val resourceProfile = resourceProfileFromId(id) + val compatibleIdSet = resourceProfileIdToResourceProfile.filter { + case (otherId: Int, _: ResourceProfile) => + id != otherId && rpEntry.resourcesCompatible(resourceProfile) + }.map(_._1).toSet + writeLock.lock() + resourceProfileIdToCompatibleResourceProfileIds.put(id, compatibleIdSet) + writeLock.unlock() + } } finally { + writeLock.unlock() readLock.unlock() } } From 1c08b44a37b81600d3968b6d9e48ba52b3d87196 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Thu, 13 Jan 2022 22:26:04 +0800 Subject: [PATCH 12/26] cache compatible resource profiles when added --- .../resource/ResourceProfileManager.scala | 54 ++++--------------- 1 file changed, 10 insertions(+), 44 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala index 17b11d053ed62..d9a1652c0f8c7 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala @@ -24,6 +24,7 @@ import scala.collection.mutable.HashMap import org.apache.spark.{SparkConf, SparkException} import org.apache.spark.annotation.Evolving import org.apache.spark.internal.Logging +import org.apache.spark.internal.config.SCHEDULER_REUSE_COMPATIBLE_EXECUTORS import org.apache.spark.internal.config.Tests._ import org.apache.spark.scheduler.{LiveListenerBus, SparkListenerResourceProfileAdded} import org.apache.spark.util.Utils @@ -39,6 +40,8 @@ import org.apache.spark.util.Utils.isTesting private[spark] class ResourceProfileManager(sparkConf: SparkConf, listenerBus: LiveListenerBus) extends Logging { private val resourceProfileIdToResourceProfile = new HashMap[Int, ResourceProfile]() + + private val reuseExecutors = sparkConf.get(SCHEDULER_REUSE_COMPATIBLE_EXECUTORS) private val resourceProfileIdToCompatibleResourceProfileIds = new HashMap[Int, Set[Int]]() private val (readLock, writeLock) = { @@ -85,7 +88,9 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, try { if (!resourceProfileIdToResourceProfile.contains(rp.id)) { val prev = resourceProfileIdToResourceProfile.put(rp.id, rp) - computeCompatibleProfileIds() + if (reuseExecutors) { + computeCompatibleProfileIds() + } if (prev.isEmpty) putNewProfile = true } } finally { @@ -130,36 +135,13 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, } } - /* - * Get all compatible profiles including itself - */ -// def getCompatibleProfiles(rp: ResourceProfile): Map[Int, ResourceProfile] = { -// readLock.lock() -// try { -// resourceProfileIdToResourceProfile.filter { case (_, rpEntry) => -// rpEntry.resourcesCompatible(rp) -// }.toMap -// } finally { -// readLock.unlock() -// } -// } - - /* - * Get all compatible profile IDs excluding itself - */ def getOtherCompatibleProfileIds(rpId: Int): Set[Int] = { -// val resourceProfile = resourceProfileFromId(rpId) -// readLock.lock() -// try { -// resourceProfileIdToResourceProfile.filter { case (id: Int, rpEntry: ResourceProfile) => -// id != rpId && rpEntry.resourcesCompatible(resourceProfile) -// }.map(_._1).toSet -// } finally { -// readLock.unlock() -// } - resourceProfileIdToCompatibleResourceProfileIds.getOrElse(rpId, null) + resourceProfileIdToCompatibleResourceProfileIds.getOrElse(rpId, Set()) } + /* + * Compute and cache all compatible profile IDs excluding itself + */ def computeCompatibleProfileIds(): Unit = { readLock.lock() try { @@ -172,7 +154,6 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, }.map(_._1).toSet writeLock.lock() resourceProfileIdToCompatibleResourceProfileIds.put(id, compatibleIdSet) - writeLock.unlock() } } finally { writeLock.unlock() @@ -180,21 +161,6 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, } } - /* - * Get all compatible profile count including itself - */ -// def getCompatibleProfileCount(rpId: Int): Int = { -// val resourceProfile = resourceProfileFromId(rpId) -// readLock.lock() -// try { -// resourceProfileIdToResourceProfile.count { case (_, rpEntry) => -// rpEntry.resourcesCompatible(resourceProfile) -// } -// } finally { -// readLock.unlock() -// } -// } - def dumpResourceProfile(): Unit = { readLock.lock() try { From 9ec2615974ce50e6eebc65c26b58a654cb76c684 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Thu, 13 Jan 2022 22:27:06 +0800 Subject: [PATCH 13/26] add ResourceProfileCompatiblePolicy. todo refine. --- .../spark/resource/ResourceProfile.scala | 50 +++++++++++-------- 1 file changed, 30 insertions(+), 20 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala index 559e6b5c03b81..e6f02614cad3f 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala @@ -242,36 +242,46 @@ class ResourceProfile( rp.taskResources == taskResources && rp.executorResources == executorResources } - // check that executor cores resources are equal, but the task resources could be different - // custom resources are not supported right now - private[spark] def resourcesCompatible(rp: ResourceProfile): Boolean = { - !hasCustomExecutorResources(this) && !hasCustomExecutorResources(rp) && - rp.executorResources("cores") == executorResources("cores") - } - private[spark] def resourcesCompatible(rp: ResourceProfile, - isCompatibleFunc: (ResourceProfile, ResourceProfile) => Boolean = isCompatibleExecutorsDefault) + resourceProfileCompatibleFunc: (ResourceProfile, ResourceProfile) => Boolean + = ResourceProfileCompatiblePolicy.resourceProfileCompatibleWithEqualCores) : Boolean = { - isCompatibleFunc(this, rp) + resourceProfileCompatibleFunc(this, rp) } - private def isCompatibleExecutorsDefault(prevRP: ResourceProfile, - curRP: ResourceProfile ): Boolean = { - !hasCustomExecutorResources(prevRP) && !hasCustomExecutorResources(curRP) && - prevRP.executorResources("cores") == curRP.executorResources("cores") + override def hashCode(): Int = Seq(taskResources, executorResources).hashCode() + + override def toString(): String = { + s"Profile: id = ${_id}, executor resources: ${executorResources.mkString(",")}, " + + s"task resources: ${taskResources.mkString(",")}" } +} - private def isCompatibleExecutorsLarger(prevRP: ResourceProfile, - curRP: ResourceProfile ): Boolean = { +/** + * Resource profile compatibility based on user policy + * strict match: + * only reuse executors with exact same resources (including all 3rd party resources), + * there is no resource waste but less user flexibility + * 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 requirements. But in this policy user should know + * there is some resource waste. They need to tradeoff reuse executor or create new ones. + */ +object ResourceProfileCompatiblePolicy { + + def resourceProfileCompatibleWithEqualCores(prevRP: ResourceProfile, + curRP: ResourceProfile): Boolean = { !hasCustomExecutorResources(prevRP) && !hasCustomExecutorResources(curRP) && prevRP.executorResources("cores") == curRP.executorResources("cores") } - override def hashCode(): Int = Seq(taskResources, executorResources).hashCode() - - override def toString(): String = { - s"Profile: id = ${_id}, executor resources: ${executorResources.mkString(",")}, " + - s"task resources: ${taskResources.mkString(",")}" + def resourceProfileCompatibleWithMoreCores(prevRP: ResourceProfile, + curRP: ResourceProfile): Boolean = { + !hasCustomExecutorResources(prevRP) && !hasCustomExecutorResources(curRP) && + prevRP.executorResources("cores").amount > curRP.executorResources("cores").amount } } From ba9365245bd9aba64c78b85be02965fc841f5742 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Fri, 14 Jan 2022 12:51:04 +0800 Subject: [PATCH 14/26] refactor --- .../spark/ExecutorAllocationManager.scala | 20 +------- .../resource/ResourceProfileManager.scala | 51 +++++++++++-------- .../spark/scheduler/TaskSchedulerImpl.scala | 2 +- .../ExecutorAllocationManagerSuite.scala | 2 +- 4 files changed, 32 insertions(+), 43 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala index 34c3117d8b8bb..385a99af0ef2c 100644 --- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala +++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala @@ -494,19 +494,8 @@ private[spark] class ExecutorAllocationManager( numExecutorsTargetPerResourceProfileId(rpId) - oldNumExecutorsTarget } -// private def executorCountWithCompatibleResourceProfile(rpId: Int): Int = { -// val compatibleProfileIds = resourceProfileManager.getOtherCompatibleProfileIds(rpId) -// -// logDebug("compatibleProfileIds for rpId " + rpId + ": " + compatibleProfileIds.mkString(" ")) -// -// compatibleProfileIds.map { executorMonitor.executorCountWithResourceProfile(_) }.sum -// } - private def numExecutorsTargetsCompatibleProfiles(rpId: Int): Int = { - val compatibleProfileIds = resourceProfileManager.getOtherCompatibleProfileIds(rpId) - -// logDebug("compatibleProfileIds for rpId " + rpId + ": " + compatibleProfileIds.mkString(" ")) - + val compatibleProfileIds = resourceProfileManager.getCompatibleProfileIds(rpId) compatibleProfileIds.map { numExecutorsTargetPerResourceProfileId(_) }.sum } @@ -560,13 +549,6 @@ private[spark] class ExecutorAllocationManager( val adjustedMinNumExecutors = math.max(0, minNumExecutors - numCompatibleExecutors) val adjustedMaxNumExecutors = math.max(1, maxNumExecutors - numCompatibleExecutors) - // scalastyle:off println - println(rpId, minNumExecutors, maxNumExecutors, numCompatibleExecutors) - // scalastyle:on println - - // assert(adjustedMinNumExecutors >= 0, s"$adjustedMinNumExecutors") - // assert(adjustedMaxNumExecutors > 0, s"$adjustedMaxNumExecutors") - math.max(math.min(numExecutorsTarget, adjustedMaxNumExecutors), adjustedMinNumExecutors) } diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala index d9a1652c0f8c7..5e35d9dbc84c0 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala @@ -89,7 +89,7 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, if (!resourceProfileIdToResourceProfile.contains(rp.id)) { val prev = resourceProfileIdToResourceProfile.put(rp.id, rp) if (reuseExecutors) { - computeCompatibleProfileIds() + calculateCompatibleProfileIds() } if (prev.isEmpty) putNewProfile = true } @@ -135,37 +135,44 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, } } - def getOtherCompatibleProfileIds(rpId: Int): Set[Int] = { - resourceProfileIdToCompatibleResourceProfileIds.getOrElse(rpId, Set()) - } - - /* - * Compute and cache all compatible profile IDs excluding itself - */ - def computeCompatibleProfileIds(): Unit = { + private [spark] def getCompatibleProfileIds(rpId: Int): Set[Int] = { readLock.lock() try { - resourceProfileIdToResourceProfile.foreach { - case (id: Int, rpEntry: ResourceProfile) => - val resourceProfile = resourceProfileFromId(id) - val compatibleIdSet = resourceProfileIdToResourceProfile.filter { - case (otherId: Int, _: ResourceProfile) => - id != otherId && rpEntry.resourcesCompatible(resourceProfile) - }.map(_._1).toSet - writeLock.lock() - resourceProfileIdToCompatibleResourceProfileIds.put(id, compatibleIdSet) - } + resourceProfileIdToCompatibleResourceProfileIds.getOrElse(rpId, Set()) } finally { - writeLock.unlock() readLock.unlock() } } - def dumpResourceProfile(): Unit = { + /* + * Calculate and cache all compatible profile IDs excluding itself + * This operation is done when new ResourceProfile added + */ + private def calculateCompatibleProfileIds(): Unit = { + resourceProfileIdToResourceProfile.foreach { + case (id: Int, rpEntry: ResourceProfile) => + val resourceProfile = resourceProfileFromId(id) + val compatibleIdSet = resourceProfileIdToResourceProfile.filter { + case (otherId: Int, _: ResourceProfile) => + id != otherId && rpEntry.resourcesCompatible(resourceProfile) + }.map(_._1).toSet + resourceProfileIdToCompatibleResourceProfileIds.put(id, compatibleIdSet) + } + } + + private [spark] def dumpResourceProfiles(): Unit = { readLock.lock() try { + logDebug("dump resource profiles:") resourceProfileIdToResourceProfile.foreach { case (id: Int, rpEntry: ResourceProfile) => - logDebug("rpId " + id + ": " + rpEntry) + logDebug("-- rpId " + id + ": " + rpEntry) + } + if (reuseExecutors) { + logDebug("dump compatible resource profile ids:") + resourceProfileIdToCompatibleResourceProfileIds.foreach { + case (id: Int, compatibleIds: Set[Int]) => + logDebug("-- rpId " + id + ": " + compatibleIds) + } } } finally { readLock.unlock() diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala index 7b73bce47d98c..9f2073554de10 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala @@ -388,7 +388,7 @@ private[spark] class TaskSchedulerImpl( val taskSetRpID = taskSet.taskSet.resourceProfileId val assignTasks = if (reuseExecutors) { - val compatibleProfiles = sc.resourceProfileManager.getOtherCompatibleProfileIds(taskSetRpID) + val compatibleProfiles = sc.resourceProfileManager.getCompatibleProfileIds(taskSetRpID) taskSetRpID == shuffledOffers(i).resourceProfileId || compatibleProfiles.contains(shuffledOffers(i).resourceProfileId) } else { diff --git a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala index 5810d01af9b06..253c74770d7f1 100644 --- a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala +++ b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala @@ -296,7 +296,7 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite { rp1.require(execReqs).require(taskReqs) val rprof1 = rp1.build rpManager.addResourceProfile(rprof1) - rpManager.dumpResourceProfile() + rpManager.dumpResourceProfiles() post(SparkListenerStageSubmitted(createStageInfo(1, 1000, rp = rprof1))) val updatesNeeded = new mutable.HashMap[ResourceProfile, ExecutorAllocationManager.TargetNumUpdates] From 0465af33cdf223400ed7fa3f97146a05f20fcc6c Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Mon, 17 Jan 2022 14:52:44 +0800 Subject: [PATCH 15/26] Add resourceProfileCompatibleWithPolicy --- .../spark/resource/ResourceProfile.scala | 64 ++++++++++++++----- 1 file changed, 48 insertions(+), 16 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala index e6f02614cad3f..beaf4513edeb8 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala @@ -29,7 +29,6 @@ import org.apache.spark.annotation.{Evolving, Since} import org.apache.spark.internal.Logging import org.apache.spark.internal.config._ import org.apache.spark.internal.config.Python.PYSPARK_EXECUTOR_MEMORY -import org.apache.spark.resource.ResourceProfile.hasCustomExecutorResources import org.apache.spark.util.Utils /** @@ -259,29 +258,62 @@ class ResourceProfile( /** * Resource profile compatibility based on user policy - * strict match: - * only reuse executors with exact same resources (including all 3rd party resources), - * there is no resource waste but less user flexibility - * 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 requirements. But in this policy user should know - * there is some resource waste. They need to tradeoff reuse executor or create new ones. + * + * EQUAL_RESOURCES: + * only reuse executors with same resources (including cores, memory and all 3rd party + * resources) + * MORE_RESOURCES: + * reuse executors with equal or more resources (including cores, memory and all 3rd party + * resources) + * Notes: + * if EQUAL_RESOURCES is specified, there is no resource waste but less + * user flexibility. + * if MORE_RESOURCES is specified, users should know there are some resources + * wasted. They need to tradeoff between reusing executors and creating new ones. */ -object ResourceProfileCompatiblePolicy { +object ResourceProfileCompatiblePolicy extends Enumeration { + + type ResourceProfileCompatiblePolicy = Value + + val EQUAL_RESOURCES, MORE_RESOURCES = Value + + def resourceProfileCompatibleWithPolicy( + prevRP: ResourceProfile, curRP: ResourceProfile, + resourceNames: Set[String], policy: ResourceProfileCompatiblePolicy): Boolean = { + resourceNames.forall { resourceName: String => + policy match { + case EQUAL_RESOURCES => + ! prevRP.executorResources.get(resourceName).isEmpty && + ! curRP.executorResources.get(resourceName).isEmpty && + prevRP.executorResources(resourceName).amount == + curRP.executorResources(resourceName).amount + case MORE_RESOURCES => + ! prevRP.executorResources.get(resourceName).isEmpty && + ! curRP.executorResources.get(resourceName).isEmpty && + prevRP.executorResources(resourceName).amount >= + curRP.executorResources(resourceName).amount + } + } + } def resourceProfileCompatibleWithEqualCores(prevRP: ResourceProfile, curRP: ResourceProfile): Boolean = { - !hasCustomExecutorResources(prevRP) && !hasCustomExecutorResources(curRP) && - prevRP.executorResources("cores") == curRP.executorResources("cores") + resourceProfileCompatibleWithPolicy(prevRP, curRP, Set("cores"), EQUAL_RESOURCES) } def resourceProfileCompatibleWithMoreCores(prevRP: ResourceProfile, curRP: ResourceProfile): Boolean = { - !hasCustomExecutorResources(prevRP) && !hasCustomExecutorResources(curRP) && - prevRP.executorResources("cores").amount > curRP.executorResources("cores").amount + resourceProfileCompatibleWithPolicy(prevRP, curRP, Set("cores"), MORE_RESOURCES) + } + + def resourceProfileCompatibleWithEqualResources( + prevRP: ResourceProfile, curRP: ResourceProfile, resourceNames: Set[String]): Boolean = { + resourceProfileCompatibleWithPolicy(prevRP, curRP, resourceNames, EQUAL_RESOURCES) + } + + def resourceProfileCompatibleWithMoreResources( + prevRP: ResourceProfile, curRP: ResourceProfile, resourceNames: Set[String]): Boolean = { + resourceProfileCompatibleWithPolicy(prevRP, curRP, resourceNames, MORE_RESOURCES) } } From 67d2835918b3be6e4da943736bafe304be610189 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Mon, 17 Jan 2022 15:26:52 +0800 Subject: [PATCH 16/26] Add withResources with reuse params --- .../main/scala/org/apache/spark/rdd/RDD.scala | 17 +++++++++++++++++ .../spark/resource/ResourceProfileManager.scala | 15 +++++++++++++++ core/src/test/resources/log4j2.properties | 8 ++++++-- 3 files changed, 38 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/rdd/RDD.scala b/core/src/main/scala/org/apache/spark/rdd/RDD.scala index 4c39d178d38e8..66ecf18bee397 100644 --- a/core/src/main/scala/org/apache/spark/rdd/RDD.scala +++ b/core/src/main/scala/org/apache/spark/rdd/RDD.scala @@ -45,6 +45,7 @@ import org.apache.spark.partial.CountEvaluator import org.apache.spark.partial.GroupedCountEvaluator import org.apache.spark.partial.PartialResult import org.apache.spark.resource.ResourceProfile +import org.apache.spark.resource.ResourceProfileCompatiblePolicy.ResourceProfileCompatiblePolicy import org.apache.spark.storage.{RDDBlockId, StorageLevel} import org.apache.spark.util.{BoundedPriorityQueue, Utils} import org.apache.spark.util.collection.{ExternalAppendOnlyMap, OpenHashMap, @@ -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 + */ + @Experimental + @Since("3.3.0") + def withResources(rp: ResourceProfile, + reuseResourceNames: Set[String], reusePolicy: ResourceProfileCompatiblePolicy): this.type = { + resourceProfile = Option(rp) + sc.resourceProfileManager.updateResourceReusePolicy(reuseResourceNames, reusePolicy) + sc.resourceProfileManager.addResourceProfile(resourceProfile.get) + this + } + /** * Get the ResourceProfile specified with this RDD or null if it wasn't specified. * @return the user specified ResourceProfile or null (for Java compatibility) if diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala index 5e35d9dbc84c0..c6e6d933286c3 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala @@ -26,6 +26,7 @@ import org.apache.spark.annotation.Evolving import org.apache.spark.internal.Logging import org.apache.spark.internal.config.SCHEDULER_REUSE_COMPATIBLE_EXECUTORS import org.apache.spark.internal.config.Tests._ +import org.apache.spark.resource.ResourceProfileCompatiblePolicy.{EQUAL_RESOURCES, ResourceProfileCompatiblePolicy} import org.apache.spark.scheduler.{LiveListenerBus, SparkListenerResourceProfileAdded} import org.apache.spark.util.Utils import org.apache.spark.util.Utils.isTesting @@ -54,6 +55,9 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, def defaultResourceProfile: ResourceProfile = defaultProfile + var reuseResourceNames: Set[String] = Set("cores") + var reusePolicy = EQUAL_RESOURCES + private val dynamicEnabled = Utils.isDynamicAllocationEnabled(sparkConf) private val master = sparkConf.getOption("spark.master") private val isYarn = master.isDefined && master.get.equals("yarn") @@ -81,6 +85,17 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, true } + def updateResourceReusePolicy(resourceNames: Set[String], + policy: ResourceProfileCompatiblePolicy): Unit = { + writeLock.lock() + try { + reuseResourceNames = resourceNames + reusePolicy = policy + } finally { + writeLock.unlock() + } + } + def addResourceProfile(rp: ResourceProfile): Unit = { isSupported(rp) var putNewProfile = false diff --git a/core/src/test/resources/log4j2.properties b/core/src/test/resources/log4j2.properties index c6cd10d639e69..8573a42a85662 100644 --- a/core/src/test/resources/log4j2.properties +++ b/core/src/test/resources/log4j2.properties @@ -16,8 +16,9 @@ # # Set everything to be logged to the file target/unit-tests.log -rootLogger.level = info -rootLogger.appenderRef.file.ref = ${sys:test.appender:-File} +rootLogger.level = debug +#rootLogger.appenderRef.file.ref = ${sys:test.appender:-File} +rootLogger.appenderRef.stdout.ref = console appender.file.type = File appender.file.name = File @@ -34,6 +35,9 @@ appender.console.target = SYSTEM_ERR appender.console.layout.type = PatternLayout appender.console.layout.pattern = %t: %m%n +logger.test.name = org.apache.spark.ResourceProfileManager +logger.test.level = debug + # Ignore messages below warning level from Jetty, because it's a bit verbose logger.jetty.name = org.sparkproject.jetty logger.jetty.level = warn From 36984632ee3467702d90dd0f5ae7822160c54fe4 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Tue, 18 Jan 2022 12:00:45 +0800 Subject: [PATCH 17/26] Add JsonProtocol, Event Listener and WebUI --- .../resource/ResourceProfileManager.scala | 3 +- .../spark/scheduler/SparkListener.scala | 5 ++- .../spark/status/AppStatusListener.scala | 5 ++- .../org/apache/spark/status/LiveEntity.scala | 4 +- .../org/apache/spark/status/api/v1/api.scala | 3 +- .../apache/spark/ui/env/EnvironmentPage.scala | 12 +++++- .../org/apache/spark/util/JsonProtocol.scala | 38 +++++++++++++++---- .../apache/spark/util/JsonProtocolSuite.scala | 2 +- 8 files changed, 56 insertions(+), 16 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala index c6e6d933286c3..af45f0738a164 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala @@ -116,7 +116,8 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, // force the computation of maxTasks and limitingResource now so we don't have cost later rp.limitingResource(sparkConf) logInfo(s"Added ResourceProfile id: ${rp.id}") - listenerBus.post(SparkListenerResourceProfileAdded(rp)) + listenerBus.post(SparkListenerResourceProfileAdded(rp, + resourceProfileIdToCompatibleResourceProfileIds.get(rp.id))) } } diff --git a/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala b/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala index a9d8634794028..637a3221511ae 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala +++ b/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, + compatibleResourceProfileIds: Option[Set[Int]]) extends SparkListenerEvent /** diff --git a/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala b/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala index 35c43b06c284f..21ed2e828d984 100644 --- a/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala +++ b/core/src/main/scala/org/apache/spark/status/AppStatusListener.scala @@ -158,10 +158,11 @@ private[spark] class AppStatusListener( None } val liveRP = new LiveResourceProfile(event.resourceProfile.id, - event.resourceProfile.executorResources, event.resourceProfile.taskResources, maxTasks) + event.resourceProfile.executorResources, event.resourceProfile.taskResources, + event.compatibleResourceProfileIds, maxTasks) liveResourceProfiles(event.resourceProfile.id) = liveRP val rpInfo = new v1.ResourceProfileInfo(liveRP.resourceProfileId, - liveRP.executorResources, liveRP.taskResources) + liveRP.executorResources, liveRP.taskResources, liveRP.compatibleResourceProfileIds) kvstore.write(new ResourceProfileWrapper(rpInfo)) } diff --git a/core/src/main/scala/org/apache/spark/status/LiveEntity.scala b/core/src/main/scala/org/apache/spark/status/LiveEntity.scala index c5d72d3cfea68..b96831196e247 100644 --- a/core/src/main/scala/org/apache/spark/status/LiveEntity.scala +++ b/core/src/main/scala/org/apache/spark/status/LiveEntity.scala @@ -248,10 +248,12 @@ private class LiveResourceProfile( val resourceProfileId: Int, val executorResources: Map[String, ExecutorResourceRequest], val taskResources: Map[String, TaskResourceRequest], + val compatibleResourceProfileIds: Option[Set[Int]], val maxTasksPerExecutor: Option[Int]) extends LiveEntity { def toApi(): v1.ResourceProfileInfo = { - new v1.ResourceProfileInfo(resourceProfileId, executorResources, taskResources) + new v1.ResourceProfileInfo(resourceProfileId, executorResources, + taskResources, compatibleResourceProfileIds) } override protected def doUpdate(): Any = { diff --git a/core/src/main/scala/org/apache/spark/status/api/v1/api.scala b/core/src/main/scala/org/apache/spark/status/api/v1/api.scala index 86ddd3bb22b0d..51d3ee3c61548 100644 --- a/core/src/main/scala/org/apache/spark/status/api/v1/api.scala +++ b/core/src/main/scala/org/apache/spark/status/api/v1/api.scala @@ -65,7 +65,8 @@ case class ApplicationAttemptInfo private[spark]( class ResourceProfileInfo private[spark]( val id: Int, val executorResources: Map[String, ExecutorResourceRequest], - val taskResources: Map[String, TaskResourceRequest]) + val taskResources: Map[String, TaskResourceRequest], + val compatibleResourceProfileIds: Option[Set[Int]]) class ExecutorStageSummary private[spark]( val taskTime : Long, diff --git a/core/src/main/scala/org/apache/spark/ui/env/EnvironmentPage.scala b/core/src/main/scala/org/apache/spark/ui/env/EnvironmentPage.scala index 2f5b73118927b..a9355ecf45534 100644 --- a/core/src/main/scala/org/apache/spark/ui/env/EnvironmentPage.scala +++ b/core/src/main/scala/org/apache/spark/ui/env/EnvironmentPage.scala @@ -61,10 +61,20 @@ private[ui] class EnvironmentPage( }.mkString("\n") } + def constructCompatibleResourceProfileIdsString(ids: Option[Set[Int]]): String = { + ids match { + case Some(value) => value.mkString(", ") + case None => "N/A" + } + } + val resourceProfileInfo = store.resourceProfileInfo().map { rinfo => val einfo = constructExecutorRequestString(rinfo.executorResources) + val compatibleInfo = constructCompatibleResourceProfileIdsString( + rinfo.compatibleResourceProfileIds) val tinfo = constructTaskRequestString(rinfo.taskResources) - val res = s"Executor Reqs:\n$einfo\nTask Reqs:\n$tinfo" + val res = s"Compatible Resource Profile Ids: $compatibleInfo\n" + + s"Executor Reqs:\n$einfo\nTask Reqs:\n$tinfo" (rinfo.id.toString, res) }.toMap diff --git a/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala b/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala index 4e68ee0ed83cd..629a56f07d067 100644 --- a/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala +++ b/core/src/main/scala/org/apache/spark/util/JsonProtocol.scala @@ -228,12 +228,24 @@ private[spark] object JsonProtocol { } def resourceProfileAddedToJson(profileAdded: SparkListenerResourceProfileAdded): JValue = { - ("Event" -> SPARK_LISTENER_EVENT_FORMATTED_CLASS_NAMES.resourceProfileAdded) ~ - ("Resource Profile Id" -> profileAdded.resourceProfile.id) ~ - ("Executor Resource Requests" -> - executorResourceRequestMapToJson(profileAdded.resourceProfile.executorResources)) ~ - ("Task Resource Requests" -> - taskResourceRequestMapToJson(profileAdded.resourceProfile.taskResources)) + profileAdded.compatibleResourceProfileIds match { + case Some(value) => + ("Event" -> SPARK_LISTENER_EVENT_FORMATTED_CLASS_NAMES.resourceProfileAdded) ~ + ("Resource Profile Id" -> profileAdded.resourceProfile.id) ~ + ("Resource Profile Compatible Ids" -> + resourceProfileCompatibleIdsToJson(Some(value))) + ("Executor Resource Requests" -> + executorResourceRequestMapToJson(profileAdded.resourceProfile.executorResources)) ~ + ("Task Resource Requests" -> + taskResourceRequestMapToJson(profileAdded.resourceProfile.taskResources)) + case None => + ("Event" -> SPARK_LISTENER_EVENT_FORMATTED_CLASS_NAMES.resourceProfileAdded) ~ + ("Resource Profile Id" -> profileAdded.resourceProfile.id) ~ + ("Executor Resource Requests" -> + executorResourceRequestMapToJson(profileAdded.resourceProfile.executorResources)) ~ + ("Task Resource Requests" -> + taskResourceRequestMapToJson(profileAdded.resourceProfile.taskResources)) + } } def executorAddedToJson(executorAdded: SparkListenerExecutorAdded): JValue = { @@ -544,6 +556,13 @@ private[spark] object JsonProtocol { ("Disk Size" -> blockUpdatedInfo.diskSize) } + def resourceProfileCompatibleIdsToJson(ids: Option[Set[Int]]): JValue = { + ids match { + case Some(value) => JSet(value.map(JInt(_))) + case None => JSet(Set[JValue]()) + } + } + def executorResourceRequestToJson(execReq: ExecutorResourceRequest): JValue = { ("Resource Name" -> execReq.resourceName) ~ ("Amount" -> execReq.amount) ~ @@ -736,11 +755,16 @@ private[spark] object JsonProtocol { def resourceProfileAddedFromJson(json: JValue): SparkListenerResourceProfileAdded = { val profId = (json \ "Resource Profile Id").extract[Int] + val compatibleIds = (json \ "Resource Profile Compatible Ids").extract[Set[Int]] val executorReqs = executorResourceRequestMapFromJson(json \ "Executor Resource Requests") val taskReqs = taskResourceRequestMapFromJson(json \ "Task Resource Requests") val rp = new ResourceProfile(executorReqs.toMap, taskReqs.toMap) rp.setResourceProfileId(profId) - SparkListenerResourceProfileAdded(rp) + if (!compatibleIds.isEmpty) { + SparkListenerResourceProfileAdded(rp, Some(compatibleIds)) + } else { + SparkListenerResourceProfileAdded(rp, None) + } } def executorResourceRequestFromJson(json: JValue): ExecutorResourceRequest = { diff --git a/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala b/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala index 4eea2256553f5..0fa4e979b7f49 100644 --- a/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala +++ b/core/src/test/scala/org/apache/spark/util/JsonProtocolSuite.scala @@ -137,7 +137,7 @@ class JsonProtocolSuite extends SparkFunSuite { rprofBuilder.require(taskReq).require(execReq) val resourceProfile = rprofBuilder.build resourceProfile.setResourceProfileId(21) - val resourceProfileAdded = SparkListenerResourceProfileAdded(resourceProfile) + val resourceProfileAdded = SparkListenerResourceProfileAdded(resourceProfile, None) testEvent(stageSubmitted, stageSubmittedJsonString) testEvent(stageCompleted, stageCompletedJsonString) testEvent(taskStart, taskStartJsonString) From 97a32cf2bf6050499763454a1df0fb718b730b73 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Tue, 18 Jan 2022 15:52:25 +0800 Subject: [PATCH 18/26] fix mima --- project/MimaExcludes.scala | 2 ++ 1 file changed, 2 insertions(+) diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala index b985f95b85c6d..51bf979981751 100644 --- a/project/MimaExcludes.scala +++ b/project/MimaExcludes.scala @@ -41,6 +41,8 @@ object MimaExcludes { ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.executor.CoarseGrainedExecutorBackend#Arguments.*"), ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.spark.executor.CoarseGrainedExecutorBackend#Arguments.*"), ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.executor.CoarseGrainedExecutorBackend$Arguments$"), + // [SPARK-36699][Core] Reuse compatible executors for stage-level scheduling + ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.scheduler.SparkListenerResourceProfileAdded$") // [SPARK-37391][SQL] JdbcConnectionProvider tells if it modifies security context ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.jdbc.JdbcConnectionProvider.modifiesSecurityContext"), From c22277e074022244997218de9756ed85d1393830 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Tue, 18 Jan 2022 16:16:09 +0800 Subject: [PATCH 19/26] nit --- project/MimaExcludes.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala index 51bf979981751..257ca4a5a607a 100644 --- a/project/MimaExcludes.scala +++ b/project/MimaExcludes.scala @@ -42,7 +42,7 @@ object MimaExcludes { ProblemFilters.exclude[IncompatibleResultTypeProblem]("org.apache.spark.executor.CoarseGrainedExecutorBackend#Arguments.*"), ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.executor.CoarseGrainedExecutorBackend$Arguments$"), // [SPARK-36699][Core] Reuse compatible executors for stage-level scheduling - ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.scheduler.SparkListenerResourceProfileAdded$") + ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.scheduler.SparkListenerResourceProfileAdded$"), // [SPARK-37391][SQL] JdbcConnectionProvider tells if it modifies security context ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.jdbc.JdbcConnectionProvider.modifiesSecurityContext"), From 266c31d283d7e25241047c245ab66f3cd018e482 Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Tue, 18 Jan 2022 16:43:46 +0800 Subject: [PATCH 20/26] nit --- project/MimaExcludes.scala | 2 ++ 1 file changed, 2 insertions(+) diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala index 257ca4a5a607a..0cf78fa754fc4 100644 --- a/project/MimaExcludes.scala +++ b/project/MimaExcludes.scala @@ -43,6 +43,8 @@ object MimaExcludes { ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.executor.CoarseGrainedExecutorBackend$Arguments$"), // [SPARK-36699][Core] Reuse compatible executors for stage-level scheduling ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.scheduler.SparkListenerResourceProfileAdded$"), + ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.scheduler.SparkListenerResourceProfileAdded.*"), + ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.ResourceProfileInfo.this"), // [SPARK-37391][SQL] JdbcConnectionProvider tells if it modifies security context ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.spark.sql.jdbc.JdbcConnectionProvider.modifiesSecurityContext"), From 437a1ab4af9c3dfaa8a9f8c8ecb3a66cc101227c Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Wed, 19 Jan 2022 11:46:09 +0800 Subject: [PATCH 21/26] revert log4j2.properties --- core/src/test/resources/log4j2.properties | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/core/src/test/resources/log4j2.properties b/core/src/test/resources/log4j2.properties index 8573a42a85662..c6cd10d639e69 100644 --- a/core/src/test/resources/log4j2.properties +++ b/core/src/test/resources/log4j2.properties @@ -16,9 +16,8 @@ # # Set everything to be logged to the file target/unit-tests.log -rootLogger.level = debug -#rootLogger.appenderRef.file.ref = ${sys:test.appender:-File} -rootLogger.appenderRef.stdout.ref = console +rootLogger.level = info +rootLogger.appenderRef.file.ref = ${sys:test.appender:-File} appender.file.type = File appender.file.name = File @@ -35,9 +34,6 @@ appender.console.target = SYSTEM_ERR appender.console.layout.type = PatternLayout appender.console.layout.pattern = %t: %m%n -logger.test.name = org.apache.spark.ResourceProfileManager -logger.test.level = debug - # Ignore messages below warning level from Jetty, because it's a bit verbose logger.jetty.name = org.sparkproject.jetty logger.jetty.level = warn From 9155338c9177083aff978422007c3267bb2c6a9c Mon Sep 17 00:00:00 2001 From: "Wu, Xiaochang" Date: Wed, 19 Jan 2022 18:00:09 +0800 Subject: [PATCH 22/26] Update calculateAvailableSlots --- .../scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala index 9f2073554de10..a8ec0f0ad3acf 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala @@ -1207,6 +1207,7 @@ private[spark] object TaskSchedulerImpl { availableCpus: Array[Int], availableResources: Array[Map[String, Int]]): Int = { val resourceProfile = scheduler.sc.resourceProfileManager.resourceProfileFromId(rpId) + val compatibleProfileIds = scheduler.sc.resourceProfileManager.getCompatibleProfileIds(rpId) val coresKnown = resourceProfile.isCoresLimitKnown val (limitingResource, limitedByCpu) = { val limiting = resourceProfile.limitingResource(conf) @@ -1221,7 +1222,9 @@ private[spark] object TaskSchedulerImpl { val taskLimit = resourceProfile.taskResources.get(limitingResource).map(_.amount).get availableCpus.zip(availableResources).zip(availableRPIds) - .filter { case (_, id) => id == rpId } + // if reuse executors, compatible ones are included, + // otherwise compatibleProfileIds will be empty + .filter { case (_, id) => id == rpId || compatibleProfileIds.contains(id)} .map { case ((cpu, resources), _) => val numTasksPerExecCores = cpu / cpusPerTask if (limitedByCpu) { From 8deb499df7db62f7575f0d0ead57c590fd058ec8 Mon Sep 17 00:00:00 2001 From: Xiaochang Wu Date: Mon, 18 Apr 2022 17:47:27 +0800 Subject: [PATCH 23/26] nit --- .../org/apache/spark/ExecutorAllocationManager.scala | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala index 385a99af0ef2c..84315d5caa9a2 100644 --- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala +++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala @@ -303,8 +303,7 @@ private[spark] class ExecutorAllocationManager( val maxNeeded = math.ceil(numRunningOrPendingTasks * executorAllocationRatio / tasksPerExecutor).toInt - logDebug(s"max needed for rpId: $rpId numpending: $numRunningOrPendingTasks," + - s" tasksperexecutor: $tasksPerExecutor = $maxNeeded") + logDebug(s"max executors needed for rpId $rpId: $maxNeeded") val maxNeededWithSpeculationLocalityOffset = if (tasksPerExecutor > 1 && maxNeeded == 1 && pendingSpeculative > 0) { @@ -523,10 +522,10 @@ private[spark] class ExecutorAllocationManager( } } else { val numCompatibleExecutors = numExecutorsTargetsCompatibleProfiles(rpId) - if (oldNumExecutorsTarget + numCompatibleExecutors >= maxNumExecutors) { + if (oldNumExecutorsTarget + numCompatibleExecutors >= maxNumExecutors) { logDebug("Not adding executors because our current target total is already " + - s"${oldNumExecutorsTarget} (limit: max $maxNumExecutors - " + - s"reused $numCompatibleExecutors)") + s"${oldNumExecutorsTarget} (limit: max $maxNumExecutors)" + + s"(reused $numCompatibleExecutors)") numExecutorsToAddPerResourceProfileId(rpId) = 1 return 0 } From 2abef37b6cd21759bad0a8880a076145a3471805 Mon Sep 17 00:00:00 2001 From: Xiaochang Wu Date: Wed, 20 Apr 2022 11:24:03 +0800 Subject: [PATCH 24/26] add .doc for reuseCompatibleExecutors --- .../main/scala/org/apache/spark/internal/config/package.scala | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index 5a29ebc583f72..50d057ac31fe6 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -2030,6 +2030,10 @@ package object config { private[spark] val SCHEDULER_REUSE_COMPATIBLE_EXECUTORS = ConfigBuilder("spark.scheduler.reuseCompatibleExecutors") .version("3.3.0") + .doc("Whether to reuse compatible executors for stage-level scheduling. " + + "When enabled, the executors are reused over different stages when " + + "their resource profiles are compatible. The user should specify compatible " + + "policy for resource profiles.") .booleanConf .createWithDefault(false) From 379fbeac078cfff0582192cebed4fcb3867fd5c7 Mon Sep 17 00:00:00 2001 From: Xiaochang Wu Date: Wed, 20 Apr 2022 14:21:33 +0800 Subject: [PATCH 25/26] nit identation --- .../org/apache/spark/scheduler/TaskSchedulerImpl.scala | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala index a8ec0f0ad3acf..d82df1a7b5342 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala @@ -392,10 +392,10 @@ private[spark] class TaskSchedulerImpl( taskSetRpID == shuffledOffers(i).resourceProfileId || compatibleProfiles.contains(shuffledOffers(i).resourceProfileId) } else { - // make the resource profile id a hard requirement for now - ie only put tasksets - // on executors where resource profile exactly matches. - taskSetRpID == shuffledOffers(i).resourceProfileId - } + // make the resource profile id a hard requirement for now - ie only put tasksets + // on executors where resource profile exactly matches. + taskSetRpID == shuffledOffers(i).resourceProfileId + } if (assignTasks) { val taskResAssignmentsOpt = resourcesMeetTaskRequirements(taskSet, availableCpus(i), From b2d16b8ca16f1d3110ddb1d1368d9c079a2fb837 Mon Sep 17 00:00:00 2001 From: Xiaochang Wu Date: Thu, 21 Apr 2022 14:56:30 +0800 Subject: [PATCH 26/26] refactor ResourceProfileCompatiblePolicyInterface --- .../main/scala/org/apache/spark/rdd/RDD.scala | 23 +--- .../spark/resource/ResourceProfile.scala | 66 +---------- .../ResourceProfileCompatiblePolicy.scala | 106 ++++++++++++++++++ .../resource/ResourceProfileManager.scala | 16 +-- 4 files changed, 121 insertions(+), 90 deletions(-) create mode 100644 core/src/main/scala/org/apache/spark/resource/ResourceProfileCompatiblePolicy.scala diff --git a/core/src/main/scala/org/apache/spark/rdd/RDD.scala b/core/src/main/scala/org/apache/spark/rdd/RDD.scala index 7b27357aad0a5..beb56cc599774 100644 --- a/core/src/main/scala/org/apache/spark/rdd/RDD.scala +++ b/core/src/main/scala/org/apache/spark/rdd/RDD.scala @@ -44,8 +44,7 @@ import org.apache.spark.partial.BoundedDouble import org.apache.spark.partial.CountEvaluator import org.apache.spark.partial.GroupedCountEvaluator import org.apache.spark.partial.PartialResult -import org.apache.spark.resource.ResourceProfile -import org.apache.spark.resource.ResourceProfileCompatiblePolicy.ResourceProfileCompatiblePolicy +import org.apache.spark.resource.{ResourceProfile, ResourceProfileCompatiblePolicyInterface} import org.apache.spark.storage.{RDDBlockId, StorageLevel} import org.apache.spark.util.{BoundedPriorityQueue, Utils} import org.apache.spark.util.collection.{ExternalAppendOnlyMap, OpenHashMap, @@ -1801,28 +1800,16 @@ abstract class RDD[T: ClassTag]( * certain cluster managers and currently requires dynamic allocation to be enabled. * It will result in new executors with the resources specified being acquired to * calculate the RDD. - */ - @Experimental - @Since("3.1.0") - def withResources(rp: ResourceProfile): this.type = { - resourceProfile = Option(rp) - sc.resourceProfileManager.addResourceProfile(resourceProfile.get) - 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 + * + * @param reusePolicy specify executor reuse policy, used only when reusing executors enabled */ @Experimental @Since("3.3.0") def withResources(rp: ResourceProfile, - reuseResourceNames: Set[String], reusePolicy: ResourceProfileCompatiblePolicy): this.type = { + reusePolicy: Option[ResourceProfileCompatiblePolicyInterface] = None): this.type = { resourceProfile = Option(rp) - sc.resourceProfileManager.updateResourceReusePolicy(reuseResourceNames, reusePolicy) sc.resourceProfileManager.addResourceProfile(resourceProfile.get) + sc.resourceProfileManager.updateResourceReusePolicy(reusePolicy) this } diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala index 8ec64d3cfd3db..f1136ce77a8ee 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfile.scala @@ -242,10 +242,9 @@ class ResourceProfile( } private[spark] def resourcesCompatible(rp: ResourceProfile, - resourceProfileCompatibleFunc: (ResourceProfile, ResourceProfile) => Boolean - = ResourceProfileCompatiblePolicy.resourceProfileCompatibleWithEqualCores) + resourceProfileCompatiblePolicy: ResourceProfileCompatiblePolicyInterface) : Boolean = { - resourceProfileCompatibleFunc(this, rp) + resourceProfileCompatiblePolicy.isCompatible(this, rp) } override def hashCode(): Int = Seq(taskResources, executorResources).hashCode() @@ -256,67 +255,6 @@ class ResourceProfile( } } -/** - * Resource profile compatibility based on user policy - * - * EQUAL_RESOURCES: - * only reuse executors with same resources (including cores, memory and all 3rd party - * resources) - * MORE_RESOURCES: - * reuse executors with equal or more resources (including cores, memory and all 3rd party - * resources) - * Notes: - * if EQUAL_RESOURCES is specified, there is no resource waste but less - * user flexibility. - * if MORE_RESOURCES is specified, users should know there are some resources - * wasted. They need to tradeoff between reusing executors and creating new ones. - */ -object ResourceProfileCompatiblePolicy extends Enumeration { - - type ResourceProfileCompatiblePolicy = Value - - val EQUAL_RESOURCES, MORE_RESOURCES = Value - - def resourceProfileCompatibleWithPolicy( - prevRP: ResourceProfile, curRP: ResourceProfile, - resourceNames: Set[String], policy: ResourceProfileCompatiblePolicy): Boolean = { - resourceNames.forall { resourceName: String => - policy match { - case EQUAL_RESOURCES => - ! prevRP.executorResources.get(resourceName).isEmpty && - ! curRP.executorResources.get(resourceName).isEmpty && - prevRP.executorResources(resourceName).amount == - curRP.executorResources(resourceName).amount - case MORE_RESOURCES => - ! prevRP.executorResources.get(resourceName).isEmpty && - ! curRP.executorResources.get(resourceName).isEmpty && - prevRP.executorResources(resourceName).amount >= - curRP.executorResources(resourceName).amount - } - } - } - - def resourceProfileCompatibleWithEqualCores(prevRP: ResourceProfile, - curRP: ResourceProfile): Boolean = { - resourceProfileCompatibleWithPolicy(prevRP, curRP, Set("cores"), EQUAL_RESOURCES) - } - - def resourceProfileCompatibleWithMoreCores(prevRP: ResourceProfile, - curRP: ResourceProfile): Boolean = { - resourceProfileCompatibleWithPolicy(prevRP, curRP, Set("cores"), MORE_RESOURCES) - } - - def resourceProfileCompatibleWithEqualResources( - prevRP: ResourceProfile, curRP: ResourceProfile, resourceNames: Set[String]): Boolean = { - resourceProfileCompatibleWithPolicy(prevRP, curRP, resourceNames, EQUAL_RESOURCES) - } - - def resourceProfileCompatibleWithMoreResources( - prevRP: ResourceProfile, curRP: ResourceProfile, resourceNames: Set[String]): Boolean = { - resourceProfileCompatibleWithPolicy(prevRP, curRP, resourceNames, MORE_RESOURCES) - } -} - object ResourceProfile extends Logging { // task resources /** diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfileCompatiblePolicy.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfileCompatiblePolicy.scala new file mode 100644 index 0000000000000..3c0669a2f9d41 --- /dev/null +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfileCompatiblePolicy.scala @@ -0,0 +1,106 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.resource + +import org.apache.spark.annotation.Since +import org.apache.spark.resource.ResourceProfileCompatiblePolicy.resourceProfileCompatibleWithEqualCores + +/** + * Resource profile compatible policy interface allows the user to customize its + * own compatible policy + * + */ +@Since("3.3.0") +trait ResourceProfileCompatiblePolicyInterface { + def getName(): String + def isCompatible(prevRP: ResourceProfile, curRP: ResourceProfile): Boolean +} + +/** + * Resource profile compatible policy: only reuse executors with same cores + */ +class ResourceProfileCompatiblePolicyEqualCores extends ResourceProfileCompatiblePolicyInterface { + override def getName(): String = { + return "EXEC_EQUAL_CORES" + } + override def isCompatible(prevRP: ResourceProfile, curRP: ResourceProfile): Boolean = { + resourceProfileCompatibleWithEqualCores(prevRP, curRP) + } + +} + +/** + * Resource profile compatibility based on user policy + * + * EQUAL_RESOURCES: + * only reuse executors with same resources (including cores, memory and all 3rd party + * resources) + * MORE_RESOURCES: + * reuse executors with equal or more resources (including cores, memory and all 3rd party + * resources) + * Notes: + * if EQUAL_RESOURCES is specified, there is no resource waste but less + * user flexibility. + * if MORE_RESOURCES is specified, users should know there are some resources + * wasted. They need to tradeoff between reusing executors and creating new ones. + */ +object ResourceProfileCompatiblePolicy extends Enumeration { + + type ResourceProfileCompatiblePolicy = Value + + val EQUAL_RESOURCES, MORE_RESOURCES = Value + + private def resourceProfileCompatibleWithPolicy( + prevRP: ResourceProfile, curRP: ResourceProfile, + resourceNames: Set[String], policy: ResourceProfileCompatiblePolicy): Boolean = { + resourceNames.forall { resourceName: String => + policy match { + case EQUAL_RESOURCES => + ! prevRP.executorResources.get(resourceName).isEmpty && + ! curRP.executorResources.get(resourceName).isEmpty && + prevRP.executorResources(resourceName).amount == + curRP.executorResources(resourceName).amount + case MORE_RESOURCES => + ! prevRP.executorResources.get(resourceName).isEmpty && + ! curRP.executorResources.get(resourceName).isEmpty && + prevRP.executorResources(resourceName).amount >= + curRP.executorResources(resourceName).amount + } + } + } + + def resourceProfileCompatibleWithEqualCores(prevRP: ResourceProfile, + curRP: ResourceProfile): Boolean = { + resourceProfileCompatibleWithPolicy(prevRP, curRP, Set("cores"), EQUAL_RESOURCES) + } + + def resourceProfileCompatibleWithMoreCores(prevRP: ResourceProfile, + curRP: ResourceProfile): Boolean = { + resourceProfileCompatibleWithPolicy(prevRP, curRP, Set("cores"), MORE_RESOURCES) + } + + def resourceProfileCompatibleWithEqualResources( + prevRP: ResourceProfile, curRP: ResourceProfile, resourceNames: Set[String]): Boolean = { + resourceProfileCompatibleWithPolicy(prevRP, curRP, resourceNames, EQUAL_RESOURCES) + } + + def resourceProfileCompatibleWithMoreResources( + prevRP: ResourceProfile, curRP: ResourceProfile, resourceNames: Set[String]): Boolean = { + resourceProfileCompatibleWithPolicy(prevRP, curRP, resourceNames, MORE_RESOURCES) + } +} diff --git a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala index af45f0738a164..b7f4874891c78 100644 --- a/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala +++ b/core/src/main/scala/org/apache/spark/resource/ResourceProfileManager.scala @@ -26,7 +26,6 @@ import org.apache.spark.annotation.Evolving import org.apache.spark.internal.Logging import org.apache.spark.internal.config.SCHEDULER_REUSE_COMPATIBLE_EXECUTORS import org.apache.spark.internal.config.Tests._ -import org.apache.spark.resource.ResourceProfileCompatiblePolicy.{EQUAL_RESOURCES, ResourceProfileCompatiblePolicy} import org.apache.spark.scheduler.{LiveListenerBus, SparkListenerResourceProfileAdded} import org.apache.spark.util.Utils import org.apache.spark.util.Utils.isTesting @@ -55,8 +54,9 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, def defaultResourceProfile: ResourceProfile = defaultProfile - var reuseResourceNames: Set[String] = Set("cores") - var reusePolicy = EQUAL_RESOURCES + // Set default reuse policy + var reusePolicy: ResourceProfileCompatiblePolicyInterface + = new ResourceProfileCompatiblePolicyEqualCores() private val dynamicEnabled = Utils.isDynamicAllocationEnabled(sparkConf) private val master = sparkConf.getOption("spark.master") @@ -85,12 +85,12 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, true } - def updateResourceReusePolicy(resourceNames: Set[String], - policy: ResourceProfileCompatiblePolicy): Unit = { + def updateResourceReusePolicy(policy: Option[ResourceProfileCompatiblePolicyInterface]): Unit = { writeLock.lock() try { - reuseResourceNames = resourceNames - reusePolicy = policy + if (policy.isDefined) { + reusePolicy = policy.get + } } finally { writeLock.unlock() } @@ -170,7 +170,7 @@ private[spark] class ResourceProfileManager(sparkConf: SparkConf, val resourceProfile = resourceProfileFromId(id) val compatibleIdSet = resourceProfileIdToResourceProfile.filter { case (otherId: Int, _: ResourceProfile) => - id != otherId && rpEntry.resourcesCompatible(resourceProfile) + id != otherId && rpEntry.resourcesCompatible(resourceProfile, reusePolicy) }.map(_._1).toSet resourceProfileIdToCompatibleResourceProfileIds.put(id, compatibleIdSet) }