Skip to content

[SPARK-12864][YARN] initialize executorIdCounter after ApplicationMaster killed for max n… #10794

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 5 commits into from
Closed

[SPARK-12864][YARN] initialize executorIdCounter after ApplicationMaster killed for max n… #10794

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
30048ac
initialize executorIdCounter after ApplicationMaster killed for max n…
zhonghaihua Jan 17, 2016
fa7d54b
change name RetrieveCurrentExecutorIdCounter to RetrieveMaxExecutorId…
zhonghaihua Feb 22, 2016
3a1724c
change the way of getting max executorId from driver
zhonghaihua Feb 25, 2016
659c505
change variables currentExecutorIdCounter to protected
zhonghaihua Feb 25, 2016
ebe3c7f
optimize and add comment annotation
zhonghaihua Apr 1, 2016
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
2 changes: 2 additions & 0 deletions core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,8 @@ private[spark] object CoarseGrainedClusterMessages {

case object RetrieveSparkProps extends CoarseGrainedClusterMessage

case object RetrieveLastAllocatedExecutorId extends CoarseGrainedClusterMessage

// Driver to executors
case class LaunchTask(data: SerializableBuffer) extends CoarseGrainedClusterMessage

Expand Down
6 changes: 6 additions & 0 deletions core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,9 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
// Executors that have been lost, but for which we don't yet know the real exit reason.
protected val executorsPendingLossReason = new HashSet[String]

// The num of current max ExecutorId used to re-register appMaster
protected var currentExecutorIdCounter = 0

class DriverEndpoint(override val rpcEnv: RpcEnv, sparkProperties: Seq[(String, String)])
extends ThreadSafeRpcEndpoint with Logging {

Expand Down Expand Up @@ -155,6 +158,9 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
// in this block are read when requesting executors
CoarseGrainedSchedulerBackend.this.synchronized {
executorDataMap.put(executorId, data)
if (currentExecutorIdCounter < executorId.toInt) {
currentExecutorIdCounter = executorId.toInt
}
Copy link
Contributor

@andrewor14 andrewor14 Feb 19, 2016

Choose a reason for hiding this comment

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

this is kind of awkward. You don't need to keep track of another variable; just compute the max executor ID when the AM asks for it. You already have all the information you need in executorDataMap.

Copy link
Contributor Author

@zhonghaihua zhonghaihua Feb 21, 2016

Choose a reason for hiding this comment

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

@andrewor14 Thank for review it. For my understanding, I don't think we can get the max executor ID in executorDataMap. Because, when AM is failure, all the executor are disconnect and be removed, by this time, as the code in method CoarseGrainedSchedulerBackend.removeExecutor show, the executor information in executorDataMap also be removed.
So, I think the executor information in executorDataMap is not complete. What do you think ?

Copy link
Contributor

@vanzin vanzin Mar 30, 2016

Choose a reason for hiding this comment

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

yes, with dynamic allocation, for example, the executor with the max known id may be gone already.

minor: executorId.toInt instead of Integer.parseInt

Copy link
Contributor Author

@zhonghaihua zhonghaihua Mar 31, 2016

Choose a reason for hiding this comment

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

@vanzin Thanks for your comments. I will optimize it.

if (numPendingExecutors > 0) {
numPendingExecutors -= 1
logDebug(s"Decremented number of pending executors ($numPendingExecutors left)")
Expand Down
20 changes: 18 additions & 2 deletions yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ import org.apache.spark.deploy.yarn.YarnSparkHadoopUtil._
import org.apache.spark.rpc.{RpcCallContext, RpcEndpointRef}
import org.apache.spark.scheduler.{ExecutorExited, ExecutorLossReason}
import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.RemoveExecutor
import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.RetrieveLastAllocatedExecutorId
import org.apache.spark.util.ThreadUtils

/**
Expand Down Expand Up @@ -81,8 +82,23 @@ private[yarn] class YarnAllocator(
new ConcurrentHashMap[ContainerId, java.lang.Boolean])

@volatile private var numExecutorsRunning = 0
// Used to generate a unique ID per executor
private var executorIdCounter = 0

/**
* Used to generate a unique ID per executor
*
* Init `executorIdCounter`. when AM restart, `executorIdCounter` will reset to 0. Then
* the id of new executor will start from 1, this will conflict with the executor has
Copy link
Contributor

@tgravescs tgravescs Mar 29, 2016

Choose a reason for hiding this comment

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

I think we need to clarify this to say this is required for client mode when driver isn't running on yarn. this isn't an issue in cluster mode.

Copy link
Contributor Author

@zhonghaihua zhonghaihua Mar 30, 2016

Choose a reason for hiding this comment

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

@tgravescs I think we can clarify this in SPARK-12864 issue. @andrewor14 What's your opinion ?

Copy link
Contributor

@tgravescs tgravescs Mar 30, 2016

Choose a reason for hiding this comment

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

I would prefer to do it here in this comment to better describe the situation this is needed. It should be a line or two and I personally much prefer that then pointing at jiras unless its a big discussion/background required, then the jira makes more sense.

Copy link
Contributor Author

@zhonghaihua zhonghaihua Mar 31, 2016

Choose a reason for hiding this comment

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

@tgravescs Ok, I will do it soon. Thanks a lot.

* already created before. So, we should initialize the `executorIdCounter` by getting
* the max executorId from driver.
*
* And this situation of executorId conflict is just in yarn client mode, so this is an issue
* in yarn client mode. For more details, can check in jira.
*
* @see SPARK-12864
*/
private var executorIdCounter: Int =
driverRef.askWithRetry[Int](RetrieveLastAllocatedExecutorId)

@volatile private var numExecutorsFailed = 0

@volatile private var targetNumExecutors =
Expand Down
3 changes: 3 additions & 0 deletions yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -287,6 +287,9 @@ private[spark] abstract class YarnSchedulerBackend(
logWarning("Attempted to kill executors before the AM has registered!")
context.reply(false)
}

case RetrieveLastAllocatedExecutorId =>
context.reply(currentExecutorIdCounter)
}

override def onDisconnected(remoteAddress: RpcAddress): Unit = {
Expand Down