Skip to content

[SPARK-21474][CORE] Make number of parallel fetches from a reducer configurable #18683

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
raajay wants to merge 2 commits into from
Closed

[SPARK-21474][CORE] Make number of parallel fetches from a reducer configurable #18683

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9bf1f9f
Make num parallel fetches from a reducer configurable
raajay Jul 19, 2017
1fc225d
Merge branch 'master' into raajay-configure-parallel-requests
raajay Jul 20, 2017
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
11 changes: 7 additions & 4 deletions core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetcherIterator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ import javax.annotation.concurrent.GuardedBy
import scala.collection.mutable
import scala.collection.mutable.{ArrayBuffer, HashMap, HashSet, Queue}

import org.apache.spark.{SparkException, TaskContext}
import org.apache.spark.{SparkEnv, SparkException, TaskContext}
import org.apache.spark.internal.Logging
import org.apache.spark.network.buffer.{FileSegmentManagedBuffer, ManagedBuffer}
import org.apache.spark.network.shuffle.{BlockFetchingListener, ShuffleClient, TempShuffleFileManager}
Expand Down Expand Up @@ -256,10 +256,13 @@ final class ShuffleBlockFetcherIterator(
}

private[this] def splitLocalRemoteBlocks(): ArrayBuffer[FetchRequest] = {
// Make remote requests at most maxBytesInFlight / 5 in length; the reason to keep them
// smaller than maxBytesInFlight is to allow multiple, parallel fetches from up to 5
val numParallelFetchRequests = SparkEnv.get.conf
.getInt("spark.reducer.numParallelFetchRequests", 5)
// Make remote requests at most maxBytesInFlight / numParallelFetchRequests in length;
// the reason to keep them smaller than maxBytesInFlight is to allow multiple,
// parallel fetches from up to numParallelFetchRequests
// nodes, rather than blocking on reading output from one node.
val targetRequestSize = math.max(maxBytesInFlight / 5, 1L)
val targetRequestSize = math.max(maxBytesInFlight / numParallelFetchRequests, 1L)
logDebug("maxBytesInFlight: " + maxBytesInFlight + ", targetRequestSize: " + targetRequestSize
+ ", maxBlocksInFlightPerAddress: " + maxBlocksInFlightPerAddress)

Expand Down