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.
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
[SPARK-10620] [SPARK-13054] Minor addendum to #10835 #10958
Changes from 6 commits
8be6c86
9de795b
5404254
822899d
ba5f555
6e4859d
3b1e414
854c71f
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -118,7 +118,7 @@ case class ExceptionFailure( | |
description: String, | ||
stackTrace: Array[StackTraceElement], | ||
fullStackTrace: String, | ||
exceptionWrapper: Option[ThrowableSerializationWrapper], | ||
private val exceptionWrapper: Option[ThrowableSerializationWrapper], | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Looks like this was private in 1.6, so this looks good. |
||
accumUpdates: Seq[AccumulableInfo] = Seq.empty[AccumulableInfo]) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Should this be private? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I left it public because |
||
extends TaskFailedReason { | ||
|
||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -1144,13 +1144,12 @@ class DAGScheduler( | |
null | ||
} | ||
|
||
// The success case is dealt with separately below. | ||
// TODO: Why post it only for failed tasks in cancelled stages? Clarify semantics here. | ||
if (event.reason != Success) { | ||
val attemptId = task.stageAttemptId | ||
listenerBus.post(SparkListenerTaskEnd( | ||
stageId, attemptId, taskType, event.reason, event.taskInfo, taskMetrics)) | ||
} | ||
// Note: this stage may already have been canceled, in which case this task end event | ||
// maybe posted after the stage completed event. There's not much we can do here without | ||
// introducing additional complexity in the scheduler to wait for all the task end events | ||
// before posting the stage completed event. | ||
listenerBus.post(SparkListenerTaskEnd( | ||
stageId, task.stageAttemptId, taskType, event.reason, event.taskInfo, taskMetrics)) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. (comment for linking) There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This should probably have a test like #10951. We should combine our efforts on this. This might make more sense to have in a separate PR from the other style things but doesn't matter to much. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. ok, I separated it There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. are you going to file a separate pr or should I just update mine ? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. you can just update yours |
||
|
||
if (!stageIdToStage.contains(task.stageId)) { | ||
// Skip all the actions if the stage has been cancelled. | ||
|
@@ -1160,8 +1159,6 @@ class DAGScheduler( | |
val stage = stageIdToStage(task.stageId) | ||
event.reason match { | ||
case Success => | ||
listenerBus.post(SparkListenerTaskEnd(stageId, stage.latestInfo.attemptId, taskType, | ||
event.reason, event.taskInfo, taskMetrics)) | ||
stage.pendingPartitions -= task.partitionId | ||
task match { | ||
case rt: ResultTask[_, _] => | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -86,7 +86,7 @@ class InternalAccumulatorSuite extends SparkFunSuite with LocalSparkContext { | |
} | ||
|
||
test("create") { | ||
val accums = create() | ||
val accums = createAll() | ||
val shuffleReadAccums = createShuffleReadAccums() | ||
val shuffleWriteAccums = createShuffleWriteAccums() | ||
val inputAccums = createInputAccums() | ||
|
@@ -122,7 +122,7 @@ class InternalAccumulatorSuite extends SparkFunSuite with LocalSparkContext { | |
} | ||
|
||
test("naming") { | ||
val accums = create() | ||
val accums = createAll() | ||
val shuffleReadAccums = createShuffleReadAccums() | ||
val shuffleWriteAccums = createShuffleWriteAccums() | ||
val inputAccums = createInputAccums() | ||
|
@@ -220,7 +220,7 @@ class InternalAccumulatorSuite extends SparkFunSuite with LocalSparkContext { | |
rdd.count() | ||
} | ||
|
||
// TODO: these two tests are incorrect; they don't actually trigger stage retries. | ||
// TODO: these two tests are incorrect; they don't actually trigger stage retries (SPARK-13053). | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This was fixed in https://issues.apache.org/jira/browse/SPARK-13053, right? |
||
ignore("internal accumulators in fully resubmitted stages") { | ||
testInternalAccumulatorsWithFailedTasks((i: Int) => true) // fail all tasks | ||
} | ||
|
@@ -236,7 +236,7 @@ class InternalAccumulatorSuite extends SparkFunSuite with LocalSparkContext { | |
} | ||
assert(Accumulators.originals.isEmpty) | ||
sc.parallelize(1 to 100).map { i => (i, i) }.reduceByKey { _ + _ }.count() | ||
val internalAccums = InternalAccumulator.create() | ||
val internalAccums = InternalAccumulator.createAll() | ||
// We ran 2 stages, so we should have 2 sets of internal accumulators, 1 for each stage | ||
assert(Accumulators.originals.size === internalAccums.size * 2) | ||
val accumsRegistered = sc.cleaner match { | ||
|
@@ -259,7 +259,7 @@ class InternalAccumulatorSuite extends SparkFunSuite with LocalSparkContext { | |
|
||
/** | ||
* Test whether internal accumulators are merged properly if some tasks fail. | ||
* TODO: make this actually retry the stage. | ||
* TODO: make this actually retry the stage (SPARK-13053). | ||
*/ | ||
private def testInternalAccumulatorsWithFailedTasks(failCondition: (Int => Boolean)): Unit = { | ||
val listener = new SaveInfoListener | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This change was already added in another patch, causing a merge conflict here.