Skip to content

Commit 6e72951

Browse files
VictsmMridul Muralidharan
authored and
Mridul Muralidharan
committed
[SPARK-36423][SHUFFLE] Randomize order of blocks in a push request to improve block merge ratio for push-based shuffle
### What changes were proposed in this pull request? On the client side, we are currently randomizing the order of push requests before processing each request. In addition we can further randomize the order of blocks within each push request before pushing them. In our benchmark, this has resulted in a 60%-70% reduction of blocks that fail to be merged due to bock collision (the existing block merge ratio is already pretty good in general, and this further improves it). ### Why are the changes needed? Improve block merge ratio for push-based shuffle ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Straightforward small change, no additional test needed. Closes #33649 from Victsm/SPARK-36423. Lead-authored-by: Min Shen <[email protected]> Co-authored-by: Min Shen <[email protected]> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>
1 parent 41b011e commit 6e72951

File tree

1 file changed

+9
-3
lines changed

1 file changed

+9
-3
lines changed

core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -242,10 +242,16 @@ private[spark] class ShuffleBlockPusher(conf: SparkConf) extends Logging {
242242
handleResult(PushResult(blockId, exception))
243243
}
244244
}
245+
// In addition to randomizing the order of the push requests, further randomize the order
246+
// of blocks within the push request to further reduce the likelihood of shuffle server side
247+
// collision of pushed blocks. This does not increase the cost of reading unmerged shuffle
248+
// files on the executor side, because we are still reading MB-size chunks and only randomize
249+
// the in-memory sliced buffers post reading.
250+
val (blockPushIds, blockPushBuffers) = Utils.randomize(blockIds.zip(
251+
sliceReqBufferIntoBlockBuffers(request.reqBuffer, request.blocks.map(_._2)))).unzip
245252
SparkEnv.get.blockManager.blockStoreClient.pushBlocks(
246-
address.host, address.port, blockIds.toArray,
247-
sliceReqBufferIntoBlockBuffers(request.reqBuffer, request.blocks.map(_._2)),
248-
blockPushListener)
253+
address.host, address.port, blockPushIds.toArray,
254+
blockPushBuffers.toArray, blockPushListener)
249255
}
250256

251257
/**

0 commit comments

Comments
 (0)