sunchao commented on code in PR #57528:
URL: https://github.com/apache/spark/pull/57528#discussion_r3653391516
##########
core/src/main/scala/org/apache/spark/scheduler/MapStatus.scala:
##########
@@ -288,20 +288,21 @@ private[spark] object HighlyCompressedMapStatus {
.getOrElse(config.SHUFFLE_ACCURATE_BLOCK_THRESHOLD.defaultValue.get)
val threshold =
if (accurateBlockSkewedFactor > 0) {
- val sortedSizes = uncompressedSizes.sorted
- val medianSize: Long = Utils.median(sortedSizes, true)
val maxAccurateSkewedBlockNumber =
Math.min(
Option(SparkEnv.get)
.map(_.conf.get(config.SHUFFLE_MAX_ACCURATE_SKEWED_BLOCK_NUMBER))
.getOrElse(config.SHUFFLE_MAX_ACCURATE_SKEWED_BLOCK_NUMBER.defaultValue.get),
totalNumBlocks
)
+ // Only two order statistics are needed here, so they are selected in
O(totalNumBlocks)
+ // instead of sorting the sizes, which every map task would otherwise
pay for.
+ val sizes = uncompressedSizes.clone()
+ val medianSize: Long = Utils.medianInPlace(sizes)
+ val smallestAccurateSize =
+ Utils.nthSmallest(sizes, totalNumBlocks -
maxAccurateSkewedBlockNumber)
Review Comment:
[P1] Enforce the skewed-block cap when the cutoff has ties
`nthSmallest` returns the 100th-largest block *size*, but the classification
below records every block with `size >= threshold`. This means ties at the
cutoff bypass `spark.shuffle.maxAccurateSkewedBlockNumber`.
For example, with the proposed defaults and 50,000 reduce partitions,
consider 25,001 blocks of 1 KiB and 24,999 blocks of 8 KiB. The median is 1
KiB, the 100th-largest size is 8 KiB, and the threshold is 8 KiB, so **all
24,999 larger blocks are recorded instead of at most 100**. `writeExternal`
writes at least five bytes per recorded entry, so a 10,000-map stage produces
1,249,950,000 bytes of uncompressed map-status entries, before the much larger
driver-side collection overhead. The base branch's disabled default does not
record any of these blocks.
Since this PR enables this path by default, please enforce the actual
cardinality bound, including deterministic handling of ties, while preserving
mandatory accurate recording above `spark.shuffle.accurateBlockThreshold`.
Please also add a regression test containing more than 100 equal-sized blocks
at the cutoff.
##########
core/src/main/scala/org/apache/spark/util/Utils.scala:
##########
@@ -3152,6 +3152,59 @@ private[spark] object Utils
}
}
+ /**
+ * Return the n-th smallest element (0-indexed) of a long array, reordering
`sizes` in place.
+ *
+ * This is a quickselect, which runs in O(sizes.length) on average. Callers
that only need a few
+ * order statistics should prefer it over sorting the whole array, and must
not rely on the
+ * element order of `sizes` afterwards.
+ */
+ def nthSmallest(sizes: Array[Long], n: Int): Long = {
+ require(n >= 0 && n < sizes.length, s"n must be in [0, ${sizes.length})
but was $n")
+ var low = 0
+ var high = sizes.length - 1
+ while (low < high) {
+ // The middle element keeps already sorted and reverse sorted inputs,
which are both common
+ // for shuffle block sizes, away from the quadratic worst case.
+ val pivot = sizes(low + (high - low) / 2)
Review Comment:
[P1] Bound the worst-case cost of default-enabled quickselect
Always selecting the middle element makes the algorithm quadratic for a
valid organ-pipe reducer-size distribution: `Array.tabulate(n)(i => math.min(i,
n - 1 - i).toLong)`. Replaying the exact selections made by
`HighlyCompressedMapStatus` produces 805,003 comparisons at 2,048 partitions,
3,174,774 at 4,096, 12,682,753 at 8,192, and 50,455,932 at 16,384. Doubling the
number of partitions approximately quadruples the work.
This becomes a production regression because the PR changes
`spark.shuffle.accurateBlockSkewedFactor` from `-1.0` to `5.0`, making every
map task in a qualifying shuffle execute these selections; the base default
performs no selection. At 16,384 partitions, a 10,000-map stage would require
more than 504 billion comparisons just to construct map statuses.
Please use a worst-case-bounded selection algorithm, or an
introspective/randomized pivot with a safe fallback, and add an organ-pipe
input to the tests and benchmark.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
