Skip to content

[SPARK-28560][SQL] Optimize shuffle reader to local shuffle reader when smj converted to bhj in adaptive execution #25295

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
JkSelf wants to merge 11 commits into from
Closed

[SPARK-28560][SQL] Optimize shuffle reader to local shuffle reader when smj converted to bhj in adaptive execution #25295

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
bf6b6ad
change the shuffle read to local shuffle read when the sort merge joi…
JkSelf Jul 29, 2019
3b3d6ce
fix the can't zip rdd with unequal numbers of partitions and resolve …
JkSelf Aug 2, 2019
bdce1c3
move the rule of converting the shuffle reader to local shuffle reade…
JkSelf Sep 18, 2019
e5f3c9a
resolve comments
JkSelf Sep 20, 2005
d21c99f
small rebase fix
JkSelf Sep 20, 2005
bb10b18
java doc style
JkSelf Sep 20, 2005
72ce82e
resolve the comments
JkSelf Oct 11, 2019
f0ffff6
code style
JkSelf Oct 11, 2019
f10e5b2
resolve the comments
JkSelf Oct 12, 2019
37111ff
resolve the small comments
JkSelf Oct 15, 2019
9c1dc55
resolve the conflicts
JkSelf Oct 15, 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
resolve the conflicts
  • Loading branch information
@JkSelf
JkSelf committed Oct 15, 2019
commit 9c1dc5538afce26c4e693e353d8d4ef4231bb78c
25 changes: 4 additions & 21 deletions core/src/main/scala/org/apache/spark/MapOutputTracker.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -757,12 +757,7 @@ private[spark] class MapOutputTrackerMaster(
startPartition,
endPartition,
statuses,
<<<<<<< HEAD
mapId)
=======
useOldFetchProtocol,
Some(mapId))
>>>>>>> resolve the comments
}
case None =>
Iterator.empty
Expand Down Expand Up @@ -823,12 +818,8 @@ private[spark] class MapOutputTrackerWorker(conf: SparkConf) extends MapOutputTr
s"partitions $startPartition-$endPartition")
val statuses = getStatuses(shuffleId)
try {
<<<<<<< HEAD
MapOutputTracker.convertMapStatuses(shuffleId, startPartition, endPartition, statuses, mapId)
=======
MapOutputTracker.convertMapStatuses(shuffleId, startPartition, endPartition, statuses,
useOldFetchProtocol, Some(mapId))
>>>>>>> resolve the comments
MapOutputTracker.convertMapStatuses(shuffleId, startPartition, endPartition,
statuses, Some(mapId))
} catch {
case e: MetadataFetchFailedException =>
// We experienced a fetch failure so our mapStatuses cache is outdated; clear it:
Expand Down Expand Up @@ -980,7 +971,6 @@ private[spark] object MapOutputTracker extends Logging {
startPartition: Int,
endPartition: Int,
statuses: Array[MapStatus],
useOldFetchProtocol: Boolean,
mapId : Option[Int] = None): Iterator[(BlockManagerId, Seq[(BlockId, Long, Int)])] = {
assert (statuses != null)
val splitsByAddress = new HashMap[BlockManagerId, ListBuffer[(BlockId, Long, Int)]]
Expand All @@ -994,15 +984,8 @@ private[spark] object MapOutputTracker extends Logging {
for (part <- startPartition until endPartition) {
val size = status.getSizeForBlock(part)
if (size != 0) {
if (useOldFetchProtocol) {
// While we use the old shuffle fetch protocol, we use mapIndex as mapId in the
// ShuffleBlockId.
splitsByAddress.getOrElseUpdate(status.location, ListBuffer()) +=
((ShuffleBlockId(shuffleId, mapIndex, part), size, mapIndex))
} else {
splitsByAddress.getOrElseUpdate(status.location, ListBuffer()) +=
((ShuffleBlockId(shuffleId, status.mapTaskId, part), size, mapIndex))
}
splitsByAddress.getOrElseUpdate(status.location, ListBuffer()) +=
((ShuffleBlockId(shuffleId, status.mapId, part), size, mapIndex))
}
}
}
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/org/apache/spark/shuffle/BlockStoreShuffleReader.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ private[spark] class BlockStoreShuffleReader[K, C](

/** Read the combined key-values for this reduce task */
override def read(): Iterator[Product2[K, C]] = {
val blocksByAddress = (mapId) match {
val blocksByAddress = mapId match {
case (Some(mapId)) => mapOutputTracker.getMapSizesByExecutorId(
handle.shuffleId,
startPartition,
Expand Down