Skip to content

[SPARK-25299] ShuffleLocation/FetchFailed integrations with scheduler #548

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
yifeih wants to merge 27 commits into from
Closed

[SPARK-25299] ShuffleLocation/FetchFailed integrations with scheduler #548

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
36d4898
semantic fix that hasn't merged yet (to revert)
yifeih May 9, 2019
48f52c8
specify ShuffleLocation interface
yifeih May 9, 2019
d161feb
more api changes
yifeih May 9, 2019
070151b
wip
yifeih May 10, 2019
190e2de
fix
yifeih May 10, 2019
a34a103
style
yifeih May 11, 2019
93c2449
fix style, fix test
yifeih May 13, 2019
fcb6ac7
cleanup
yifeih May 13, 2019
715a7ed
use seq in map output tracker
yifeih May 13, 2019
b68254c
always remove for remote
yifeih May 13, 2019
df8d457
fix some tests
yifeih May 14, 2019
97f6ff3
json serialize
yifeih May 14, 2019
09eeb13
array -> seq
yifeih May 14, 2019
ba78f9a
fix serialization
yifeih May 14, 2019
f50a309
fix javadoc
yifeih May 14, 2019
b7c1e2a
write async scheduler tests
yifeih May 15, 2019
de824d4
fix the async location tests
yifeih May 16, 2019
7794b68
add tests for distributed file server case
yifeih May 16, 2019
9604889
use scala in tests
yifeih May 16, 2019
58c8ef2
more buffers
yifeih May 17, 2019
506a3f7
use Seq.empty instead of null
yifeih May 17, 2019
118a2bf
add flag for removing other map statuses
yifeih May 17, 2019
bf8aeab
small fixes
yifeih May 17, 2019
994ff5a
use seq empty
yifeih May 20, 2019
7758a90
simplify async case
yifeih May 20, 2019
cebe386
remove another null to Seq.empty
yifeih May 20, 2019
056c87b
one more
yifeih May 20, 2019
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
Prev Previous commit
Next Next commit
fix style, fix test
  • Loading branch information
@yifeih
yifeih committed May 13, 2019
commit 93c244990d066696c1b42cdf5a2eea8a5856c243
2 changes: 0 additions & 2 deletions core/src/main/java/org/apache/spark/api/shuffle/ShuffleBlockInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,6 @@

package org.apache.spark.api.shuffle;

import org.apache.spark.api.java.Optional;

import java.util.Objects;

/**
Expand Down
16 changes: 8 additions & 8 deletions core/src/test/scala/org/apache/spark/MapOutputTrackerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,13 +69,13 @@ class MapOutputTrackerSuite extends SparkFunSuite {
tracker.registerMapOutput(10, 1, MapStatus(BlockManagerId("b", "hostB", 1000),
Array(10000L, 1000L)))
val statuses = tracker.getMapSizesByShuffleLocation(10, 0)
assert(statuses.toSet ===
Seq(
(Some(DefaultMapShuffleLocations.get(BlockManagerId("a", "hostA", 1000))),
ArrayBuffer((ShuffleBlockId(10, 0, 0), size1000))),
(Some(DefaultMapShuffleLocations.get(BlockManagerId("b", "hostB", 1000))),
ArrayBuffer((ShuffleBlockId(10, 1, 0), size10000))))
.toSet)
val output = Seq(
(DefaultMapShuffleLocations.get(BlockManagerId("a", "hostA", 1000)).getLocationsForBlock(0),
ArrayBuffer((ShuffleBlockId(10, 0, 0), size1000))),
(DefaultMapShuffleLocations.get(BlockManagerId("b", "hostB", 1000)).getLocationsForBlock(0),
ArrayBuffer((ShuffleBlockId(10, 1, 0), size10000))))
.toSet
assert(statuses.toSet === output)
assert(0 == tracker.getNumCachedSerializedBroadcast)
tracker.stop()
rpcEnv.shutdown()
Expand Down Expand Up @@ -157,7 +157,7 @@ class MapOutputTrackerSuite extends SparkFunSuite {
slaveTracker.updateEpoch(masterTracker.getEpoch)
assert(slaveTracker.getMapSizesByShuffleLocation(10, 0).toSeq ===
Seq(
(Some(DefaultMapShuffleLocations.get(BlockManagerId("a", "hostA", 1000))),
(DefaultMapShuffleLocations.get(BlockManagerId("a", "hostA", 1000)).getLocationsForBlock(0),
ArrayBuffer((ShuffleBlockId(10, 0, 0), size1000)))))
assert(0 == masterTracker.getNumCachedSerializedBroadcast)

Expand Down