Victsm commented on a change in pull request #30164: URL: https://github.com/apache/spark/pull/30164#discussion_r524541577
########## File path: core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala ########## @@ -657,6 +688,43 @@ class BlockManagerMasterEndpoint( } } + private def getShufflePushMergerLocations( + numMergersNeeded: Int, + hostsToFilter: Set[String]): Seq[BlockManagerId] = { + val blockManagersWithExecutors = blockManagerIdByExecutor.groupBy(_._2.host) + .mapValues(_.head).values.map(_._2).toSet + val filteredBlockManagersWithExecutors = blockManagersWithExecutors + .filterNot(x => hostsToFilter.contains(x.host)) + val filteredMergersWithExecutors = filteredBlockManagersWithExecutors.map( + x => BlockManagerId(x.executorId, x.host, StorageUtils.externalShuffleServicePort(conf))) + + // Enough mergers are available as part of active executors list + if (filteredMergersWithExecutors.size >= numMergersNeeded) { + filteredMergersWithExecutors.toSeq + } else { + // Delta mergers added from inactive mergers list to the active mergers list + val filteredMergersWithExecutorsHosts = filteredMergersWithExecutors.map(_.host) + val filteredMergersWithoutExecutors = shuffleMergerLocations.values + .filterNot(x => hostsToFilter.contains(x.host)) + .filterNot(x => filteredMergersWithExecutorsHosts.contains(x.host)) + val randomFilteredMergersLocations = + if (filteredMergersWithoutExecutors.size > + numMergersNeeded - filteredMergersWithExecutors.size) { + Utils.randomize(filteredMergersWithoutExecutors) + } else { + filteredMergersWithoutExecutors + } + filteredMergersWithExecutors.toSeq ++ randomFilteredMergersLocations + .take(numMergersNeeded - filteredMergersWithExecutors.size) Review comment: If # elements in `randomFilteredMergersLocations` is less than `numMergersNeeded - filteredMergersWithExecutors.size`, `take()` would just return all elements. What's the reason for `take()` cannot be called here if `randomize()` is not performed? ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org