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
zhonghaihua 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 3 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 RetrieveMaxExecutorId extends CoarseGrainedClusterMessage

@tgravescs tgravescs Mar 29, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

we seem to be intermixing current and Max. Max makes me think that this is some limit so can we change things to be consistent and perhaps use CurrentExecutorId or LargestAllocated

@zhonghaihua zhonghaihua Mar 30, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@tgravescs Thanks for reviewing. Because this variables is the max executorId in previous AM, and this is just called by initializing AM. Our intention is to get the max executorId from all executor in previous AM. So I think maxExecutorId is ok.
@andrewor14 What's your opinion ?

@tgravescs tgravescs Mar 30, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I understand what the variable is, but readers looking at just this message without other context wouldn't necessarily know. When I look at the name I think its giving me some limit for max executor id. For instance we have many configs that set max of things ( for instance spark.reducer.maxSizeInFlight, spark.shuffle.io.maxRetries, etc.) That is why I would like the name clarified.

If we change the calling context, then its not just the max executor id for the last AM, its the last executor id that was allocated. So perhaps rename to be RetrieveLastAllocatedExecutorId

@vanzin vanzin Mar 30, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

how about MaxKnownExecutorId?

@zhonghaihua zhonghaihua Mar 31, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@tgravescs Ok，I get your mean. Thanks a lot.
Use this name RetrieveLastAllocatedExecutorId is ok ? @vanzin What's your opinion ?


// 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
var currentExecutorIdCounter = 0

@jerryshao jerryshao Feb 25, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Please add scope keyword protected

@zhonghaihua zhonghaihua Feb 25, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Hi @jerryshao , thanks for your comments. The master branch is different from branch-1.5.x version. In master branch,CoarseGrainedSchedulerBackend is belong to module core and YarnSchedulerBackend is belong to module yarn , while in branch-1.5.x version it is belong to the same package. So, from my understanding, protected is unsuited here, right?

@jerryshao jerryshao Feb 25, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I don't think so. Though they're in different modules, still they're under same package, please see other variables like hostToLocalTaskCount.

@zhonghaihua zhonghaihua Feb 25, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Hi @jerryshao , you are right. I fix it now. Thanks a lot.


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 < Integer.parseInt(executorId)) {
currentExecutorIdCounter = Integer.parseInt(executorId)
}

@andrewor14 andrewor14 Feb 19, 2016

Copy link
Copy Markdown
Contributor

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.

@zhonghaihua zhonghaihua Feb 21, 2016

Copy link
Copy Markdown
Contributor Author

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 ?

@vanzin vanzin Mar 30, 2016

Copy link
Copy Markdown
Contributor

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

@zhonghaihua zhonghaihua Mar 31, 2016

Copy link
Copy Markdown
Contributor Author

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
17 changes: 15 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.RetrieveMaxExecutorId
import org.apache.spark.util.ThreadUtils

/**
Expand Down Expand Up @@ -81,8 +82,20 @@ 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

@tgravescs tgravescs Mar 29, 2016

Copy link
Copy Markdown
Contributor

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.

@zhonghaihua zhonghaihua Mar 30, 2016

Copy link
Copy Markdown
Contributor Author

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 ?

@tgravescs tgravescs Mar 30, 2016

Copy link
Copy Markdown
Contributor

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.

@zhonghaihua zhonghaihua Mar 31, 2016

Copy link
Copy Markdown
Contributor Author

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.
*
* @see SPARK-12864
*/
private var executorIdCounter: Int = {

@tgravescs tgravescs Mar 29, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

So I think there is a race condition here. For instance, AM dies right after it sends launch command for a new executor, if the AM comes up and get the currentExecutorId right before the executor registers, this number is going to be off by one (or possibly more then one if it sent launch commands for many). What happens in this case? Do we have a check to make sure we don't allow executor to register with same id?

@zhonghaihua zhonghaihua Mar 30, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

@tgravescs Thanks your comment. Yes. I think in this situation, when Am comes up, if the executor has already registered, the execuotr information must set in executorDataMap. And in executor registers process, it has a check to make sure we don't allow executor to register with same id.
You can see it in https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala#L139

@tgravescs tgravescs Mar 30, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Great, thanks

@vanzin vanzin Mar 30, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

There is a check in CoarseGrainedSchedulerBackend:

      case RegisterExecutor(executorId, executorRef, hostPort, cores, logUrls) =>
        if (executorDataMap.contains(executorId)) {
          context.reply(RegisterExecutorFailed("Duplicate executor ID: " + executorId))

That would cause the container failure count to go up, though. Probably ok? I don't see a way to fix that race without updating the driver with a synchronous RPC every time a new container is launched.

driverRef.askWithRetry[Int](RetrieveMaxExecutorId) + 1

@jerryshao jerryshao Feb 25, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Is it necessary to request the executor id every time allocating a new executor?

@zhonghaihua zhonghaihua Feb 25, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Hi @jerryshao , thanks for reviewing. Allocating a new executor won't execute that code. It just request the executor id when YarnAllocator being created.

@jerryshao jerryshao Feb 25, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I see, thanks.

@jerryshao jerryshao Feb 25, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I think here we should also make sure driverRef is already established, both in yarn client and cluster mode, did you test locally?

@zhonghaihua zhonghaihua Feb 25, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Yes, but just in yarn-client mode. Is it necessary to test in yarn-cluster mode ? @jerryshao

@zhonghaihua zhonghaihua Feb 25, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Hi @jerryshao , I made a test just now. It can work in yarn cluster mode. From my understanding, driverRef is created by ApplicationMaster, so, I think at this time driverEndpoint is started. What's your opinion ? Thanks a lot.

@jerryshao jerryshao Feb 25, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Great. Another thing is that current executor id is started from 1, not 0, we'd better keep that convention.

@zhonghaihua zhonghaihua Feb 25, 2016

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I see, thanks a lot.

@tgravescs tgravescs Mar 29, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I'm also a bit on the fence here whether we do this for cluster mode. We don't need to so its just an extra overhead but at the same time its nice not having a bunch of if checks.. @vanzin opinion?

@vanzin vanzin Mar 30, 2016

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I don't think it makes a difference; it's just one extra RPC call during initialization. In cluster mode that's a local call which should also be much cheaper.

An alternative would be to change the RegisterClusterManager message so that it returns this data to the AM (see AMEndpoint.onStart in ApplicationMaster.scala).

}
@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 RetrieveMaxExecutorId =>
context.reply(currentExecutorIdCounter)
}

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