You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ma...@apache.org on 2013/12/06 20:50:02 UTC
[01/12] git commit: Removed redundant residual re: reverted
refactoring.
Updated Branches:
refs/heads/master bfa68609d -> e0392343a
Removed redundant residual re: reverted refactoring.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/94087c46
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/94087c46
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/94087c46
Branch: refs/heads/master
Commit: 94087c463b41a92a9462b954f1f6452614569fe5
Parents: 982797d
Author: Mark Hamstra <ma...@gmail.com>
Authored: Wed Nov 20 15:47:30 2013 -0800
Committer: Mark Hamstra <ma...@gmail.com>
Committed: Tue Dec 3 09:57:31 2013 -0800
----------------------------------------------------------------------
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/94087c46/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index bf5827d..be46f74 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -384,7 +384,7 @@ class DAGScheduler(
.format(jobId, stageId))
} else {
jobSet -= jobId
- if ((jobSet - jobId).isEmpty) { // no other job needs this stage
+ if (jobSet.isEmpty) { // no other job needs this stage
p(stageId)
}
}
[05/12] git commit: Refactoring to make job removal, stage removal,
task cancellation clearer
Posted by ma...@apache.org.
Refactoring to make job removal, stage removal, task cancellation clearer
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/686a420d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/686a420d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/686a420d
Branch: refs/heads/master
Commit: 686a420ddc33407050d9019711cbe801fc352fa3
Parents: 205566e
Author: Mark Hamstra <ma...@gmail.com>
Authored: Fri Nov 22 10:20:09 2013 -0800
Committer: Mark Hamstra <ma...@gmail.com>
Committed: Tue Dec 3 09:57:32 2013 -0800
----------------------------------------------------------------------
.../apache/spark/scheduler/DAGScheduler.scala | 76 ++++++++++----------
1 file changed, 39 insertions(+), 37 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/686a420d/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index 6f9d4d5..b8b3ac0 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -370,9 +370,11 @@ class DAGScheduler(
}
}
- // Removes job and applies p to any stages that aren't needed by any other jobs
- private def forIndependentStagesOfRemovedJob(jobId: Int)(p: Int => Unit) {
+ // Removes job and any stages that are not needed by any other job. Returns the set of ids for stages that
+ // were removed and whose associated tasks may need to be cancelled.
+ private def removeJobAndIndependentStages(jobId: Int): Set[Int] = {
val registeredStages = jobIdToStageIds(jobId)
+ val independentStages = new HashSet[Int]()
if (registeredStages.isEmpty) {
logError("No stages registered for job " + jobId)
} else {
@@ -382,49 +384,51 @@ class DAGScheduler(
logError("Job %d not registered for stage %d even though that stage was registered for the job"
.format(jobId, stageId))
} else {
+ def removeStage(stageId: Int) {
+ // data structures based on Stage
+ stageIdToStage.get(stageId).foreach { s =>
+ if (running.contains(s)) {
+ logDebug("Removing running stage %d".format(stageId))
+ running -= s
+ }
+ stageToInfos -= s
+ shuffleToMapStage.keys.filter(shuffleToMapStage(_) == s).foreach(shuffleToMapStage.remove)
+ if (pendingTasks.contains(s) && !pendingTasks(s).isEmpty) {
+ logDebug("Removing pending status for stage %d".format(stageId))
+ }
+ pendingTasks -= s
+ if (waiting.contains(s)) {
+ logDebug("Removing stage %d from waiting set.".format(stageId))
+ waiting -= s
+ }
+ if (failed.contains(s)) {
+ logDebug("Removing stage %d from failed set.".format(stageId))
+ failed -= s
+ }
+ }
+ // data structures based on StageId
+ stageIdToStage -= stageId
+ stageIdToJobIds -= stageId
+
+ logDebug("After removal of stage %d, remaining stages = %d".format(stageId, stageIdToStage.size))
+ }
+
jobSet -= jobId
if (jobSet.isEmpty) { // no other job needs this stage
- p(stageId)
+ independentStages += stageId
+ removeStage(stageId)
}
}
}
}
- }
-
- private def removeStage(stageId: Int) {
- // data structures based on Stage
- stageIdToStage.get(stageId).foreach { s =>
- if (running.contains(s)) {
- logDebug("Removing running stage %d".format(stageId))
- running -= s
- }
- stageToInfos -= s
- shuffleToMapStage.keys.filter(shuffleToMapStage(_) == s).foreach(shuffleToMapStage.remove(_))
- if (pendingTasks.contains(s) && !pendingTasks(s).isEmpty) {
- logDebug("Removing pending status for stage %d".format(stageId))
- }
- pendingTasks -= s
- if (waiting.contains(s)) {
- logDebug("Removing stage %d from waiting set.".format(stageId))
- waiting -= s
- }
- if (failed.contains(s)) {
- logDebug("Removing stage %d from failed set.".format(stageId))
- failed -= s
- }
- }
- // data structures based on StageId
- stageIdToStage -= stageId
- stageIdToJobIds -= stageId
-
- logDebug("After removal of stage %d, remaining stages = %d".format(stageId, stageIdToStage.size))
+ independentStages.toSet
}
private def jobIdToStageIdsRemove(jobId: Int) {
if (!jobIdToStageIds.contains(jobId)) {
logDebug("Trying to remove unregistered job " + jobId)
} else {
- forIndependentStagesOfRemovedJob(jobId) { removeStage }
+ removeJobAndIndependentStages(jobId)
jobIdToStageIds -= jobId
}
}
@@ -987,10 +991,8 @@ class DAGScheduler(
if (!jobIdToStageIds.contains(jobId)) {
logDebug("Trying to cancel unregistered job " + jobId)
} else {
- forIndependentStagesOfRemovedJob(jobId) { stageId =>
- taskSched.cancelTasks(stageId)
- removeStage(stageId)
- }
+ val independentStages = removeJobAndIndependentStages(jobId)
+ independentStages.foreach { taskSched.cancelTasks }
val error = new SparkException("Job %d cancelled".format(jobId))
val job = idToActiveJob(jobId)
job.listener.jobFailed(error)
[04/12] git commit: Actor instead of eventQueue for LocalJobCompleted
Posted by ma...@apache.org.
Actor instead of eventQueue for LocalJobCompleted
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/6f8359b5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/6f8359b5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/6f8359b5
Branch: refs/heads/master
Commit: 6f8359b5ad6c069c6105631a6c74e225b866cfce
Parents: 51458ab
Author: Mark Hamstra <ma...@gmail.com>
Authored: Tue Nov 19 10:16:48 2013 -0800
Committer: Mark Hamstra <ma...@gmail.com>
Committed: Tue Dec 3 09:57:31 2013 -0800
----------------------------------------------------------------------
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/6f8359b5/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index 10417b9..ad436f8 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -694,7 +694,7 @@ class DAGScheduler(
case e: Exception =>
job.listener.jobFailed(e)
} finally {
- eventQueue.put(LocalJobCompleted(job.finalStage))
+ eventProcessActor ! LocalJobCompleted(job.finalStage)
}
}
[12/12] git commit: Merge pull request #190 from
markhamstra/Stages4Jobs
Posted by ma...@apache.org.
Merge pull request #190 from markhamstra/Stages4Jobs
stageId <--> jobId mapping in DAGScheduler
Okay, I think this one is ready to go -- or at least it's ready for review and discussion. It's a carry-over of https://github.com/mesos/spark/pull/842 with updates for the newer job cancellation functionality. The prior discussion still applies. I've actually changed the job cancellation flow a bit: Instead of ``cancelTasks`` going to the TaskScheduler and then ``taskSetFailed`` coming back to the DAGScheduler (resulting in ``abortStage`` there), the DAGScheduler now takes care of figuring out which stages should be cancelled, tells the TaskScheduler to cancel tasks for those stages, then does the cleanup within the DAGScheduler directly without the need for any further prompting by the TaskScheduler.
I know of three outstanding issues, each of which can and should, I believe, be handled in follow-up pull requests:
1) https://spark-project.atlassian.net/browse/SPARK-960
2) JobLogger should be re-factored to eliminate duplication
3) Related to 2), the WebUI should also become a consumer of the DAGScheduler's new understanding of the relationship between jobs and stages so that it can display progress indication and the like grouped by job. Right now, some of this information is just being sent out as part of ``SparkListenerJobStart`` messages, but more or different job <--> stage information may need to be exported from the DAGScheduler to meet listeners needs.
Except for the eventQueue -> Actor commit, the rest can be cherry-picked almost cleanly into branch-0.8. A little merging is needed in MapOutputTracker and the DAGScheduler. Merged versions of those files are in https://github.com/markhamstra/incubator-spark/tree/aba2b40ce04ee9b7b9ea260abb6f09e050142d43
Note that between the recent Actor change in the DAGScheduler and the cleaning up of DAGScheduler data structures on job completion in this PR, some races have been introduced into the DAGSchedulerSuite. Those tests usually pass, and I don't think that better-behaved code that doesn't directly inspect DAGScheduler data structures should be seeing any problems, but I'll work on fixing DAGSchedulerSuite as either an addition to this PR or as a separate request.
UPDATE: Fixed the race that I introduced. Created a JIRA issue (SPARK-965) for the one that was introduced with the switch to eventProcessorActor in the DAGScheduler.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/e0392343
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/e0392343
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/e0392343
Branch: refs/heads/master
Commit: e0392343a026d632bac0df0ad4f399fce742c151
Parents: bfa6860 403234d
Author: Matei Zaharia <ma...@eecs.berkeley.edu>
Authored: Fri Dec 6 11:49:59 2013 -0800
Committer: Matei Zaharia <ma...@eecs.berkeley.edu>
Committed: Fri Dec 6 11:49:59 2013 -0800
----------------------------------------------------------------------
.../org/apache/spark/MapOutputTracker.scala | 8 +-
.../apache/spark/scheduler/DAGScheduler.scala | 278 +++++++++++++++----
.../spark/scheduler/DAGSchedulerEvent.scala | 3 +-
.../apache/spark/scheduler/SparkListener.scala | 2 +-
.../scheduler/cluster/ClusterScheduler.scala | 4 +-
.../cluster/ClusterTaskSetManager.scala | 2 +-
.../spark/scheduler/local/LocalScheduler.scala | 27 +-
.../org/apache/spark/JobCancellationSuite.scala | 4 +-
.../spark/scheduler/DAGSchedulerSuite.scala | 43 ++-
9 files changed, 280 insertions(+), 91 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/e0392343/core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterScheduler.scala
----------------------------------------------------------------------
[02/12] git commit: Added stageId <--> jobId mapping in DAGScheduler
...and make sure that DAGScheduler data structures are cleaned up on job
completion. Initial effort and discussion at
https://github.com/mesos/spark/pull/842
Posted by ma...@apache.org.
Added stageId <--> jobId mapping in DAGScheduler
...and make sure that DAGScheduler data structures are cleaned up on job completion.
Initial effort and discussion at https://github.com/mesos/spark/pull/842
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/51458ab4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/51458ab4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/51458ab4
Branch: refs/heads/master
Commit: 51458ab4a16a2d365f5de756d2fac942b766feca
Parents: 58d9bbc
Author: Mark Hamstra <ma...@gmail.com>
Authored: Mon Nov 11 16:06:12 2013 -0800
Committer: Mark Hamstra <ma...@gmail.com>
Committed: Tue Dec 3 09:57:31 2013 -0800
----------------------------------------------------------------------
.../org/apache/spark/MapOutputTracker.scala | 8 +-
.../apache/spark/scheduler/DAGScheduler.scala | 277 +++++++++++++++----
.../spark/scheduler/DAGSchedulerEvent.scala | 5 +-
.../apache/spark/scheduler/SparkListener.scala | 2 +-
.../scheduler/cluster/ClusterScheduler.scala | 4 +-
.../cluster/ClusterTaskSetManager.scala | 2 +-
.../spark/scheduler/local/LocalScheduler.scala | 27 +-
.../org/apache/spark/JobCancellationSuite.scala | 4 +-
.../spark/scheduler/DAGSchedulerSuite.scala | 45 ++-
9 files changed, 286 insertions(+), 88 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/51458ab4/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
index 5e465fa..b4d0b70 100644
--- a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
+++ b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
@@ -244,12 +244,12 @@ private[spark] class MapOutputTrackerMaster extends MapOutputTracker {
case Some(bytes) =>
return bytes
case None =>
- statuses = mapStatuses(shuffleId)
+ statuses = mapStatuses.getOrElse(shuffleId, Array[MapStatus]())
epochGotten = epoch
}
}
// If we got here, we failed to find the serialized locations in the cache, so we pulled
- // out a snapshot of the locations as "locs"; let's serialize and return that
+ // out a snapshot of the locations as "statuses"; let's serialize and return that
val bytes = MapOutputTracker.serializeMapStatuses(statuses)
logInfo("Size of output statuses for shuffle %d is %d bytes".format(shuffleId, bytes.length))
// Add them into the table only if the epoch hasn't changed while we were working
@@ -274,6 +274,10 @@ private[spark] class MapOutputTrackerMaster extends MapOutputTracker {
override def updateEpoch(newEpoch: Long) {
// This might be called on the MapOutputTrackerMaster if we're running in local mode.
}
+
+ def has(shuffleId: Int): Boolean = {
+ cachedSerializedStatuses.get(shuffleId).isDefined || mapStatuses.contains(shuffleId)
+ }
}
private[spark] object MapOutputTracker {
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/51458ab4/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index a785a16..10417b9 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -121,9 +121,13 @@ class DAGScheduler(
private val nextStageId = new AtomicInteger(0)
- private val stageIdToStage = new TimeStampedHashMap[Int, Stage]
+ private[scheduler] val jobIdToStageIds = new TimeStampedHashMap[Int, HashSet[Int]]
- private val shuffleToMapStage = new TimeStampedHashMap[Int, Stage]
+ private[scheduler] val stageIdToJobIds = new TimeStampedHashMap[Int, HashSet[Int]]
+
+ private[scheduler] val stageIdToStage = new TimeStampedHashMap[Int, Stage]
+
+ private[scheduler] val shuffleToMapStage = new TimeStampedHashMap[Int, Stage]
private[spark] val stageToInfos = new TimeStampedHashMap[Stage, StageInfo]
@@ -232,7 +236,7 @@ class DAGScheduler(
shuffleToMapStage.get(shuffleDep.shuffleId) match {
case Some(stage) => stage
case None =>
- val stage = newStage(shuffleDep.rdd, shuffleDep.rdd.partitions.size, Some(shuffleDep), jobId)
+ val stage = newOrUsedStage(shuffleDep.rdd, shuffleDep.rdd.partitions.size, shuffleDep, jobId)
shuffleToMapStage(shuffleDep.shuffleId) = stage
stage
}
@@ -241,7 +245,8 @@ class DAGScheduler(
/**
* Create a Stage for the given RDD, either as a shuffle map stage (for a ShuffleDependency) or
* as a result stage for the final RDD used directly in an action. The stage will also be
- * associated with the provided jobId.
+ * associated with the provided jobId.. Shuffle map stages, whose shuffleId may have previously
+ * been registered in the MapOutputTracker, should be (re)-created using newOrUsedStage.
*/
private def newStage(
rdd: RDD[_],
@@ -251,21 +256,45 @@ class DAGScheduler(
callSite: Option[String] = None)
: Stage =
{
- if (shuffleDep != None) {
- // Kind of ugly: need to register RDDs with the cache and map output tracker here
- // since we can't do it in the RDD constructor because # of partitions is unknown
- logInfo("Registering RDD " + rdd.id + " (" + rdd.origin + ")")
- mapOutputTracker.registerShuffle(shuffleDep.get.shuffleId, rdd.partitions.size)
- }
val id = nextStageId.getAndIncrement()
val stage =
new Stage(id, rdd, numTasks, shuffleDep, getParentStages(rdd, jobId), jobId, callSite)
stageIdToStage(id) = stage
+ registerJobIdWithStages(jobId, stage)
stageToInfos(stage) = new StageInfo(stage)
stage
}
/**
+ * Create a shuffle map Stage for the given RDD. The stage will also be associated with the
+ * provided jobId. If a stage for the shuffleId existed previously so that the shuffleId is
+ * present in the MapOutputTracker, then the number and location of available outputs are
+ * recovered from the MapOutputTracker
+ */
+ private def newOrUsedStage(
+ rdd: RDD[_],
+ numTasks: Int,
+ shuffleDep: ShuffleDependency[_,_],
+ jobId: Int,
+ callSite: Option[String] = None)
+ : Stage =
+ {
+ val stage = newStage(rdd, numTasks, Some(shuffleDep), jobId, callSite)
+ if (mapOutputTracker.has(shuffleDep.shuffleId)) {
+ val serLocs = mapOutputTracker.getSerializedMapOutputStatuses(shuffleDep.shuffleId)
+ val locs = MapOutputTracker.deserializeMapStatuses(serLocs)
+ for (i <- 0 until locs.size) stage.outputLocs(i) = List(locs(i))
+ stage.numAvailableOutputs = locs.size
+ } else {
+ // Kind of ugly: need to register RDDs with the cache and map output tracker here
+ // since we can't do it in the RDD constructor because # of partitions is unknown
+ logInfo("Registering RDD " + rdd.id + " (" + rdd.origin + ")")
+ mapOutputTracker.registerShuffle(shuffleDep.shuffleId, rdd.partitions.size)
+ }
+ stage
+ }
+
+ /**
* Get or create the list of parent stages for a given RDD. The stages will be assigned the
* provided jobId if they haven't already been created with a lower jobId.
*/
@@ -317,6 +346,91 @@ class DAGScheduler(
}
/**
+ * Registers the given jobId among the jobs that need the given stage and
+ * all of that stage's ancestors.
+ */
+ private def registerJobIdWithStages(jobId: Int, stage: Stage) {
+ def registerJobIdWithStageList(stages: List[Stage]) {
+ if (!stages.isEmpty) {
+ val s = stages.head
+ stageIdToJobIds.getOrElseUpdate(s.id, new HashSet[Int]()) += jobId
+ val parents = getParentStages(s.rdd, jobId)
+ val parentsWithoutThisJobId = parents.filter(p => !stageIdToJobIds.get(p.id).exists(_.contains(jobId)))
+ registerJobIdWithStageList(parentsWithoutThisJobId ++ stages.tail)
+ }
+ }
+ registerJobIdWithStageList(List(stage))
+ }
+
+ private def jobIdToStageIdsAdd(jobId: Int) {
+ val stageSet = jobIdToStageIds.getOrElseUpdate(jobId, new HashSet[Int]())
+ stageIdToJobIds.foreach { case (stageId, jobSet) =>
+ if (jobSet.contains(jobId)) {
+ stageSet += stageId
+ }
+ }
+ }
+
+ // Removes job and applies p to any stages that aren't needed by any other jobs
+ private def forIndependentStagesOfRemovedJob(jobId: Int)(p: Int => Unit) {
+ val registeredStages = jobIdToStageIds(jobId)
+ if (registeredStages.isEmpty) {
+ logError("No stages registered for job " + jobId)
+ } else {
+ stageIdToJobIds.filterKeys(stageId => registeredStages.contains(stageId)).foreach {
+ case (stageId, jobSet) =>
+ if (!jobSet.contains(jobId)) {
+ logError("Job %d not registered for stage %d even though that stage was registered for the job"
+ .format(jobId, stageId))
+ } else {
+ jobSet -= jobId
+ if ((jobSet - jobId).isEmpty) { // no other job needs this stage
+ p(stageId)
+ }
+ }
+ }
+ }
+ }
+
+ private def removeStage(stageId: Int) {
+ // data structures based on Stage
+ stageIdToStage.get(stageId).foreach { s =>
+ if (running.contains(s)) {
+ logDebug("Removing running stage %d".format(stageId))
+ running -= s
+ }
+ stageToInfos -= s
+ shuffleToMapStage.keys.filter(shuffleToMapStage(_) == s).foreach(shuffleToMapStage.remove(_))
+ if (pendingTasks.contains(s) && !pendingTasks(s).isEmpty) {
+ logDebug("Removing pending status for stage %d".format(stageId))
+ }
+ pendingTasks -= s
+ if (waiting.contains(s)) {
+ logDebug("Removing stage %d from waiting set.".format(stageId))
+ waiting -= s
+ }
+ if (failed.contains(s)) {
+ logDebug("Removing stage %d from failed set.".format(stageId))
+ failed -= s
+ }
+ }
+ // data structures based on StageId
+ stageIdToStage -= stageId
+ stageIdToJobIds -= stageId
+
+ logDebug("After removal of stage %d, remaining stages = %d".format(stageId, stageIdToStage.size))
+ }
+
+ private def jobIdToStageIdsRemove(jobId: Int) {
+ if (!jobIdToStageIds.contains(jobId)) {
+ logDebug("Trying to remove unregistered job " + jobId)
+ } else {
+ forIndependentStagesOfRemovedJob(jobId) { removeStage }
+ jobIdToStageIds -= jobId
+ }
+ }
+
+ /**
* Submit a job to the job scheduler and get a JobWaiter object back. The JobWaiter object
* can be used to block until the the job finishes executing or can be used to cancel the job.
*/
@@ -435,35 +549,33 @@ class DAGScheduler(
// Compute very short actions like first() or take() with no parent stages locally.
runLocally(job)
} else {
- listenerBus.post(SparkListenerJobStart(job, properties))
idToActiveJob(jobId) = job
activeJobs += job
resultStageToJob(finalStage) = job
+ jobIdToStageIdsAdd(jobId)
+ listenerBus.post(SparkListenerJobStart(job, jobIdToStageIds(jobId).toArray, properties))
submitStage(finalStage)
}
case JobCancelled(jobId) =>
- // Cancel a job: find all the running stages that are linked to this job, and cancel them.
- running.filter(_.jobId == jobId).foreach { stage =>
- taskSched.cancelTasks(stage.id)
- }
+ handleJobCancellation(jobId)
+ idToActiveJob.get(jobId).foreach(job => activeJobs -= job)
+ idToActiveJob -= jobId
case JobGroupCancelled(groupId) =>
// Cancel all jobs belonging to this job group.
// First finds all active jobs with this group id, and then kill stages for them.
- val jobIds = activeJobs.filter(groupId == _.properties.get(SparkContext.SPARK_JOB_GROUP_ID))
- .map(_.jobId)
- if (!jobIds.isEmpty) {
- running.filter(stage => jobIds.contains(stage.jobId)).foreach { stage =>
- taskSched.cancelTasks(stage.id)
- }
- }
+ val activeInGroup = activeJobs.filter(groupId == _.properties.get(SparkContext.SPARK_JOB_GROUP_ID))
+ val jobIds = activeInGroup.map(_.jobId)
+ jobIds.foreach { handleJobCancellation }
+ activeJobs -- activeInGroup
+ idToActiveJob -- jobIds
case AllJobsCancelled =>
// Cancel all running jobs.
- running.foreach { stage =>
- taskSched.cancelTasks(stage.id)
- }
+ running.map(_.jobId).foreach { handleJobCancellation }
+ activeJobs.clear()
+ idToActiveJob.clear()
case ExecutorGained(execId, host) =>
handleExecutorGained(execId, host)
@@ -493,8 +605,13 @@ class DAGScheduler(
listenerBus.post(SparkListenerTaskEnd(task, reason, taskInfo, taskMetrics))
handleTaskCompletion(completion)
+ case LocalJobCompleted(stage) =>
+ stageIdToJobIds -= stage.id // clean up data structures that were populated for a local job,
+ stageIdToStage -= stage.id // but that won't get cleaned up via the normal paths through
+ stageToInfos -= stage // completion events or stage abort
+
case TaskSetFailed(taskSet, reason) =>
- abortStage(stageIdToStage(taskSet.stageId), reason)
+ stageIdToStage.get(taskSet.stageId).foreach { abortStage(_, reason) }
case ResubmitFailedStages =>
if (failed.size > 0) {
@@ -576,30 +693,52 @@ class DAGScheduler(
} catch {
case e: Exception =>
job.listener.jobFailed(e)
+ } finally {
+ eventQueue.put(LocalJobCompleted(job.finalStage))
+ }
+ }
+
+ /** Finds the earliest-created active job that needs the stage */
+ // TODO: Probably should actually find among the active jobs that need this
+ // stage the one with the highest priority (highest-priority pool, earliest created).
+ // That should take care of at least part of the priority inversion problem with
+ // cross-job dependencies.
+ private def activeJobForStage(stage: Stage): Option[Int] = {
+ if (stageIdToJobIds.contains(stage.id)) {
+ val jobsThatUseStage: Array[Int] = stageIdToJobIds(stage.id).toArray.sorted
+ jobsThatUseStage.find(idToActiveJob.contains(_))
+ } else {
+ None
}
}
/** Submits stage, but first recursively submits any missing parents. */
private def submitStage(stage: Stage) {
- logDebug("submitStage(" + stage + ")")
- if (!waiting(stage) && !running(stage) && !failed(stage)) {
- val missing = getMissingParentStages(stage).sortBy(_.id)
- logDebug("missing: " + missing)
- if (missing == Nil) {
- logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents")
- submitMissingTasks(stage)
- running += stage
- } else {
- for (parent <- missing) {
- submitStage(parent)
+ val jobId = activeJobForStage(stage)
+ if (jobId.isDefined) {
+ logDebug("submitStage(" + stage + ")")
+ if (!waiting(stage) && !running(stage) && !failed(stage)) {
+ val missing = getMissingParentStages(stage).sortBy(_.id)
+ logDebug("missing: " + missing)
+ if (missing == Nil) {
+ logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents")
+ submitMissingTasks(stage, jobId.get)
+ running += stage
+ } else {
+ for (parent <- missing) {
+ submitStage(parent)
+ }
+ waiting += stage
}
- waiting += stage
}
+ } else {
+ abortStage(stage, "No active job for stage " + stage.id)
}
}
+
/** Called when stage's parents are available and we can now do its task. */
- private def submitMissingTasks(stage: Stage) {
+ private def submitMissingTasks(stage: Stage, jobId: Int) {
logDebug("submitMissingTasks(" + stage + ")")
// Get our pending tasks and remember them in our pendingTasks entry
val myPending = pendingTasks.getOrElseUpdate(stage, new HashSet)
@@ -620,7 +759,7 @@ class DAGScheduler(
}
}
- val properties = if (idToActiveJob.contains(stage.jobId)) {
+ val properties = if (idToActiveJob.contains(jobId)) {
idToActiveJob(stage.jobId).properties
} else {
//this stage will be assigned to "default" pool
@@ -703,6 +842,7 @@ class DAGScheduler(
resultStageToJob -= stage
markStageAsFinished(stage)
listenerBus.post(SparkListenerJobEnd(job, JobSucceeded))
+ jobIdToStageIdsRemove(job.jobId)
}
job.listener.taskSucceeded(rt.outputId, event.result)
}
@@ -738,7 +878,7 @@ class DAGScheduler(
changeEpoch = true)
}
clearCacheLocs()
- if (stage.outputLocs.count(_ == Nil) != 0) {
+ if (stage.outputLocs.exists(_ == Nil)) {
// Some tasks had failed; let's resubmit this stage
// TODO: Lower-level scheduler should also deal with this
logInfo("Resubmitting " + stage + " (" + stage.name +
@@ -755,9 +895,12 @@ class DAGScheduler(
}
waiting --= newlyRunnable
running ++= newlyRunnable
- for (stage <- newlyRunnable.sortBy(_.id)) {
+ for {
+ stage <- newlyRunnable.sortBy(_.id)
+ jobId <- activeJobForStage(stage)
+ } {
logInfo("Submitting " + stage + " (" + stage.rdd + "), which is now runnable")
- submitMissingTasks(stage)
+ submitMissingTasks(stage, jobId)
}
}
}
@@ -841,11 +984,31 @@ class DAGScheduler(
}
}
+ private def handleJobCancellation(jobId: Int) {
+ if (!jobIdToStageIds.contains(jobId)) {
+ logDebug("Trying to cancel unregistered job " + jobId)
+ } else {
+ forIndependentStagesOfRemovedJob(jobId) { stageId =>
+ taskSched.cancelTasks(stageId)
+ removeStage(stageId)
+ }
+ val error = new SparkException("Job %d cancelled".format(jobId))
+ val job = idToActiveJob(jobId)
+ job.listener.jobFailed(error)
+ listenerBus.post(SparkListenerJobEnd(job, JobFailed(error, Some(job.finalStage))))
+ jobIdToStageIds -= jobId
+ }
+ }
+
/**
* Aborts all jobs depending on a particular Stage. This is called in response to a task set
* being canceled by the TaskScheduler. Use taskSetFailed() to inject this event from outside.
*/
private def abortStage(failedStage: Stage, reason: String) {
+ if (!stageIdToStage.contains(failedStage.id)) {
+ // Skip all the actions if the stage has been removed.
+ return
+ }
val dependentStages = resultStageToJob.keys.filter(x => stageDependsOn(x, failedStage)).toSeq
stageToInfos(failedStage).completionTime = Some(System.currentTimeMillis())
for (resultStage <- dependentStages) {
@@ -853,6 +1016,7 @@ class DAGScheduler(
val error = new SparkException("Job aborted: " + reason)
job.listener.jobFailed(error)
listenerBus.post(SparkListenerJobEnd(job, JobFailed(error, Some(failedStage))))
+ jobIdToStageIdsRemove(job.jobId)
idToActiveJob -= resultStage.jobId
activeJobs -= job
resultStageToJob -= resultStage
@@ -926,21 +1090,18 @@ class DAGScheduler(
}
private def cleanup(cleanupTime: Long) {
- var sizeBefore = stageIdToStage.size
- stageIdToStage.clearOldValues(cleanupTime)
- logInfo("stageIdToStage " + sizeBefore + " --> " + stageIdToStage.size)
-
- sizeBefore = shuffleToMapStage.size
- shuffleToMapStage.clearOldValues(cleanupTime)
- logInfo("shuffleToMapStage " + sizeBefore + " --> " + shuffleToMapStage.size)
-
- sizeBefore = pendingTasks.size
- pendingTasks.clearOldValues(cleanupTime)
- logInfo("pendingTasks " + sizeBefore + " --> " + pendingTasks.size)
-
- sizeBefore = stageToInfos.size
- stageToInfos.clearOldValues(cleanupTime)
- logInfo("stageToInfos " + sizeBefore + " --> " + stageToInfos.size)
+ Map(
+ "stageIdToStage" -> stageIdToStage,
+ "shuffleToMapStage" -> shuffleToMapStage,
+ "pendingTasks" -> pendingTasks,
+ "stageToInfos" -> stageToInfos,
+ "jobIdToStageIds" -> jobIdToStageIds,
+ "stageIdToJobIds" -> stageIdToJobIds).
+ foreach { case(s, t) => {
+ val sizeBefore = t.size
+ t.clearOldValues(cleanupTime)
+ logInfo("%s %d --> %d".format(s, sizeBefore, t.size))
+ }}
}
def stop() {
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/51458ab4/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
index 5353cd2..bf8dfb5 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
@@ -65,8 +65,9 @@ private[scheduler] case class CompletionEvent(
taskMetrics: TaskMetrics)
extends DAGSchedulerEvent
-private[scheduler]
-case class ExecutorGained(execId: String, host: String) extends DAGSchedulerEvent
+private[scheduler] case class LocalJobCompleted(stage: Stage) extends DAGSchedulerEvent
+
+private[scheduler] case class ExecutorGained(execId: String, host: String) extends DAGSchedulerEvent
private[scheduler] case class ExecutorLost(execId: String) extends DAGSchedulerEvent
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/51458ab4/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala b/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala
index a35081f..3841b56 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/SparkListener.scala
@@ -37,7 +37,7 @@ case class SparkListenerTaskGettingResult(
case class SparkListenerTaskEnd(task: Task[_], reason: TaskEndReason, taskInfo: TaskInfo,
taskMetrics: TaskMetrics) extends SparkListenerEvents
-case class SparkListenerJobStart(job: ActiveJob, properties: Properties = null)
+case class SparkListenerJobStart(job: ActiveJob, stageIds: Array[Int], properties: Properties = null)
extends SparkListenerEvents
case class SparkListenerJobEnd(job: ActiveJob, jobResult: JobResult)
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/51458ab4/core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterScheduler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterScheduler.scala
index c1e65a3..bd0a39b 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterScheduler.scala
@@ -173,7 +173,9 @@ private[spark] class ClusterScheduler(val sc: SparkContext)
backend.killTask(tid, execId)
}
}
- tsm.error("Stage %d was cancelled".format(stageId))
+ logInfo("Stage %d was cancelled".format(stageId))
+ tsm.removeAllRunningTasks()
+ taskSetFinished(tsm)
}
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/51458ab4/core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterTaskSetManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterTaskSetManager.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterTaskSetManager.scala
index 8884ea8..9496179 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterTaskSetManager.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterTaskSetManager.scala
@@ -574,7 +574,7 @@ private[spark] class ClusterTaskSetManager(
runningTasks = runningTasksSet.size
}
- private def removeAllRunningTasks() {
+ private[cluster] def removeAllRunningTasks() {
val numRunningTasks = runningTasksSet.size
runningTasksSet.clear()
if (parent != null) {
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/51458ab4/core/src/main/scala/org/apache/spark/scheduler/local/LocalScheduler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/local/LocalScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/local/LocalScheduler.scala
index 5af5116..01e9516 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/local/LocalScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/local/LocalScheduler.scala
@@ -144,7 +144,8 @@ private[spark] class LocalScheduler(val threads: Int, val maxFailures: Int, val
localActor ! KillTask(tid)
}
}
- tsm.error("Stage %d was cancelled".format(stageId))
+ logInfo("Stage %d was cancelled".format(stageId))
+ taskSetFinished(tsm)
}
}
@@ -192,17 +193,19 @@ private[spark] class LocalScheduler(val threads: Int, val maxFailures: Int, val
synchronized {
taskIdToTaskSetId.get(taskId) match {
case Some(taskSetId) =>
- val taskSetManager = activeTaskSets(taskSetId)
- taskSetTaskIds(taskSetId) -= taskId
-
- state match {
- case TaskState.FINISHED =>
- taskSetManager.taskEnded(taskId, state, serializedData)
- case TaskState.FAILED =>
- taskSetManager.taskFailed(taskId, state, serializedData)
- case TaskState.KILLED =>
- taskSetManager.error("Task %d was killed".format(taskId))
- case _ => {}
+ val taskSetManager = activeTaskSets.get(taskSetId)
+ taskSetManager.foreach { tsm =>
+ taskSetTaskIds(taskSetId) -= taskId
+
+ state match {
+ case TaskState.FINISHED =>
+ tsm.taskEnded(taskId, state, serializedData)
+ case TaskState.FAILED =>
+ tsm.taskFailed(taskId, state, serializedData)
+ case TaskState.KILLED =>
+ tsm.error("Task %d was killed".format(taskId))
+ case _ => {}
+ }
}
case None =>
logInfo("Ignoring update from TID " + taskId + " because its task set is gone")
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/51458ab4/core/src/test/scala/org/apache/spark/JobCancellationSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/JobCancellationSuite.scala b/core/src/test/scala/org/apache/spark/JobCancellationSuite.scala
index d8a0e98..1121e06 100644
--- a/core/src/test/scala/org/apache/spark/JobCancellationSuite.scala
+++ b/core/src/test/scala/org/apache/spark/JobCancellationSuite.scala
@@ -114,7 +114,7 @@ class JobCancellationSuite extends FunSuite with ShouldMatchers with BeforeAndAf
// Once A is cancelled, job B should finish fairly quickly.
assert(jobB.get() === 100)
}
-
+/*
test("two jobs sharing the same stage") {
// sem1: make sure cancel is issued after some tasks are launched
// sem2: make sure the first stage is not finished until cancel is issued
@@ -148,7 +148,7 @@ class JobCancellationSuite extends FunSuite with ShouldMatchers with BeforeAndAf
intercept[SparkException] { f1.get() }
intercept[SparkException] { f2.get() }
}
-
+ */
def testCount() {
// Cancel before launching any tasks
{
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/51458ab4/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
index a4d41eb..8ce8c68 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
@@ -206,6 +206,7 @@ class DAGSchedulerSuite extends FunSuite with BeforeAndAfter with LocalSparkCont
submit(rdd, Array(0))
complete(taskSets(0), List((Success, 42)))
assert(results === Map(0 -> 42))
+ assertDataStructuresEmpty
}
test("local job") {
@@ -218,7 +219,10 @@ class DAGSchedulerSuite extends FunSuite with BeforeAndAfter with LocalSparkCont
}
val jobId = scheduler.nextJobId.getAndIncrement()
runEvent(JobSubmitted(jobId, rdd, jobComputeFunc, Array(0), true, null, listener))
+ assert(scheduler.stageToInfos.size === 1)
+ runEvent(LocalJobCompleted(scheduler.stageToInfos.keys.head))
assert(results === Map(0 -> 42))
+ assertDataStructuresEmpty
}
test("run trivial job w/ dependency") {
@@ -227,6 +231,7 @@ class DAGSchedulerSuite extends FunSuite with BeforeAndAfter with LocalSparkCont
submit(finalRdd, Array(0))
complete(taskSets(0), Seq((Success, 42)))
assert(results === Map(0 -> 42))
+ assertDataStructuresEmpty
}
test("cache location preferences w/ dependency") {
@@ -239,12 +244,14 @@ class DAGSchedulerSuite extends FunSuite with BeforeAndAfter with LocalSparkCont
assertLocations(taskSet, Seq(Seq("hostA", "hostB")))
complete(taskSet, Seq((Success, 42)))
assert(results === Map(0 -> 42))
+ assertDataStructuresEmpty
}
test("trivial job failure") {
submit(makeRdd(1, Nil), Array(0))
failed(taskSets(0), "some failure")
assert(failure.getMessage === "Job aborted: some failure")
+ assertDataStructuresEmpty
}
test("run trivial shuffle") {
@@ -260,6 +267,7 @@ class DAGSchedulerSuite extends FunSuite with BeforeAndAfter with LocalSparkCont
Array(makeBlockManagerId("hostA"), makeBlockManagerId("hostB")))
complete(taskSets(1), Seq((Success, 42)))
assert(results === Map(0 -> 42))
+ assertDataStructuresEmpty
}
test("run trivial shuffle with fetch failure") {
@@ -285,6 +293,7 @@ class DAGSchedulerSuite extends FunSuite with BeforeAndAfter with LocalSparkCont
assert(mapOutputTracker.getServerStatuses(shuffleId, 0).map(_._1.host) === Array("hostA", "hostB"))
complete(taskSets(3), Seq((Success, 43)))
assert(results === Map(0 -> 42, 1 -> 43))
+ assertDataStructuresEmpty
}
test("ignore late map task completions") {
@@ -313,6 +322,7 @@ class DAGSchedulerSuite extends FunSuite with BeforeAndAfter with LocalSparkCont
Array(makeBlockManagerId("hostB"), makeBlockManagerId("hostA")))
complete(taskSets(1), Seq((Success, 42), (Success, 43)))
assert(results === Map(0 -> 42, 1 -> 43))
+ assertDataStructuresEmpty
}
test("run trivial shuffle with out-of-band failure and retry") {
@@ -329,15 +339,16 @@ class DAGSchedulerSuite extends FunSuite with BeforeAndAfter with LocalSparkCont
complete(taskSets(0), Seq(
(Success, makeMapStatus("hostA", 1)),
(Success, makeMapStatus("hostB", 1))))
- // have hostC complete the resubmitted task
- complete(taskSets(1), Seq((Success, makeMapStatus("hostC", 1))))
- assert(mapOutputTracker.getServerStatuses(shuffleId, 0).map(_._1) ===
- Array(makeBlockManagerId("hostC"), makeBlockManagerId("hostB")))
- complete(taskSets(2), Seq((Success, 42)))
- assert(results === Map(0 -> 42))
- }
-
- test("recursive shuffle failures") {
+ // have hostC complete the resubmitted task
+ complete(taskSets(1), Seq((Success, makeMapStatus("hostC", 1))))
+ assert(mapOutputTracker.getServerStatuses(shuffleId, 0).map(_._1) ===
+ Array(makeBlockManagerId("hostC"), makeBlockManagerId("hostB")))
+ complete(taskSets(2), Seq((Success, 42)))
+ assert(results === Map(0 -> 42))
+ assertDataStructuresEmpty
+ }
+
+ test("recursive shuffle failures") {
val shuffleOneRdd = makeRdd(2, Nil)
val shuffleDepOne = new ShuffleDependency(shuffleOneRdd, null)
val shuffleTwoRdd = makeRdd(2, List(shuffleDepOne))
@@ -363,6 +374,7 @@ class DAGSchedulerSuite extends FunSuite with BeforeAndAfter with LocalSparkCont
complete(taskSets(4), Seq((Success, makeMapStatus("hostA", 1))))
complete(taskSets(5), Seq((Success, 42)))
assert(results === Map(0 -> 42))
+ assertDataStructuresEmpty
}
test("cached post-shuffle") {
@@ -394,6 +406,7 @@ class DAGSchedulerSuite extends FunSuite with BeforeAndAfter with LocalSparkCont
complete(taskSets(3), Seq((Success, makeMapStatus("hostD", 1))))
complete(taskSets(4), Seq((Success, 42)))
assert(results === Map(0 -> 42))
+ assertDataStructuresEmpty
}
/**
@@ -413,4 +426,18 @@ class DAGSchedulerSuite extends FunSuite with BeforeAndAfter with LocalSparkCont
private def makeBlockManagerId(host: String): BlockManagerId =
BlockManagerId("exec-" + host, host, 12345, 0)
+ private def assertDataStructuresEmpty = {
+ assert(scheduler.pendingTasks.isEmpty)
+ assert(scheduler.activeJobs.isEmpty)
+ assert(scheduler.failed.isEmpty)
+ assert(scheduler.idToActiveJob.isEmpty)
+ assert(scheduler.jobIdToStageIds.isEmpty)
+ assert(scheduler.stageIdToJobIds.isEmpty)
+ assert(scheduler.stageIdToStage.isEmpty)
+ assert(scheduler.stageToInfos.isEmpty)
+ assert(scheduler.resultStageToJob.isEmpty)
+ assert(scheduler.running.isEmpty)
+ assert(scheduler.shuffleToMapStage.isEmpty)
+ assert(scheduler.waiting.isEmpty)
+ }
}
[03/12] git commit: Fixed intended side-effects
Posted by ma...@apache.org.
Fixed intended side-effects
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/982797dc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/982797dc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/982797dc
Branch: refs/heads/master
Commit: 982797dcbafa4c1149ad354b0c5a07e3f74fe005
Parents: 6f8359b
Author: Mark Hamstra <ma...@gmail.com>
Authored: Tue Nov 19 16:59:42 2013 -0800
Committer: Mark Hamstra <ma...@gmail.com>
Committed: Tue Dec 3 09:57:31 2013 -0800
----------------------------------------------------------------------
.../src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/982797dc/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index ad436f8..bf5827d 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -568,8 +568,8 @@ class DAGScheduler(
val activeInGroup = activeJobs.filter(groupId == _.properties.get(SparkContext.SPARK_JOB_GROUP_ID))
val jobIds = activeInGroup.map(_.jobId)
jobIds.foreach { handleJobCancellation }
- activeJobs -- activeInGroup
- idToActiveJob -- jobIds
+ activeJobs --= activeInGroup
+ idToActiveJob --= jobIds
case AllJobsCancelled =>
// Cancel all running jobs.
[09/12] git commit: Tightly couple stageIdToJobIds and jobIdToStageIds
Posted by ma...@apache.org.
Tightly couple stageIdToJobIds and jobIdToStageIds
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/9ae2d094
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/9ae2d094
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/9ae2d094
Branch: refs/heads/master
Commit: 9ae2d094a967782e3f5a624dd854059a40430ee6
Parents: 27c45e5
Author: Mark Hamstra <ma...@gmail.com>
Authored: Fri Nov 22 13:14:26 2013 -0800
Committer: Mark Hamstra <ma...@gmail.com>
Committed: Tue Dec 3 09:57:32 2013 -0800
----------------------------------------------------------------------
.../apache/spark/scheduler/DAGScheduler.scala | 29 ++++++++------------
1 file changed, 12 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/9ae2d094/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index aeac14a..01c5133 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -259,7 +259,7 @@ class DAGScheduler(
val stage =
new Stage(id, rdd, numTasks, shuffleDep, getParentStages(rdd, jobId), jobId, callSite)
stageIdToStage(id) = stage
- registerJobIdWithStages(jobId, stage)
+ updateJobIdStageIdMaps(jobId, stage)
stageToInfos(stage) = new StageInfo(stage)
stage
}
@@ -348,30 +348,24 @@ class DAGScheduler(
* Registers the given jobId among the jobs that need the given stage and
* all of that stage's ancestors.
*/
- private def registerJobIdWithStages(jobId: Int, stage: Stage) {
- def registerJobIdWithStageList(stages: List[Stage]) {
+ private def updateJobIdStageIdMaps(jobId: Int, stage: Stage) {
+ def updateJobIdStageIdMapsList(stages: List[Stage]) {
if (!stages.isEmpty) {
val s = stages.head
stageIdToJobIds.getOrElseUpdate(s.id, new HashSet[Int]()) += jobId
+ jobIdToStageIds.getOrElseUpdate(jobId, new HashSet[Int]()) += s.id
val parents = getParentStages(s.rdd, jobId)
val parentsWithoutThisJobId = parents.filter(p => !stageIdToJobIds.get(p.id).exists(_.contains(jobId)))
- registerJobIdWithStageList(parentsWithoutThisJobId ++ stages.tail)
+ updateJobIdStageIdMapsList(parentsWithoutThisJobId ++ stages.tail)
}
}
- registerJobIdWithStageList(List(stage))
+ updateJobIdStageIdMapsList(List(stage))
}
- private def jobIdToStageIdsAdd(jobId: Int) {
- val stageSet = jobIdToStageIds.getOrElseUpdate(jobId, new HashSet[Int]())
- stageIdToJobIds.foreach { case (stageId, jobSet) =>
- if (jobSet.contains(jobId)) {
- stageSet += stageId
- }
- }
- }
-
- // Removes job and any stages that are not needed by any other job. Returns the set of ids for stages that
- // were removed. The associated tasks for those stages need to be cancelled if we got here via job cancellation.
+ /**
+ * Removes job and any stages that are not needed by any other job. Returns the set of ids for stages that
+ * were removed. The associated tasks for those stages need to be cancelled if we got here via job cancellation.
+ */
private def removeJobAndIndependentStages(jobId: Int): Set[Int] = {
val registeredStages = jobIdToStageIds(jobId)
val independentStages = new HashSet[Int]()
@@ -555,7 +549,6 @@ class DAGScheduler(
idToActiveJob(jobId) = job
activeJobs += job
resultStageToJob(finalStage) = job
- jobIdToStageIdsAdd(jobId)
listenerBus.post(SparkListenerJobStart(job, jobIdToStageIds(jobId).toArray, properties))
submitStage(finalStage)
}
@@ -605,9 +598,11 @@ class DAGScheduler(
handleTaskCompletion(completion)
case LocalJobCompleted(stage) =>
+ val jobId = stageIdToJobIds(stage.id).head
stageIdToJobIds -= stage.id // clean up data structures that were populated for a local job,
stageIdToStage -= stage.id // but that won't get cleaned up via the normal paths through
stageToInfos -= stage // completion events or stage abort
+ jobIdToStageIds -= jobId
case TaskSetFailed(taskSet, reason) =>
stageIdToStage.get(taskSet.stageId).foreach { abortStage(_, reason) }
[10/12] git commit: Local jobs post SparkListenerJobEnd,
and DAGScheduler data structure cleanup always occurs before any
posting of SparkListenerJobEnd.
Posted by ma...@apache.org.
Local jobs post SparkListenerJobEnd, and DAGScheduler data structure
cleanup always occurs before any posting of SparkListenerJobEnd.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/c9fcd909
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/c9fcd909
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/c9fcd909
Branch: refs/heads/master
Commit: c9fcd909d0f86b08935a132409888b30e989bca4
Parents: 9ae2d09
Author: Mark Hamstra <ma...@gmail.com>
Authored: Sun Nov 24 17:49:14 2013 -0800
Committer: Mark Hamstra <ma...@gmail.com>
Committed: Tue Dec 3 09:57:32 2013 -0800
----------------------------------------------------------------------
.../org/apache/spark/scheduler/DAGScheduler.scala | 17 ++++++++++-------
.../apache/spark/scheduler/DAGSchedulerEvent.scala | 2 +-
2 files changed, 11 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c9fcd909/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index 01c5133..b371a24 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -597,12 +597,13 @@ class DAGScheduler(
listenerBus.post(SparkListenerTaskEnd(task, reason, taskInfo, taskMetrics))
handleTaskCompletion(completion)
- case LocalJobCompleted(stage) =>
- val jobId = stageIdToJobIds(stage.id).head
+ case LocalJobCompleted(job, result) =>
+ val stage = job.finalStage
stageIdToJobIds -= stage.id // clean up data structures that were populated for a local job,
stageIdToStage -= stage.id // but that won't get cleaned up via the normal paths through
stageToInfos -= stage // completion events or stage abort
- jobIdToStageIds -= jobId
+ jobIdToStageIds -= job.jobId
+ listenerBus.post(SparkListenerJobEnd(job, result))
case TaskSetFailed(taskSet, reason) =>
stageIdToStage.get(taskSet.stageId).foreach { abortStage(_, reason) }
@@ -672,6 +673,7 @@ class DAGScheduler(
// Broken out for easier testing in DAGSchedulerSuite.
protected def runLocallyWithinThread(job: ActiveJob) {
+ var jobResult: JobResult = JobSucceeded
try {
SparkEnv.set(env)
val rdd = job.finalStage.rdd
@@ -686,9 +688,10 @@ class DAGScheduler(
}
} catch {
case e: Exception =>
+ jobResult = JobFailed(e, Some(job.finalStage))
job.listener.jobFailed(e)
} finally {
- eventProcessActor ! LocalJobCompleted(job.finalStage)
+ eventProcessActor ! LocalJobCompleted(job, jobResult)
}
}
@@ -835,8 +838,8 @@ class DAGScheduler(
activeJobs -= job
resultStageToJob -= stage
markStageAsFinished(stage)
- listenerBus.post(SparkListenerJobEnd(job, JobSucceeded))
jobIdToStageIdsRemove(job.jobId)
+ listenerBus.post(SparkListenerJobEnd(job, JobSucceeded))
}
job.listener.taskSucceeded(rt.outputId, event.result)
}
@@ -987,10 +990,10 @@ class DAGScheduler(
val error = new SparkException("Job %d cancelled".format(jobId))
val job = idToActiveJob(jobId)
job.listener.jobFailed(error)
- listenerBus.post(SparkListenerJobEnd(job, JobFailed(error, Some(job.finalStage))))
jobIdToStageIds -= jobId
activeJobs -= job
idToActiveJob -= jobId
+ listenerBus.post(SparkListenerJobEnd(job, JobFailed(error, Some(job.finalStage))))
}
}
@@ -1009,11 +1012,11 @@ class DAGScheduler(
val job = resultStageToJob(resultStage)
val error = new SparkException("Job aborted: " + reason)
job.listener.jobFailed(error)
- listenerBus.post(SparkListenerJobEnd(job, JobFailed(error, Some(failedStage))))
jobIdToStageIdsRemove(job.jobId)
idToActiveJob -= resultStage.jobId
activeJobs -= job
resultStageToJob -= resultStage
+ listenerBus.post(SparkListenerJobEnd(job, JobFailed(error, Some(failedStage))))
}
if (dependentStages.isEmpty) {
logInfo("Ignoring failure of " + failedStage + " because all jobs depending on it are done")
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c9fcd909/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
index bf8dfb5..aa496b7 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
@@ -65,7 +65,7 @@ private[scheduler] case class CompletionEvent(
taskMetrics: TaskMetrics)
extends DAGSchedulerEvent
-private[scheduler] case class LocalJobCompleted(stage: Stage) extends DAGSchedulerEvent
+private[scheduler] case class LocalJobCompleted(job: ActiveJob, result: JobResult) extends DAGSchedulerEvent
private[scheduler] case class ExecutorGained(execId: String, host: String) extends DAGSchedulerEvent
[11/12] git commit: SparkListenerJobStart posted from local jobs
Posted by ma...@apache.org.
SparkListenerJobStart posted from local jobs
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/403234dd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/403234dd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/403234dd
Branch: refs/heads/master
Commit: 403234dd0d63a7e89f3304d7bb31e3675d405a13
Parents: f55d0b9
Author: Mark Hamstra <ma...@gmail.com>
Authored: Tue Nov 26 22:25:20 2013 -0800
Committer: Mark Hamstra <ma...@gmail.com>
Committed: Tue Dec 3 09:57:32 2013 -0800
----------------------------------------------------------------------
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/403234dd/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index b849867..f9cd021 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -544,6 +544,7 @@ class DAGScheduler(
logInfo("Missing parents: " + getMissingParentStages(finalStage))
if (allowLocal && finalStage.parents.size == 0 && partitions.length == 1) {
// Compute very short actions like first() or take() with no parent stages locally.
+ listenerBus.post(SparkListenerJobStart(job, Array(), properties))
runLocally(job)
} else {
idToActiveJob(jobId) = job
[06/12] git commit: Cleaned up job cancellation handling
Posted by ma...@apache.org.
Cleaned up job cancellation handling
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/27c45e52
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/27c45e52
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/27c45e52
Branch: refs/heads/master
Commit: 27c45e523620d801d547f167a5a33d71ee3af7b5
Parents: 686a420
Author: Mark Hamstra <ma...@gmail.com>
Authored: Fri Nov 22 11:14:39 2013 -0800
Committer: Mark Hamstra <ma...@gmail.com>
Committed: Tue Dec 3 09:57:32 2013 -0800
----------------------------------------------------------------------
.../scala/org/apache/spark/scheduler/DAGScheduler.scala | 12 +++++-------
1 file changed, 5 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/27c45e52/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index b8b3ac0..aeac14a 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -371,7 +371,7 @@ class DAGScheduler(
}
// Removes job and any stages that are not needed by any other job. Returns the set of ids for stages that
- // were removed and whose associated tasks may need to be cancelled.
+ // were removed. The associated tasks for those stages need to be cancelled if we got here via job cancellation.
private def removeJobAndIndependentStages(jobId: Int): Set[Int] = {
val registeredStages = jobIdToStageIds(jobId)
val independentStages = new HashSet[Int]()
@@ -562,8 +562,6 @@ class DAGScheduler(
case JobCancelled(jobId) =>
handleJobCancellation(jobId)
- idToActiveJob.get(jobId).foreach(job => activeJobs -= job)
- idToActiveJob -= jobId
case JobGroupCancelled(groupId) =>
// Cancel all jobs belonging to this job group.
@@ -571,14 +569,12 @@ class DAGScheduler(
val activeInGroup = activeJobs.filter(groupId == _.properties.get(SparkContext.SPARK_JOB_GROUP_ID))
val jobIds = activeInGroup.map(_.jobId)
jobIds.foreach { handleJobCancellation }
- activeJobs --= activeInGroup
- idToActiveJob --= jobIds
case AllJobsCancelled =>
// Cancel all running jobs.
running.map(_.jobId).foreach { handleJobCancellation }
- activeJobs.clear()
- idToActiveJob.clear()
+ activeJobs.clear() // These should already be empty by this point,
+ idToActiveJob.clear() // but just in case we lost track of some jobs...
case ExecutorGained(execId, host) =>
handleExecutorGained(execId, host)
@@ -998,6 +994,8 @@ class DAGScheduler(
job.listener.jobFailed(error)
listenerBus.post(SparkListenerJobEnd(job, JobFailed(error, Some(job.finalStage))))
jobIdToStageIds -= jobId
+ activeJobs -= job
+ idToActiveJob -= jobId
}
}
[08/12] git commit: Synchronous, inline cleanup after runLocally
Posted by ma...@apache.org.
Synchronous, inline cleanup after runLocally
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/f55d0b93
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/f55d0b93
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/f55d0b93
Branch: refs/heads/master
Commit: f55d0b935d7c148f49b15932938e91150b64466f
Parents: c9fcd90
Author: Mark Hamstra <ma...@gmail.com>
Authored: Tue Nov 26 14:06:59 2013 -0800
Committer: Mark Hamstra <ma...@gmail.com>
Committed: Tue Dec 3 09:57:32 2013 -0800
----------------------------------------------------------------------
.../org/apache/spark/scheduler/DAGScheduler.scala | 15 ++++++---------
.../apache/spark/scheduler/DAGSchedulerEvent.scala | 2 --
.../apache/spark/scheduler/DAGSchedulerSuite.scala | 2 --
3 files changed, 6 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/f55d0b93/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index b371a24..b849867 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -597,14 +597,6 @@ class DAGScheduler(
listenerBus.post(SparkListenerTaskEnd(task, reason, taskInfo, taskMetrics))
handleTaskCompletion(completion)
- case LocalJobCompleted(job, result) =>
- val stage = job.finalStage
- stageIdToJobIds -= stage.id // clean up data structures that were populated for a local job,
- stageIdToStage -= stage.id // but that won't get cleaned up via the normal paths through
- stageToInfos -= stage // completion events or stage abort
- jobIdToStageIds -= job.jobId
- listenerBus.post(SparkListenerJobEnd(job, result))
-
case TaskSetFailed(taskSet, reason) =>
stageIdToStage.get(taskSet.stageId).foreach { abortStage(_, reason) }
@@ -691,7 +683,12 @@ class DAGScheduler(
jobResult = JobFailed(e, Some(job.finalStage))
job.listener.jobFailed(e)
} finally {
- eventProcessActor ! LocalJobCompleted(job, jobResult)
+ val s = job.finalStage
+ stageIdToJobIds -= s.id // clean up data structures that were populated for a local job,
+ stageIdToStage -= s.id // but that won't get cleaned up via the normal paths through
+ stageToInfos -= s // completion events or stage abort
+ jobIdToStageIds -= job.jobId
+ listenerBus.post(SparkListenerJobEnd(job, jobResult))
}
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/f55d0b93/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
index aa496b7..add1187 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerEvent.scala
@@ -65,8 +65,6 @@ private[scheduler] case class CompletionEvent(
taskMetrics: TaskMetrics)
extends DAGSchedulerEvent
-private[scheduler] case class LocalJobCompleted(job: ActiveJob, result: JobResult) extends DAGSchedulerEvent
-
private[scheduler] case class ExecutorGained(execId: String, host: String) extends DAGSchedulerEvent
private[scheduler] case class ExecutorLost(execId: String) extends DAGSchedulerEvent
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/f55d0b93/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
index 8ce8c68..706d84a 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala
@@ -219,8 +219,6 @@ class DAGSchedulerSuite extends FunSuite with BeforeAndAfter with LocalSparkCont
}
val jobId = scheduler.nextJobId.getAndIncrement()
runEvent(JobSubmitted(jobId, rdd, jobComputeFunc, Array(0), true, null, listener))
- assert(scheduler.stageToInfos.size === 1)
- runEvent(LocalJobCompleted(scheduler.stageToInfos.keys.head))
assert(results === Map(0 -> 42))
assertDataStructuresEmpty
}
[07/12] git commit: Improved comment
Posted by ma...@apache.org.
Improved comment
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/205566e5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/205566e5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/205566e5
Branch: refs/heads/master
Commit: 205566e56e2891245b2d7820bfb3629945a2dcd9
Parents: 94087c4
Author: Mark Hamstra <ma...@gmail.com>
Authored: Wed Nov 20 14:49:09 2013 -0800
Committer: Mark Hamstra <ma...@gmail.com>
Committed: Tue Dec 3 09:57:32 2013 -0800
----------------------------------------------------------------------
.../main/scala/org/apache/spark/scheduler/DAGScheduler.scala | 7 +++----
1 file changed, 3 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/205566e5/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index be46f74..6f9d4d5 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -243,10 +243,9 @@ class DAGScheduler(
}
/**
- * Create a Stage for the given RDD, either as a shuffle map stage (for a ShuffleDependency) or
- * as a result stage for the final RDD used directly in an action. The stage will also be
- * associated with the provided jobId.. Shuffle map stages, whose shuffleId may have previously
- * been registered in the MapOutputTracker, should be (re)-created using newOrUsedStage.
+ * Create a Stage -- either directly for use as a result stage, or as part of the (re)-creation
+ * of a shuffle map stage in newOrUsedStage. The stage will be associated with the provided
+ * jobId. Production of shuffle map stages should always use newOrUsedStage, not newStage directly.
*/
private def newStage(
rdd: RDD[_],