[
https://issues.apache.org/jira/browse/SPARK-48290?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18099113#comment-18099113
]
Vivek Gangavarapu commented on SPARK-48290:
-------------------------------------------
I spent some time reproducing it on current master (5.0.0-SNAPSHOT). It still
happens, and I think the root cause is clear.
Local repro: inner join with ~95% of rows on a single key,
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=1MB, everything
else default. The only thing I varied was spark.sql.shuffle.partitions:
||shuffle.partitions||extra conf||skew detected||
|1999|(defaults)|yes – SortMergeJoin(skew=true)|
|2001|(defaults)|no|
|4000|(defaults)|no|
|2001|spark.shuffle.accurateBlockSkewedFactor=5.0|yes|
|2001|spark.shuffle.minNumPartitionsToHighlyCompress=100000|yes|
So it isn't really about AQE. Once the shuffle has more than
spark.shuffle.minNumPartitionsToHighlyCompress (2000) partitions, map statuses
switch to HighlyCompressedMapStatus, which only keeps exact sizes for blocks
above spark.shuffle.accurateBlockThreshold (100MB) and reports avgSize for all
the rest. In your case a 263GB reduce partition spread across 10335 mappers is
only ~25MB per block, well under 100MB, so every block gets flattened to the
average. MapOutputTracker.getStatistics then sums those per reduce id, and AQE
sees min == median == max – exactly the log line you posted. That also explains
why spark.shuffle.accurateBlockThreshold didn't help no matter how low you set
it: what matters is the per-block size, not the per-partition total.
The interesting part is that the fix for this already exists. SPARK-36967 added
spark.shuffle.accurateBlockSkewedFactor in 3.3 to keep skewed blocks accurate,
but it ships disabled (-1.0), even though the config doc recommends setting it
to the same value as spark.sql.adaptive.skewJoin.skewedPartitionFactor. Setting
it to 5.0 fixes your case, as the table above shows.
I'd like to make that work out of the box. My concern with simply flipping the
default is cost: the enabled path does a full uncompressedSizes.sorted on every
map task, which is O(P log P) plus an array copy per task, and that is
presumably why it was left off. So what I'm planning is:
1. rewrite HighlyCompressedMapStatus.apply to do a single pass – a bounded
min-heap for the top-K skewed blocks and quickselect for the median, instead of
sorting the whole array
2. then flip spark.shuffle.accurateBlockSkewedFactor to 5.0 by default, keeping
-1.0 as the opt-out
3. add a benchmark (extending MapStatusesConvertBenchmark) with numbers at
2k/10k/50k partitions, plus the driver-side memory delta from the extra
accurate entries, so the default change is backed by data rather than opinion
Does that direction sound reasonable? If there are no objections I'll put up a
PR. Happy to split it into the algorithmic change first and the default flip as
a follow-up if reviewers would rather see those separately.
> AQE not working when joining dataframes with more than 2000 partitions
> ----------------------------------------------------------------------
>
> Key: SPARK-48290
> URL: https://issues.apache.org/jira/browse/SPARK-48290
> Project: Spark
> Issue Type: Bug
> Components: Optimizer, SQL
> Affects Versions: 3.3.2, 3.5.1
> Environment: spark-standalone
> spark3.5.1
> Reporter: André F.
> Priority: Major
>
> We are joining 2 large dataframes with a considerable skew on the left side
> in one specific key (>2000 skew ratio).
> {code:java}
> left side num partitions: 10335
> right side num partitions: 1241
> left side num rows: 20181947343
> right side num rows: 107462219 {code}
> Since we have `{{{}spark.sql.adaptive.enabled{}}} ` we expect AQE to act
> during the join, dealing with the skewed partition automatically.
> During their join, we can see the following log indicating that the skew was
> not detected since their statistics looks weirdly equal for min/median/max
> sizes:
> {code:java}
> OptimizeSkewedJoin: number of skewed partitions: left 0, right 0
> OptimizeSkewedJoin:
> Optimizing skewed join.
> Left side partitions size info:
> median size: 780925482, max size: 780925482, min size: 780925482, avg size:
> 780925482
> Right side partitions size info:
> median size: 3325797, max size: 3325797, min size: 3325797, avg size: 3325797
> {code}
> Looking at this log line and the spark configuration possibilities, our two
> main hypotheses to work around this behavior and correctly detect the skew
> were:
> # Increasing the `minNumPartitionsToHighlyCompress` so that Spark doesn’t
> convert the statistics into a `CompressedMapStatus` and therefore is able to
> identify the skewed partition.
> # Allowing spark to use a `HighlyCompressedMapStatus`, but change other
> configurations such as `spark.shuffle.accurateBlockThreshold` and
> `spark.shuffle.accurateBlockSkewedFactor` so that even then the size of the
> skewed partitions/blocks is accurately registered and consequently used in
> the optimization.
> We tried different values for `spark.shuffle.accurateBlockThreshold` (even
> absurd ones like 1MB) and nothing seem to work. The statistics indicates that
> the min/median and max are the same somehow and thus, the skew is not
> detected.
> However, when forcibly reducing `spark.sql.shuffle.partitions` to less than
> 2000 partitions, the statistics looked correct and the optimized skewed join
> acts as it should:
> {code:java}
> OptimizeSkewedJoin: number of skewed partitions: left 1, right 0
> OptimizeSkewedJoin: Left side partition 42 (263 GB) is skewed, split it into
> 337 parts.
> OptimizeSkewedJoin:
> Optimizing skewed join.
> Left side partitions size info:
> median size: 862803419, max size: 282616632301, min size: 842320875, avg
> size: 1019367139
> Right side partitions size info:
> median size: 4320067, max size: 4376957, min size: 4248989, avg size: 4319766
> {code}
> Should we assume that the statistics are becoming corrupted when Spark uses
> `HighlyCompressedMapStatus`? Should we try another configuration property to
> try to work around this problem? (Assuming that fine tuning all dataframes in
> skewed joins in our ETL to have less than 2000 partitions is not an option)
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
