Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-3411] Improve load-balancing of concurrently-submitted drivers across workers #1106

Closed
wants to merge 7 commits into from
Closed

[SPARK-3411] Improve load-balancing of concurrently-submitted drivers across workers #1106

Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
bbc7087
Optimize the schedule in Master
WangTaoTheTonic Sep 5, 2014
bc91bb1
Avoid shuffle every time we schedule the driver using round robin
WangTaoTheTonic Sep 6, 2014
2ca3091
fix checkstyle
WangTaoTheTonic Sep 8, 2014
2835929
solve the failed test and avoid filtering
WangTaoTheTonic Sep 8, 2014
f674e59
add comment and minor fix
WangTaoTheTonic Sep 9, 2014
b6560cf
solve the shuffle problem for HashSet
WangTaoTheTonic Sep 9, 2014
d1a928b
Minor adjustment
WangTaoTheTonic Sep 10, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 15 additions & 3 deletions core/src/main/scala/org/apache/spark/deploy/master/Master.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -481,13 +481,25 @@ private[spark] class Master(
if (state != RecoveryState.ALIVE) { return }

// First schedule drivers, they take strict precedence over applications
val shuffledWorkers = Random.shuffle(workers) // Randomization helps balance drivers
for (worker <- shuffledWorkers if worker.state == WorkerState.ALIVE) {
for (driver <- List(waitingDrivers: _*)) { // iterate over a copy of waitingDrivers
// Randomization helps balance drivers
val shuffledAliveWorkers = Random.shuffle(workers.toSeq.filter(_.state == WorkerState.ALIVE))
val aliveWorkerNum = shuffledAliveWorkers.size
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These variables are now misnamed. These are no longer just "aliveWorkers"

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there any value to keeping workers whose state is not WorkerState.ALIVE in the array of shuffled workers, or should this be:

val shuffledAliveWorkers = Random.shuffle(workers.toSeq.filter(_.state == WorkerState.ALIVE))

var curPos = 0
for (driver <- waitingDrivers.toList) { // iterate over a copy of waitingDrivers
// We assign workers to each waiting driver in a round-robin fashion. For each driver, we
// start from the last worker that was assigned a driver, and continue onwards until we have
// explored all alive workers.
curPos = (curPos + 1) % aliveWorkerNum
val startPos = curPos
var launched = false
while (curPos != startPos && !launched) {
val worker = shuffledAliveWorkers(curPos)
if (worker.memoryFree >= driver.desc.mem && worker.coresFree >= driver.desc.cores) {
launchDriver(worker, driver)
waitingDrivers -= driver
launched = true
}
curPos = (curPos + 1) % aliveWorkerNum
}
}

Expand Down