[SPARK-5259][CORE]Make sure shuffle metadata already in mapOutPutTracker while submitting tasks of the shuffleMapStage #4055

suyanNone · 2015-01-15T04:07:57Z

[SPARK-5259]Add task equal() and hashcode() to avoid stage.pendingTasks not accurate while stage was retry
desc:
while run a spark job, it occurs one stage keep retrying and keep throwing FetchMetadataException

reason:
Map Stage 1-> Map Stage2

MapStage1 retry, so have 2 taskSet, taskSet0.0 and TaskSet0.1
TaskSet0.0 and TaskSet0.1 are all running.

When to submit Map Stage2？

       if (!mapStage.isAvailable) {
                  missing += mapStage
                }

  def isAvailable: Boolean = {
    if (!isShuffleMap) {
      true
    } else {
      numAvailableOutputs == numPartitions
    }
  }

how numAvailableOutputs change?

stage.addOutputLoc(smt.partitionId, status)

  def addOutputLoc(partition: Int, status: MapStatus) {
    val prevList = outputLocs(partition)
    outputLocs(partition) = status :: prevList
    if (prevList == Nil) {
      numAvailableOutputs += 1
    }
  }

When to register Map Stage1 out put?

if (runningStages.contains(stage) && stage.pendingTasks.isEmpty) {
}

how stage.pendingTasks change?

event.reason match {
      case Success =>
        listenerBus.post(SparkListenerTaskEnd(stageId, stage.latestInfo.attemptId, taskType,
          event.reason, event.taskInfo, event.taskMetrics))
        stage.pendingTasks -= task

because, because Task not override hashcode and equal, so run same partition task in different TaskSet is different task. and pendingTask is clear when retry map stage, so the pendingTask is always for the new retry TaskSet. then the previous taskset complete some task which have same partition in the latest taskSet. stage.pengdingTask -= task not affect anything. but it affect stage.numAvailableOutputs, because it just identified by partition Id.
So it may result in some stage have submit while his dependency map stage have not registered its output in MapOutputTracker.

SparkQA · 2015-01-15T05:19:32Z

Test build #25592 has finished for PR 4055 at commit e5af95c.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class ExperimentalMethods protected[sql](sqlContext: SQLContext)

srowen · 2015-01-22T10:53:48Z

core/src/main/scala/org/apache/spark/scheduler/Task.scala

This seems like an excessively complex way of writing 31 * stageId.hashCode + partitionId.hashCode. I don't think FP is the way to do this.

Maybe a better way is (stageId + partitionId) * (stageId + partitionId + 1) / 2 + partitionId.
See http://en.wikipedia.org/wiki/Pairing_function#Cantor_pairing_function

cloud-fan · 2015-01-23T08:36:45Z

According to your case, I think we can do one more improvement in submitMissingTasks. If the stage is map stage and stage.pendingTasks is not empty, we should not regenerate all tasks but just submit the pending tasks.

SparkQA · 2015-01-26T08:40:07Z

Test build #26086 has finished for PR 4055 at commit a181772.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

lianhuiwang · 2015-01-26T09:36:40Z

@JoshRosen i think that's ok. because change of code is very small and there is no influence for current logic.

suyanNone · 2015-01-26T11:30:52Z

@srowen, the original hashCode() generated by idea auto-generated hashcode feature. Now I already refined by your comments.
About canEqual(), I just according programming in Scala describe:

to Resolve 
warning: non variable type-argument T in type
pattern is unchecked since it is eliminated by erasure
case that: Branch[T] => this.elem == that.elem &&

class Branch[T](
  val elem: T,
  val left: Tree[T],
  val right: Tree[T]
) extends Tree[T] {
  override def equals(other: Any) = other match {
    case that: Branch[_] => (that canEqual this) &&
                                          this.elem == that.elem &&
                                          this.left == that.left &&
                                          this.right == that.right
    case _ => false
}
   def canEqual(other: Any) = other.isInstanceOf[Branch[_]]
   override def hashCode: Int = 41 * (41 * (41 + elem.hashCode) + left.hashCode) + right.hashCode
}
  Listing 30.4 · A parameterized type with equals and

@cloud-fan , I think 31*hashcode + hashcode is more common method to do that.

suyanNone · 2015-01-26T13:06:50Z

@cloud-fan
According current code, it may not easy to change to not re-submit task in pendingTasks. To be honest, current DAGScheduler is complicated but at some point, is simple for some situation, still a lot may need to be improved.

re-submit occurs have failed-stage and due to a fetch failed. a fetch-failed means current running TaskSet is dead(called zombie), so it just have the already scheduled on Executor task are running, others in stage.pendingTask will never be scheduled in previous taskset. but it can do to just resubmit not scheduled tasks.

suyanNone · 2015-01-26T13:13:27Z

@cloud-fan btw, Do you know HarryZhang? ZJU VLIS Lab

cloud-fan · 2015-01-26T13:30:13Z

@suyanNone Thanks for the explanation of re-submit!
What's the Chinese name of HarryZhang? We don't use English name in the lab……

suyanNone · 2015-01-27T02:22:44Z

@cloud-fan ZhangLei, SunHongLiang, HanLi, ChenXingYu, blabla...I am ZhangLei's classmate in ZJU.

cloud-fan · 2015-01-27T06:19:20Z

core/src/main/scala/org/apache/spark/scheduler/ResultTask.scala

You can try Seq[Int](1).isInstanceOf[Seq[String]] in REPL, it will return true.
isInstanceOf can't work on generic type because of JVM type erasure.

@cloud-fan yean, I know that. and in that class, it has no need to add parameter on class level, it could only use in function level run or runContext.

and, this code still have something to refine, like var partitionId to be val, I will refine it at later

I mean... something like other.isInstanceOf[ResultTask[_, _]] =.=

Yes, that's very slightly better. I agree

So equals is not overridden in these subclasses because equality does not depend on their additional fields? just checking that this is definitely desirable.

eh...StageId and partitionId is like a unique composite primary key in database. In current spark context, it's sure can be identified by (StageId, PartitionId), even not need to use "canEqual".

SparkQA · 2015-01-29T09:24:53Z

Test build #26307 has finished for PR 4055 at commit ce54738.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

suyanNone · 2015-01-30T03:23:12Z

@cloud-fan --!
[error] /home/jenkins/workspace/SparkPullRequestBuilder/core/src/main/scala/org/apache/spark/scheduler/ResultTask.scala:69: class ResultTask takes type parameters
[error] override def canEqual(other: Any): Boolean = other.isInstanceOf[ResultTask]

suyanNone · 2015-01-30T03:24:56Z

retest this please

SparkQA · 2015-01-30T03:55:13Z

Test build #26373 has finished for PR 4055 at commit adca2aa.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-01-30T04:01:25Z

Test build #26374 has finished for PR 4055 at commit adca2aa.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

suyanNone · 2015-01-30T07:17:02Z

retest this please

SparkQA · 2015-01-30T08:12:15Z

Test build #26386 has finished for PR 4055 at commit 076f54d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

suyanNone · 2015-02-17T10:26:40Z

@srowen @JoshRosen can some one verify this patch.

srowen · 2015-02-17T10:32:44Z

core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala

Why not string interpolation here?

@srowen task.partitionID is Int type.

No, I mean why how use the same s"..." syntax as in the line above? Int is fine.

@srowen Others do that...I can't figure out the advantages and disadvantages.

There a lot sentence like:
logInfo("Finished task %s in stage %s (TID %d) in %d ms on %s (%d/%d)".format(
logError("Task %s in stage %s (TID %d) had a not serializable result: %s; not retrying"
.format(i
abort("Task %s in stage %s (TID %d) had a not serializable result: %s".format(

and also has this:
logInfo(
s"Lost task ${info.id} in stage ${taskSet.id} (TID $tid) on executor ${info.host}: " +
s"${ef.className} (${ef.description}) [duplicate $dupCount]")

Need I to refactor it?

The s"..." didn't exist before Scala 2.10, so I think that's why the old style is still used in the code. There's no great need to change all that. I think the interpolated style is clearer, and I tend to think that we should match surrounding code style in issues like this. Since interpolation is used in the line above, it seems right to use it here. I agree it's a tiny issue either way.

SparkQA · 2015-02-27T03:46:41Z

Test build #28038 has finished for PR 4055 at commit 29684d8.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-02-27T10:48:41Z

Test build #28061 has finished for PR 4055 at commit 9025cf1.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

srowen · 2015-02-27T11:12:13Z

@cloud-fan @rxin do you have any final thoughts on this? it's looking reasonable to me though I admit I don't know this scheduler code well enough to be confident.

rxin · 2015-02-27T20:05:30Z

cc @markhamstra and @kayousterhout also

markhamstra · 2015-02-27T20:17:33Z

I'll take a look over the weekend.

cloud-fan · 2015-03-03T01:40:37Z

I'm still a little against the canEqual method. In this particular context, I think "(stageId, partitionId)" is meaningful enough to identify a task.

…hile stageId and partId are same

SparkQA · 2015-07-21T04:14:21Z

Test build #37907 has finished for PR 4055 at commit 7ae128e.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- trait ExpectsInputTypes extends Expression
- trait ImplicitCastInputTypes extends ExpectsInputTypes
- trait Unevaluable extends Expression
- trait Nondeterministic extends Expression
- trait CodegenFallback extends Expression
- case class Hex(child: Expression) extends UnaryExpression with ImplicitCastInputTypes
- case class Unhex(child: Expression) extends UnaryExpression with ImplicitCastInputTypes
- abstract class RDG extends LeafExpression with Nondeterministic
- case class Rand(seed: Long) extends RDG
- case class Randn(seed: Long) extends RDG
- case class Ascii(child: Expression) extends UnaryExpression with ImplicitCastInputTypes
- case class Base64(child: Expression) extends UnaryExpression with ImplicitCastInputTypes
- case class UnBase64(child: Expression) extends UnaryExpression with ImplicitCastInputTypes
- case class FakeFileStatus(

SparkQA · 2015-07-21T13:05:18Z

Test build #37949 has finished for PR 4055 at commit b2df3fd.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

squito · 2015-07-21T16:11:42Z

core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala

getServerStatuses has been removed in master -- I guess both of these should be

val statuses = mapOutputTracker.getMapSizesByExecutorId(0, reduceIdx) assert(statuses != null) assert(statuses.nonEmpty)

The new code will now throw an exception if we're missing the map output data, but I feel like its probably still good to leave those asserts in.

may the below code will be more better?

try { mapOutputTracker.getMapSizesByExecutorId(0, reduceIdx) } catch { case e: Exception => fail("") }

we don't use try / case e: Exception => fail("") to fail tests when there is an exception -- we just let the exception fail the test directly. You get more info in the stack trace that way. So I think its better to just leave it bare.

You could just put in a comment explaining what the point is:

// this would throw an exception if the map status hadn't been registered mapOutputTracker.getMapSizesByExecutorId(0, reduceIdx)

I still slightly prefer leaving the asserts in there. Yes, they are kinda pointless with the current behavior of getMapSizesByExecutorId -- but I'd just like to be a bit more defensive, in case that behavior changes in the future. (eg., maybe some future refactoring makes them stop throwing exceptions for some reason).

Maybe to be very clear, you could include the asserts and more comments:

// this would throw an exception if the map status hadn't been registered val statuses = mapOutputTracker.getMapSizesByExecutorId(0, reduceIdx) // really we should have already thrown an exception rather than fail either of these // asserts, but just to be extra defensive let's double check the statuses are OK assert(statuses != null) assert(statuses.nonEmpty)

This is pretty minor, though, I don't feel strongly about it.

squito · 2015-07-21T16:25:01Z

thanks for updating @suyanNone ! there are compile errors b/c of changes in master, and I left some really minor comments, but I think its basically ready.

btw, feel free to open separate jiras / prs for the other issues you found (and cc me if you like). I do think they are worth discussing, but this the most important fix.

andrewor14 · 2015-09-01T23:42:22Z

@squito @suyanNone is this superseded by #7699? If so, would you mind closing this patch?

rxin · 2015-09-21T19:31:30Z

@suyanNone can you add your git commit email to your github profile, so this commit will show up as yours?

srowen reviewed Jan 22, 2015
View reviewed changes

cloud-fan reviewed Jan 27, 2015
View reviewed changes

srowen reviewed Feb 17, 2015
View reviewed changes

suyanNone force-pushed the task-equal branch from 29684d8 to 9025cf1 Compare February 27, 2015 09:23

suyanNone added 15 commits July 21, 2015 11:51

Add sub-class canEqual to make ShuffleMapTask not equal Result Task w…

741ab4d

…hile stageId and partId are same

Refine

5ec8a82

add parameter type

9044720

Refine

03db624

Refine log message use string interpolation

8dfdc18

Refine with the latest spark

c286f7a

Refine solution and fix bug in code

dcf1533

Refine codeStyle

27da8e7

Refine Tests

bd5fec4

Refine assert exception

4dbe4d3

Refine the test 'ignore late map task completions'

9dfff63

Refine the testcase

e1e0b66

Refine scala style

2379250

Only tracker partitionId to instead of Task

314873a

Refine test name

7ae128e

suyanNone force-pushed the task-equal branch from 340a52c to 7ae128e Compare July 21, 2015 03:54

refine suite code

b2df3fd

squito reviewed Jul 21, 2015
View reviewed changes

squito mentioned this pull request Jul 27, 2015

[SPARK-5259][CORE] don't submit stage until its dependencies map outputs are registered #7699

Closed

asfgit closed this in 804a012 Sep 4, 2015

[SPARK-5259][CORE]Make sure shuffle metadata already in mapOutPutTracker while submitting tasks of the shuffleMapStage #4055

[SPARK-5259][CORE]Make sure shuffle metadata already in mapOutPutTracker while submitting tasks of the shuffleMapStage #4055

Uh oh!

Conversation

suyanNone commented Jan 15, 2015

Uh oh!

SparkQA commented Jan 15, 2015

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Jan 23, 2015

Uh oh!

SparkQA commented Jan 26, 2015

Uh oh!

lianhuiwang commented Jan 26, 2015

Uh oh!

suyanNone commented Jan 26, 2015

Uh oh!

suyanNone commented Jan 26, 2015

Uh oh!

suyanNone commented Jan 26, 2015

Uh oh!

cloud-fan commented Jan 26, 2015

Uh oh!

suyanNone commented Jan 27, 2015

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jan 29, 2015

Uh oh!

suyanNone commented Jan 30, 2015

Uh oh!

suyanNone commented Jan 30, 2015

Uh oh!

SparkQA commented Jan 30, 2015

Uh oh!

SparkQA commented Jan 30, 2015

Uh oh!

suyanNone commented Jan 30, 2015

Uh oh!

SparkQA commented Jan 30, 2015

Uh oh!

suyanNone commented Feb 17, 2015

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Feb 27, 2015

Uh oh!

SparkQA commented Feb 27, 2015

Uh oh!

srowen commented Feb 27, 2015

Uh oh!

rxin commented Feb 27, 2015

Uh oh!

markhamstra commented Feb 27, 2015

Uh oh!

cloud-fan commented Mar 3, 2015

Uh oh!

SparkQA commented Jul 21, 2015

Uh oh!