ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Fix an instance where full standalone mode executor IDs were passed to

Browse source 
StandaloneSchedulerBackend instead of the smaller IDs used within Spark
(that lack the application name).

This was reported by ClearStory in
https://github.com/clearstorydata/spark/pull/9.

Also fixed some messages that said slave instead of executor.
This commit is contained in:
Matei Zaharia 2013-09-08 18:27:50 -07:00
parent 0b957997ad
commit f9b7f58de2
2 changed files with 7 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterTaskSetManager.scala
View file

@ -335,7 +335,7 @@ private[spark] class ClusterTaskSetManager(
}
/**
* Respond to an offer of a single slave from the scheduler by finding a task
* Respond to an offer of a single executor from the scheduler by finding a task
*/
override def resourceOffer(
execId: String,
@ -358,7 +358,7 @@ private[spark] class ClusterTaskSetManager(
val task = tasks(index)
val taskId = sched.newTaskId()
// Figure out whether this should count as a preferred launch
logInfo("Starting task %s:%d as TID %s on slave %s: %s (%s)".format(
logInfo("Starting task %s:%d as TID %s on executor %s: %s (%s)".format(
taskSet.id, index, taskId, execId, host, taskLocality))
// Do various bookkeeping
copiesRunning(index) += 1

10
core/src/main/scala/org/apache/spark/scheduler/cluster/SparkDeploySchedulerBackend.scala
View file

@ -76,17 +76,17 @@ private[spark] class SparkDeploySchedulerBackend(
}
}
override def executorAdded(executorId: String, workerId: String, hostPort: String, cores: Int, memory: Int) {
override def executorAdded(fullId: String, workerId: String, hostPort: String, cores: Int, memory: Int) {
logInfo("Granted executor ID %s on hostPort %s with %d cores, %s RAM".format(
executorId, hostPort, cores, Utils.megabytesToString(memory)))
fullId, hostPort, cores, Utils.megabytesToString(memory)))
}
override def executorRemoved(executorId: String, message: String, exitStatus: Option[Int]) {
override def executorRemoved(fullId: String, message: String, exitStatus: Option[Int]) {
val reason: ExecutorLossReason = exitStatus match {
case Some(code) => ExecutorExited(code)
case None => SlaveLost(message)
}
logInfo("Executor %s removed: %s".format(executorId, message))
removeExecutor(executorId, reason.toString)
logInfo("Executor %s removed: %s".format(fullId, message))
removeExecutor(fullId.split("/")(1), reason.toString)
}
}