Github user kayousterhout commented on a diff in the pull request:
https://github.com/apache/spark/pull/7268#discussion_r34727284
--- 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) {
--- End diff --
Good point - done!
---
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]
