-
Notifications
You must be signed in to change notification settings - Fork 28.1k
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-19753][CORE] Un-register all shuffle output on a host in case of slave lost or fetch failure #17088
[SPARK-19753][CORE] Un-register all shuffle output on a host in case of slave lost or fetch failure #17088
Changes from all commits
74ca88b
6898c2b
32a2315
c7c3129
f96ec68
d4979e3
8787db1
4ca9527
9f64e29
be3b3db
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 |
---|---|---|
|
@@ -394,6 +394,68 @@ class DAGSchedulerSuite extends SparkFunSuite with LocalSparkContext with Timeou | |
assertDataStructuresEmpty() | ||
} | ||
|
||
test("All shuffle files should on the slave should be cleaned up when slave lost") { | ||
// reset the test context with the right shuffle service config | ||
afterEach() | ||
val conf = new SparkConf() | ||
conf.set("spark.shuffle.service.enabled", "true") | ||
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 we add a another test with |
||
init(conf) | ||
runEvent(ExecutorAdded("exec-hostA1", "hostA")) | ||
runEvent(ExecutorAdded("exec-hostA2", "hostA")) | ||
runEvent(ExecutorAdded("exec-hostB", "hostB")) | ||
val firstRDD = new MyRDD(sc, 3, Nil) | ||
val firstShuffleDep = new ShuffleDependency(firstRDD, new HashPartitioner(3)) | ||
val firstShuffleId = firstShuffleDep.shuffleId | ||
val shuffleMapRdd = new MyRDD(sc, 3, List(firstShuffleDep)) | ||
val shuffleDep = new ShuffleDependency(shuffleMapRdd, new HashPartitioner(3)) | ||
val reduceRdd = new MyRDD(sc, 1, List(shuffleDep)) | ||
submit(reduceRdd, Array(0)) | ||
// map stage1 completes successfully, with one task on each executor | ||
complete(taskSets(0), Seq( | ||
(Success, | ||
MapStatus(BlockManagerId("exec-hostA1", "hostA", 12345), Array.fill[Long](1)(2))), | ||
(Success, | ||
MapStatus(BlockManagerId("exec-hostA2", "hostA", 12345), Array.fill[Long](1)(2))), | ||
(Success, makeMapStatus("hostB", 1)) | ||
)) | ||
// map stage2 completes successfully, with one task on each executor | ||
complete(taskSets(1), Seq( | ||
(Success, | ||
MapStatus(BlockManagerId("exec-hostA1", "hostA", 12345), Array.fill[Long](1)(2))), | ||
(Success, | ||
MapStatus(BlockManagerId("exec-hostA2", "hostA", 12345), Array.fill[Long](1)(2))), | ||
(Success, makeMapStatus("hostB", 1)) | ||
)) | ||
// make sure our test setup is correct | ||
val initialMapStatus1 = mapOutputTracker.mapStatuses.get(0).get | ||
assert(initialMapStatus1.count(_ != null) === 3) | ||
assert(initialMapStatus1.map{_.location.executorId}.toSet === | ||
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.
|
||
Set("exec-hostA1", "exec-hostA2", "exec-hostB")) | ||
|
||
val initialMapStatus2 = mapOutputTracker.mapStatuses.get(0).get | ||
assert(initialMapStatus2.count(_ != null) === 3) | ||
assert(initialMapStatus2.map{_.location.executorId}.toSet === | ||
Set("exec-hostA1", "exec-hostA2", "exec-hostB")) | ||
|
||
// reduce stage fails with a fetch failure from one host | ||
complete(taskSets(2), Seq( | ||
(FetchFailed(BlockManagerId("exec-hostA2", "hostA", 12345), firstShuffleId, 0, 0, "ignored"), | ||
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. Seems the |
||
null) | ||
)) | ||
|
||
// Here is the main assertion -- make sure that we de-register | ||
// the map outputs for both map stage from both executors on hostA | ||
val mapStatus1 = mapOutputTracker.mapStatuses.get(0).get | ||
assert(mapStatus1.count(_ != null) === 1) | ||
assert(mapStatus1(2).location.executorId === "exec-hostB") | ||
assert(mapStatus1(2).location.host === "hostB") | ||
|
||
val mapStatus2 = mapOutputTracker.mapStatuses.get(1).get | ||
assert(mapStatus2.count(_ != null) === 1) | ||
assert(mapStatus2(2).location.executorId === "exec-hostB") | ||
assert(mapStatus2(2).location.host === "hostB") | ||
} | ||
|
||
test("zero split job") { | ||
var numResults = 0 | ||
var failureReason: Option[Exception] = None | ||
|
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.
nit: prefer string interpolation over
format
.