sunchao commented on code in PR #57528:
URL: https://github.com/apache/spark/pull/57528#discussion_r3706594029
##########
core/src/main/scala/org/apache/spark/scheduler/MapStatus.scala:
##########
@@ -315,11 +331,21 @@ private[spark] object HighlyCompressedMapStatus {
numNonEmptyBlocks += 1
// Huge blocks are not included in the calculation for average size,
thus size for smaller
// blocks is more accurate.
- if (size < threshold) {
+ var isAccurate = size >= shuffleAccurateBlockThreshold
+ if (!isAccurate && size >= threshold) {
+ if (size > skewCutoff) {
+ isAccurate = true
+ } else if (numTiedSkewedBlocksToRecord > 0) {
+ // Ties are broken by block index so that the map status stays
deterministic.
Review Comment:
[P2] The new rotation still leaves genuinely skewed reducers invisible
This remains reproducible at `d3be2ba389d6594077219f52be0d5c5919e068a5`. The
new `floorMod(mapTaskId, numTiedSkewedBlocks)` advances a 100-entry
tied-reducer window by only one position per consecutive mapper. Therefore `M`
mappers cover only `100 + M - 1` tied reducers before wraparound.
Concrete default-configuration reproduction:
- 2,001 reduce partitions and 10 map tasks.
- Reducers 0-1880 produce 5 MiB blocks on every mapper.
- Reducers 1881-2000 produce 50 MiB blocks on every mapper, below the 100
MiB mandatory-accuracy threshold.
- The accurate-skew cap remains 100 per mapper.
Each status reports `avgSize = 5,739,312` bytes for unselected blocks.
Reducers 1990-2000 never enter any accurate window, so `MapOutputTracker`
reports `57,393,120` bytes although each reducer actually contains
`524,288,000` bytes. AQE computes a skew threshold of `max(256 MiB, 5 *
57,393,120) = 286,965,600` bytes, so these genuinely skewed reducers remain
unsplit. Overall, 19 of the 120 tied reducers miss the threshold.
There is also a second failure mode even when rotation distributes
selections evenly: with 1,801 reducers producing 10 KiB blocks, 200 tied
reducers producing 100 KiB blocks, and 10,000 mappers, each tied reducer is
recorded on half the mappers. Unselected ties inflate `avgSize` to `15,087`
bytes, so each skewed reducer is reported as `636,320,000` bytes against an AQE
threshold of `754,350,000`, even though its actual size is `1,024,000,000`. All
200 genuinely skewed reducers are missed.
The new regression only covers 101 ties and 2,000 mappers, where almost
every tied block is recorded on almost every mapper. Please cover both
fewer-mapper/larger-tie groups and fairly sampled larger tie groups, while
preserving the per-status cap and AQE visibility.
##########
core/src/test/scala/org/apache/spark/scheduler/MapStatusSuite.scala:
##########
@@ -289,4 +289,172 @@ class MapStatusSuite extends SparkFunSuite {
"Only tracked skewed block size is accurate")
}
}
+
+ test("SPARK-48290: skewed blocks are recorded accurately with the default
configuration") {
+ val emptyBlocksLength = 3
+ val smallBlocksLength = 3000
+ val skewedBlocksLength = 5
+ // Well above the median block size, but below
SHUFFLE_ACCURATE_BLOCK_THRESHOLD, which is the
+ // case for a skewed partition whose rows are spread over a large number
of map tasks.
+ val skewedBlockSize = 50 * 1024 * 1024L
+
+ // No skew related config is set: this asserts the out of the box behavior.
+ val conf = new SparkConf()
+ val env = mock(classOf[SparkEnv])
+ doReturn(conf).when(env).conf
+ SparkEnv.set(env)
+
+ val emptyBlocks = createArray(emptyBlocksLength, 0L)
+ val smallBlocks = Array.tabulate[Long](smallBlocksLength)(i => i + 1)
+ val skewedBlocks = createArray(skewedBlocksLength, skewedBlockSize)
+ val allBlocks = emptyBlocks ++: smallBlocks ++: skewedBlocks
+ assert(skewedBlockSize < conf.get(config.SHUFFLE_ACCURATE_BLOCK_THRESHOLD),
+ "the skewed blocks must not be tracked as huge blocks")
+ val avg = smallBlocks.sum / smallBlocks.length
+
+ val loc = BlockManagerId("a", "b", 10)
+ val mapTaskAttemptId = 5
+ val status = compressAndDecompressMapStatus(MapStatus(loc, allBlocks,
mapTaskAttemptId))
+ assert(status.isInstanceOf[HighlyCompressedMapStatus])
+ for (i <- 0 until emptyBlocksLength) {
+ assert(status.getSizeForBlock(i) === 0L)
+ }
+ for (i <- 0 until smallBlocksLength) {
+ assert(status.getSizeForBlock(emptyBlocksLength + i) === avg,
+ "the average size must not be inflated by the skewed blocks")
+ }
+ for (i <- 0 until skewedBlocksLength) {
+ assert(status.getSizeForBlock(emptyBlocksLength + smallBlocksLength + i)
===
+ compressAndDecompressSize(skewedBlockSize),
+ "skewed block sizes must be accurate so that AQE can detect the skew")
+ }
+ }
+
+ test("SPARK-48290: blocks tied at the cutoff size do not bypass the accurate
skewed block " +
+ "limit") {
+ // The small blocks are the majority, so the median block size is
smallBlockSize. Without the
+ // limit, all of the tied blocks would be recorded, which for a stage of
10000 map tasks would
+ // add more than a gigabyte of map status entries.
+ val smallBlocksLength = 25001
+ val tiedBlocksLength = 24999
+ val smallBlockSize = 1024L
+ // Far more blocks share this size than may be recorded, and it is the
cutoff size itself:
+ // it is above the median times the skew factor, so the skew threshold is
exactly this size.
+ val tiedBlockSize = 8 * 1024L
+ val maxAccurateSkewedBlockNumber =
+ config.SHUFFLE_MAX_ACCURATE_SKEWED_BLOCK_NUMBER.defaultValue.get
+
+ // No skew related config is set: this asserts the out of the box behavior.
+ val conf = new SparkConf()
+ val env = mock(classOf[SparkEnv])
+ doReturn(conf).when(env).conf
+ SparkEnv.set(env)
+
+ val allBlocks = createArray(smallBlocksLength, smallBlockSize) ++:
+ createArray(tiedBlocksLength, tiedBlockSize)
+ assert(tiedBlockSize < conf.get(config.SHUFFLE_ACCURATE_BLOCK_THRESHOLD),
+ "the tied blocks must not be recorded as huge blocks")
+ assert(Utils.median(allBlocks, false) *
+ conf.get(config.SHUFFLE_ACCURATE_BLOCK_SKEWED_FACTOR) < tiedBlockSize,
+ "the cutoff size, not the median times the skew factor, must set the
skew threshold")
+ val numSmallBlocks = allBlocks.length - maxAccurateSkewedBlockNumber
+ val avg =
+ (smallBlockSize * smallBlocksLength +
+ tiedBlockSize * (tiedBlocksLength - maxAccurateSkewedBlockNumber)) /
numSmallBlocks
+
+ val loc = BlockManagerId("a", "b", 10)
+ val mapTaskId = 5L
+ val status = compressAndDecompressMapStatus(MapStatus(loc, allBlocks,
mapTaskId))
+ assert(status.isInstanceOf[HighlyCompressedMapStatus])
+ for (i <- 0 until smallBlocksLength) {
+ assert(status.getSizeForBlock(i) === avg)
+ }
+ // The recorded ties are a window of maxAccurateSkewedBlockNumber tied
blocks, in block index
+ // order, starting at the offset the map task id rotates to.
+ val firstAccurateTie = (mapTaskId % tiedBlocksLength).toInt
+ for (i <- 0 until tiedBlocksLength) {
+ val isAccurate =
+ i >= firstAccurateTie && i < firstAccurateTie +
maxAccurateSkewedBlockNumber
+ val expected = if (isAccurate) compressAndDecompressSize(tiedBlockSize)
else avg
+ assert(status.getSizeForBlock(smallBlocksLength + i) === expected,
+ "no more than maxAccurateSkewedBlockNumber blocks may be recorded
accurately")
+ }
+ }
+
+ test("SPARK-48290: blocks tied at the cutoff size rotate across map tasks so
that no reducer " +
+ "is hidden") {
+ // One more reducer is tied at the cutoff than may be recorded per map
task. Every map task
+ // sees the same distribution, so a fixed tie break would hide the same
reducer in every map
+ // status, and MapOutputTracker.getStatistics would report it as the
average block size.
+ val smallBlocksLength = 1900
+ val tiedBlocksLength = 101
+ val smallBlockSize = 10 * 1024L
+ val tiedBlockSize = 100 * 1024L
+ val numMapTasks = 2000
+ val maxAccurateSkewedBlockNumber =
+ config.SHUFFLE_MAX_ACCURATE_SKEWED_BLOCK_NUMBER.defaultValue.get
+ assert(tiedBlocksLength > maxAccurateSkewedBlockNumber)
+
+ // No skew related config is set: this asserts the out of the box behavior.
+ val conf = new SparkConf()
+ val env = mock(classOf[SparkEnv])
+ doReturn(conf).when(env).conf
+ SparkEnv.set(env)
+
+ val allBlocks = createArray(smallBlocksLength, smallBlockSize) ++:
+ createArray(tiedBlocksLength, tiedBlockSize)
+ val loc = BlockManagerId("a", "b", 10)
+ val statuses = (0 until numMapTasks).map { mapTaskId =>
+ compressAndDecompressMapStatus(MapStatus(loc, allBlocks, mapTaskId))
+ }
+
+ val accurateSize = compressAndDecompressSize(tiedBlockSize)
+ statuses.foreach { status =>
+ val numAccurate =
+ (smallBlocksLength until
allBlocks.length).count(status.getSizeForBlock(_) == accurateSize)
+ assert(numAccurate === maxAccurateSkewedBlockNumber,
+ "the accurate skewed block limit must still hold for every map task")
+ }
+
+ // Summing a reducer over the map statuses is what AQE sees. Each tied
reducer loses the tie on
+ // one map task out of tiedBlocksLength, so its total is off by that
fraction at worst, instead
+ // of collapsing to the average block size.
+ val realTotal = numMapTasks * accurateSize
+ for (i <- smallBlocksLength until allBlocks.length) {
+ val total = statuses.map(_.getSizeForBlock(i)).sum
+ assert(total > realTotal * 0.98,
+ s"reducer $i must not be hidden behind the average block size")
+ assert(total <= realTotal)
+ }
+ }
+
+ test("SPARK-48290: recorded skewed block sizes are held in compact storage")
{
+ // The driver retains one map status per map task for the lifetime of the
shuffle, so the
+ // accurately recorded sizes must not cost more than a few bytes each. A
mutable.Map[Int, Byte]
+ // costs an order of magnitude more, through its nodes, bucket array and
boxed reduce ids.
+ val smallBlocksLength = 1900
Review Comment:
[P2] Make the retained-memory regression test exercise
HighlyCompressedMapStatus
The test builds exactly `1900 + 100 = 2000` partitions. However,
`MapStatus.apply` selects `HighlyCompressedMapStatus` only when
`uncompressedSizes.length > spark.shuffle.minNumPartitionsToHighlyCompress`,
whose default is `2000`.
Consequently, both `retainedBytes(-1.0)` and `retainedBytes(5.0)`
instantiate `CompressedMapStatus`. That implementation ignores the skew factor,
so both retained sizes are identical, `perBlock == 0`, and this assertion
succeeds even if `HighlyCompressedMapStatus` regresses to the previously
reported boxed-map representation.
Please use at least 2,001 partitions, for example 1,901 small blocks plus
100 skewed blocks, and assert that each measured status is a
`HighlyCompressedMapStatus` before estimating its retained size.
--
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]
