Skip to content

[SPARK-1978] In some cases, spark-yarn does not automatically restart the failed container #921

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

Closed
wants to merge 10 commits into from
Closed

[SPARK-1978] In some cases, spark-yarn does not automatically restart the failed container #921

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
e00b656
In some cases, yarn does not automatically restart the container
witgo May 30, 2014
3c464bd
add time limit to allocateExecutors
witgo May 31, 2014
1faf4f4
Merge branch 'master' into allocateExecutors
witgo Jun 5, 2014
5c376e0
Merge branch 'master' of https://github.com/apache/spark into allocat…
witgo Jun 5, 2014
aff827c
review commit
witgo Jun 5, 2014
04c6f7e
Merge branch 'master' into allocateExecutors
witgo Jun 9, 2014
056b8c7
Merge branch 'master' of https://github.com/apache/spark into allocat…
witgo Jun 9, 2014
32ac7af
review commit
witgo Jun 9, 2014
8800eba
Merge branch 'master' of https://github.com/apache/spark into allocat…
witgo Jun 10, 2014
bc3aa66
review commit
witgo Jun 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
39 changes: 22 additions & 17 deletions yarn/stable/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -252,15 +252,12 @@ class ApplicationMaster(args: ApplicationMasterArguments, conf: Configuration,
try {
logInfo("Allocating " + args.numExecutors + " executors.")
// Wait until all containers have finished
// TODO: This is a bit ugly. Can we make it nicer?
// TODO: Handle container failure
yarnAllocator.addResourceRequests(args.numExecutors)
yarnAllocator.allocateResources()
// Exits the loop if the user thread exits.
while (yarnAllocator.getNumExecutorsRunning < args.numExecutors && userThread.isAlive) {
if (yarnAllocator.getNumExecutorsFailed >= maxNumExecutorFailures) {
finishApplicationMaster(FinalApplicationStatus.FAILED,
"max number of executor failures reached")
}
checkNumExecutorsFailed()
allocateMissingExecutor()
Copy link
Contributor

@tgravescs tgravescs Jun 5, 2014

Choose a reason for hiding this comment

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

Can you move these 2 lines up above the allocateResources() call. That way max failed check happens first, then we would add in any missing executors, and then call allocate.

Copy link
Contributor Author

@witgo witgo Jun 6, 2014

Choose a reason for hiding this comment

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

I think that only after the call yarnAllocator.allocateResources(),yarnAllocator.getNumExecutorsFailed value will change.

This is appropriate: after changing the value of yarnAllocator.getNumExecutorsFailed, immediately call checkNumExecutorsFailed

Copy link
Contributor

@tgravescs tgravescs Jun 9, 2014

Choose a reason for hiding this comment

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

You are correct that it isn't updated until that is called and in this has it really doesn't matter to much since we are in while loop and then after this we will just go to the reporter thread to continue to loop. I just find it odd to add in missing executors but then not call allocate immediately afterwards (we sleep and then we possibly break from the loop). For that reason I would prefer these to be moved up. Yes we have one extra unneeded call to them, but I think it flows better and will be more resilient to other code changes in the future.
Also note that launchReporterThread runs the logic the same way so right now you are doing the failed check and and allocateMissing twice in a row on the first time it goes into the reporter thread.

If you really have objections to that then we could make one call of allocateResources() outside of the while loop and then do the checkFailed, allocateMissing, and then allocateResources in side the while loop.

yarnAllocator.allocateResources()
ApplicationMaster.incrementAllocatorLoop(1)
Thread.sleep(100)
Expand Down Expand Up @@ -289,23 +286,31 @@ class ApplicationMaster(args: ApplicationMasterArguments, conf: Configuration,
}
}

private def allocateMissingExecutor() {
val missingExecutorCount = args.numExecutors - yarnAllocator.getNumExecutorsRunning -
yarnAllocator.getNumPendingAllocate
if (missingExecutorCount > 0) {
logInfo("Allocating %d containers to make up for (potentially) lost containers".
format(missingExecutorCount))
yarnAllocator.addResourceRequests(missingExecutorCount)
}
}

private def checkNumExecutorsFailed() {
if (yarnAllocator.getNumExecutorsFailed >= maxNumExecutorFailures) {
finishApplicationMaster(FinalApplicationStatus.FAILED,
"max number of executor failures reached")
}
}

private def launchReporterThread(_sleepTime: Long): Thread = {
val sleepTime = if (_sleepTime <= 0) 0 else _sleepTime

val t = new Thread {
override def run() {
while (userThread.isAlive) {
if (yarnAllocator.getNumExecutorsFailed >= maxNumExecutorFailures) {
finishApplicationMaster(FinalApplicationStatus.FAILED,
"max number of executor failures reached")
}
val missingExecutorCount = args.numExecutors - yarnAllocator.getNumExecutorsRunning -
yarnAllocator.getNumPendingAllocate
if (missingExecutorCount > 0) {
logInfo("Allocating %d containers to make up for (potentially) lost containers".
format(missingExecutorCount))
yarnAllocator.addResourceRequests(missingExecutorCount)
}
checkNumExecutorsFailed()
allocateMissingExecutor()
sendProgress()
Thread.sleep(sleepTime)
}
Expand Down
22 changes: 12 additions & 10 deletions yarn/stable/src/main/scala/org/apache/spark/deploy/yarn/ExecutorLauncher.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,17 +200,25 @@ class ExecutorLauncher(args: ApplicationMasterArguments, conf: Configuration, sp

logInfo("Allocating " + args.numExecutors + " executors.")
// Wait until all containers have finished
// TODO: This is a bit ugly. Can we make it nicer?
// TODO: Handle container failure

yarnAllocator.addResourceRequests(args.numExecutors)
yarnAllocator.allocateResources()
while ((yarnAllocator.getNumExecutorsRunning < args.numExecutors) && (!driverClosed)) {
allocateMissingExecutor()
Copy link
Contributor

@tgravescs tgravescs Jun 5, 2014

Choose a reason for hiding this comment

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

similar here can you move this up above allocateResources

Copy link
Contributor

@tgravescs tgravescs Jun 9, 2014

Choose a reason for hiding this comment

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

can you change this to have similar logic - allocate outside loop, then inside loop add missing and then allocate.

yarnAllocator.allocateResources()
Thread.sleep(100)
}

logInfo("All executors have launched.")
}

private def allocateMissingExecutor() {
val missingExecutorCount = args.numExecutors - yarnAllocator.getNumExecutorsRunning -
yarnAllocator.getNumPendingAllocate
if (missingExecutorCount > 0) {
logInfo("Allocating %d containers to make up for (potentially) lost containers".
format(missingExecutorCount))
yarnAllocator.addResourceRequests(missingExecutorCount)
}
}

// TODO: We might want to extend this to allocate more containers in case they die !
Expand All @@ -220,13 +228,7 @@ class ExecutorLauncher(args: ApplicationMasterArguments, conf: Configuration, sp
val t = new Thread {
override def run() {
while (!driverClosed) {
val missingExecutorCount = args.numExecutors - yarnAllocator.getNumExecutorsRunning -
yarnAllocator.getNumPendingAllocate
if (missingExecutorCount > 0) {
logInfo("Allocating %d containers to make up for (potentially) lost containers".
format(missingExecutorCount))
yarnAllocator.addResourceRequests(missingExecutorCount)
}
allocateMissingExecutor()
sendProgress()
Thread.sleep(sleepTime)
}
Expand Down