Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/7268#discussion_r34727337
--- Diff: core/src/main/scala/org/apache/spark/MapOutputTracker.scala ---
@@ -421,23 +432,35 @@ private[spark] object MapOutputTracker extends
Logging {
}
}
- // Convert an array of MapStatuses to locations and sizes for a given
reduce ID. If
- // any of the statuses is null (indicating a missing location due to a
failed mapper),
- // throw a FetchFailedException.
+ /**
+ * Converts an array of MapStatuses for a given reduce ID to a sequence
that, for each block
+ * manager ID, lists the shuffle block ids and corresponding shuffle
block sizes stored at that
+ * block manager.
+ *
+ * If any of the statuses is null (indicating a missing location due to
a failed mapper),
+ * throws a FetchFailedException.
+ *
+ * @return A sequence of 2-item tuples, where the first item in the
tuple is a BlockManagerId,
+ * and the second item is a sequence of (shuffle block id,
shuffle block size) tuples
+ * describing the shuffle blocks that are stored at that block
manager.
+ */
private def convertMapStatuses(
shuffleId: Int,
reduceId: Int,
- statuses: Array[MapStatus]): Array[(BlockManagerId, Long)] = {
+ statuses: Array[MapStatus]): Seq[(BlockManagerId, Seq[(BlockId,
Long)])] = {
assert (statuses != null)
- statuses.map {
- status =>
- if (status == null) {
- logError("Missing an output location for shuffle " + shuffleId)
- throw new MetadataFetchFailedException(
- shuffleId, reduceId, "Missing an output location for shuffle "
+ shuffleId)
- } else {
- (status.location, status.getSizeForBlock(reduceId))
- }
+ val splitsByAddress = new HashMap[BlockManagerId,
ArrayBuffer[(BlockId, Long)]]
+ for ((status, index) <- statuses.zipWithIndex) {
+ if (status == null) {
+ val errorMessage = s"Missing an output location for shuffle
$shuffleId"
+ logError(errorMessage)
+ throw new MetadataFetchFailedException(shuffleId, reduceId,
errorMessage)
+ } else {
+ splitsByAddress.getOrElseUpdate(status.location, ArrayBuffer()) +=
+ ((ShuffleBlockId(shuffleId, index, reduceId),
status.getSizeForBlock(reduceId)))
--- End diff --
That's true; I was thinking of this conversion as a convenience for the
caller, since the only caller needs the data in this format, so I think this is
OK.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
