sunchao commented on code in PR #57528:
URL: https://github.com/apache/spark/pull/57528#discussion_r3714053339
##########
core/src/main/scala/org/apache/spark/scheduler/MapStatus.scala:
##########
@@ -286,40 +334,63 @@ private[spark] object HighlyCompressedMapStatus {
Option(SparkEnv.get)
.map(_.conf.get(config.SHUFFLE_ACCURATE_BLOCK_THRESHOLD))
.getOrElse(config.SHUFFLE_ACCURATE_BLOCK_THRESHOLD.defaultValue.get)
+ // Sizes at or above `threshold` are recorded accurately. At most
+ // SHUFFLE_MAX_ACCURATE_SKEWED_BLOCK_NUMBER of them are strictly larger
than `skewCutoff`, and
+ // those are held as ids and sizes. There can be arbitrarily many blocks
of exactly
+ // `skewCutoff`, but they all share that one size, so only their
membership is recorded.
+ var skewCutoff = Long.MaxValue
+ var recordSkewedTies = false
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 firstAccurateIdx = totalNumBlocks - maxAccurateSkewedBlockNumber
+ skewCutoff = Utils.nthSmallest(sizes, firstAccurateIdx)
val skewSizeThreshold =
- Math.max(
- medianSize * accurateBlockSkewedFactor,
- sortedSizes(totalNumBlocks - maxAccurateSkewedBlockNumber).toDouble
- )
- Math.min(shuffleAccurateBlockThreshold.toDouble, skewSizeThreshold)
+ Math.max(medianSize * accurateBlockSkewedFactor, skewCutoff.toDouble)
+ val skewThreshold = Math.min(shuffleAccurateBlockThreshold.toDouble,
skewSizeThreshold)
+ // Every map task of a shuffle sees the same block size distribution,
so any rule that
+ // records only some of the blocks tied at the cutoff hides the same
reducers in every map
+ // status, and MapOutputTracker sums those. The hidden ones fall back
to `avgSize`, which
+ // they raise in the process, so AQE's skew threshold moves up along
with the sizes it is
+ // compared against. Recording all of the ties is what keeps them
visible.
+ recordSkewedTies = skewCutoff > 0 && skewThreshold <=
skewCutoff.toDouble
+ skewThreshold
} else {
// Disable skew detection if accurateBlockSkewedFactor <= 0
shuffleAccurateBlockThreshold.toDouble
}
- val hugeBlockSizes = mutable.Map.empty[Int, Byte]
+ val hugeBlockIds = mutable.ArrayBuilder.make[Int]
+ val hugeBlockSizes = mutable.ArrayBuilder.make[Byte]
+ val skewedBlocks = new RoaringBitmap()
while (i < totalNumBlocks) {
val size = uncompressedSizes(i)
if (size > 0) {
numNonEmptyBlocks += 1
// Huge blocks are not included in the calculation for average size,
thus size for smaller
- // blocks is more accurate.
- if (size < threshold) {
+ // blocks is more accurate. Blocks tied at the cutoff are excluded
from it as well: they
+ // are recorded accurately, so folding them into the average would
both lose nothing and
+ // inflate the size reported for the blocks that really are average.
+ val isHuge = size >= shuffleAccurateBlockThreshold ||
+ (size >= threshold && size > skewCutoff)
+ if (isHuge) {
+ hugeBlockIds += i
+ hugeBlockSizes += MapStatus.compressSize(size)
+ } else if (recordSkewedTies && size == skewCutoff) {
+ skewedBlocks.add(i)
Review Comment:
[P2] Bound cutoff-tie bitmap growth for non-contiguous reducer IDs
Recording every `skewCutoff` tie in `skewedBlocks` means
`spark.shuffle.maxAccurateSkewedBlockNumber=100` no longer bounds retained or
serialized metadata. The new regression test makes all 24,999 tied reducer IDs
consecutive, so `runOptimize()` compresses them into one run; that is not
representative of arbitrary reducer IDs.
Using this exact head and its declared RoaringBitmap 1.6.10 dependency, with
50,000 reducers, 25,001 blocks of 1 KiB, and 24,999 blocks of 8 KiB:
- Consecutive tied reducer IDs: bitmap retained payload 18 bytes; serialized
bitmap 15 bytes.
- Alternating tied reducer IDs with the identical block-size distribution:
retained payload 8,202 bytes; serialized bitmap 8,208 bytes, after the same
`trim()` and `runOptimize()` calls as production.
Thus 10,000 mappers retain at least 82,020,000 bitmap bytes, and 100,000
mappers retain at least 820,200,000 bytes, despite the configured 100-entry
cap. At 1,048,576 alternating reducers, the measured bitmap is 131,112 bytes
per map status, or 1,311,120,000 retained bytes for 10,000 mappers. These are
bitmap-payload lower bounds; object overhead and cached serialized/broadcast
copies are additional, and serialized sizes above are before compression.
The current `retained < 8 * 1024` test would already fail for the
50,000-partition alternating layout. Please bound tie metadata independently of
reducer count while preserving AQE visibility, and add an
interleaved/fragmented-cutoff-tie retained-memory and serialization regression.
--
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]
