Vivek1106-04 commented on code in PR #57528:
URL: https://github.com/apache/spark/pull/57528#discussion_r3707009427
##########
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:
You're right, and the test was worse than under-powered: it passed because
the feature never ran. 1900 + 100 = 2000 and MapStatus.apply compares with >,
so both measurements built a CompressedMapStatus, perBlock was exactly 0.0, and
0.0 < 16 held. The assertion was one-sided with no floor and no type check, so
"the skew path stored nothing" and "the skew path stored something compact"
were indistinguishable to it.
Fixed in 4196cb402ee: 1,901 small blocks plus 100 skewed, an assertion that
the block count exceeds spark.shuffle.minNumPartitionsToHighlyCompress, an
isInstanceOf[HighlyCompressedMapStatus] check on each measured status, and a
check that the expected number of skewed sizes actually came back accurate
before anything is measured.
--
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]
