Skip to content

SPARK-1235: manage the DAGScheduler EventProcessActor with supervisor and refactor the DAGScheduler with Akka #186

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

SPARK-1235: manage the DAGScheduler EventProcessActor with supervisor and refactor the DAGScheduler with Akka #186

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
810efd8
akka solution
mengxr Apr 9, 2014
b68c1c7
refactor DAGScheduler with Akka
CodingCat Apr 4, 2014
228f4b0
address comments from Mark
CodingCat Apr 10, 2014
089bc2f
address andrew's comments
CodingCat Apr 18, 2014
fdf3b17
just to retrigger the test......
CodingCat Apr 18, 2014
a7a2a97
add StageCancelled message
CodingCat Apr 22, 2014
9dfb033
remove unnecessary changes
CodingCat Apr 22, 2014
5d1636a
re-trigger the test.....
CodingCat Apr 22, 2014
ac878ab
typo fix
CodingCat Apr 22, 2014
a9eea03
address Matei's comments
CodingCat Apr 23, 2014
c048d0e
call submitWaitingStages for every event
CodingCat Apr 23, 2014
561cfbc
recover doCheckpoint
CodingCat Apr 23, 2014
cd02d9a
small fix
CodingCat Apr 23, 2014
310a579
style fix
CodingCat Apr 24, 2014
82d08b3
calling actorOf on system to ensure it is blocking
CodingCat Apr 24, 2014
35c886a
fix bug
CodingCat Apr 24, 2014
baf2d38
fix the issue brought by non-blocking actorOf
CodingCat Apr 24, 2014
124d82d
blocking the constructor until event actor is ready
CodingCat Apr 24, 2014
a7fb0ee
throw Exception on failure of creating DAG
CodingCat Apr 25, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 12 additions & 8 deletions core/src/main/scala/org/apache/spark/SparkContext.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -274,10 +274,17 @@ class SparkContext(config: SparkConf) extends Logging {

// Create and start the scheduler
private[spark] var taskScheduler = SparkContext.createTaskScheduler(this, master)
taskScheduler.start()
@volatile private[spark] var dagScheduler: DAGScheduler = _
try {
dagScheduler = new DAGScheduler(this)
} catch {
case e: Exception => throw
new SparkException("DAGScheduler cannot be initialized due to %s".format(e.getMessage))
}

@volatile private[spark] var dagScheduler = new DAGScheduler(this)
dagScheduler.start()
// start TaskScheduler after taskScheduler sets DAGScheduler reference in DAGScheduler's
// constructor
Copy link
Contributor

@andrewor14 andrewor14 Apr 16, 2014

Choose a reason for hiding this comment

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

nit: could you put these 2 in 1 line?

Copy link
Contributor Author

@CodingCat CodingCat Apr 18, 2014

Choose a reason for hiding this comment

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

I found that the line length would be 103 if I put them together....

taskScheduler.start()

private[spark] val cleaner: Option[ContextCleaner] = {
if (conf.getBoolean("spark.cleaner.referenceTracking", true)) {
Expand Down Expand Up @@ -1007,8 +1014,8 @@ class SparkContext(config: SparkConf) extends Logging {
partitions: Seq[Int],
allowLocal: Boolean,
resultHandler: (Int, U) => Unit) {
partitions.foreach{ p =>
require(p >= 0 && p < rdd.partitions.size, s"Invalid partition requested: $p")
if (dagScheduler == null) {
throw new SparkException("SparkContext has been shutdown")
}
val callSite = getCallSite
val cleanedFunc = clean(func)
Expand Down Expand Up @@ -1116,9 +1123,6 @@ class SparkContext(config: SparkConf) extends Logging {
resultHandler: (Int, U) => Unit,
resultFunc: => R): SimpleFutureAction[R] =
{
partitions.foreach{ p =>
require(p >= 0 && p < rdd.partitions.size, s"Invalid partition requested: $p")
}
val cleanF = clean(processPartition)
val callSite = getCallSite
val waiter = dagScheduler.submitJob(
Expand Down
6 changes: 3 additions & 3 deletions core/src/main/scala/org/apache/spark/rdd/RDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1112,9 +1112,9 @@ abstract class RDD[T: ClassTag](
@transient private var doCheckpointCalled = false

/**
* Performs the checkpointing of this RDD by saving this. It is called by the DAGScheduler
* after a job using this RDD has completed (therefore the RDD has been materialized and
* potentially stored in memory). doCheckpoint() is called recursively on the parent RDDs.
* Performs the checkpointing of this RDD by saving this. It is called after a job using this RDD
* has completed (therefore the RDD has been materialized and potentially stored in memory).
* doCheckpoint() is called recursively on the parent RDDs.
*/
private[spark] def doCheckpoint() {
if (!doCheckpointCalled) {
Expand Down
Loading