sunchao commented on code in PR #57528:
URL: https://github.com/apache/spark/pull/57528#discussion_r3693013574
##########
core/src/main/scala/org/apache/spark/internal/config/package.scala:
##########
@@ -1441,12 +1441,16 @@ package object config {
ConfigBuilder("spark.shuffle.accurateBlockSkewedFactor")
.doc("A shuffle block is considered as skewed and will be accurately
recorded in " +
"HighlyCompressedMapStatus if its size is larger than this factor
multiplying " +
- "the median shuffle block size or SHUFFLE_ACCURATE_BLOCK_THRESHOLD. It
is " +
- "recommended to set this parameter to be the same as
SKEW_JOIN_SKEWED_PARTITION_FACTOR." +
- "Set to -1.0 to disable this feature by default.")
+ "the median shuffle block size or SHUFFLE_ACCURATE_BLOCK_THRESHOLD.
Otherwise every " +
+ "block below SHUFFLE_ACCURATE_BLOCK_THRESHOLD is reported as the
average block size, " +
+ "which hides skew from adaptive query execution once a shuffle has
more than " +
+ "SHUFFLE_MIN_NUM_PARTS_TO_HIGHLY_COMPRESS partitions. At most " +
+ "SHUFFLE_MAX_ACCURATE_SKEWED_BLOCK_NUMBER block sizes are recorded
accurately per map " +
+ "task. The default matches SKEW_JOIN_SKEWED_PARTITION_FACTOR. Set to
-1.0 to disable " +
+ "this feature.")
.version("3.3.0")
.doubleConf
- .createWithDefault(-1.0)
+ .createWithDefault(5.0)
Review Comment:
[P2] Bound retained driver heap before enabling skew recording by default
Changing the default from `-1.0` to `5.0` means each qualifying map status
can now retain up to 100 additional entries in `mutable.Map[Int, Byte]`, and
the driver retains those deserialized map statuses. The PR estimates five bytes
per entry (500 bytes per mapper; approximately 5 MB for 10,000 mappers), but
those are only the `writeExternal` payload bytes. The live representation is a
Scala `mutable.HashMap` with nodes, bucket arrays, and typically boxed reducer
IDs.
Using the exact Scala 2.13.18 version declared by this head, I measured
10,000 `mutable.HashMap`s containing 100 reducer IDs in 1900-1999: they
retained **57,752,240 additional heap bytes**, or **5,775 bytes per mapper**,
versus the claimed **5,000,000 bytes** - an **11.55x** difference before cached
serialized statuses or broadcast copies. The same distribution projects to
approximately 551 MiB of additional retained heap at 100,000 mappers; the base
branch's disabled default records none of these subthreshold entries.
Please measure retained `MapOutputTracker` heap for realistic high-map
stages and use compact/primitive-backed storage or otherwise bound the
driver-memory cost before enabling this path by default.
##########
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] Rotate cutoff ties across map tasks to avoid hiding skew
The new tie cap always admits the lowest reducer IDs, and each mapper
observes the same tied reducers. `MapOutputTracker.getStatistics` sums each
reducer's `getSizeForBlock` across map tasks, so a tied reducer that loses here
is permanently represented by `avgSize`.
For example, use the defaults with 2,001 reducers and 10,000 map tasks. On
every mapper, reducers 0-1899 have 10 KiB blocks and reducers 1900-2000 have
100 KiB blocks. The median is 10 KiB, the cutoff is 100 KiB, and the 100-entry
cap always selects reducers 1900-1999. Reducer 2000 instead reports `(1900 *
10240 + 102400) / 1901 = 10288` bytes per mapper, so AQE sees **102,880,000
bytes** although its real size is **1,024,000,000 bytes**. The real partition
exceeds the skew threshold `max(256 MiB, 5 * 102,880,000) = 514,400,000` bytes,
but is never detected or split; the other 100 equally skewed reducers are
detected.
Please rotate deterministic cutoff-tie selection using the mapper identity
(for example `mapTaskId`) and add a regression test involving multiple map
statuses, while preserving the 100-entry cap.
--
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]
