[SPARK-31864][SQL] Adjust AQE skew join trigger condition #28669

maryannxue · 2020-05-28T22:20:45Z

What changes were proposed in this pull request?

This PR makes a minor change in deciding whether a partition is skewed by comparing the partition size to the median size of coalesced partitions instead of median size of raw partitions before coalescing.

Why are the changes needed?

This change is line with target size criteria of splitting skew join partitions and can also cope with situations of extra empty partitions caused by over-partitioning. This PR has also improved skew join tests in AQE tests.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Updated UTs.

maryannxue · 2020-05-28T22:21:12Z

cc @cloud-fan @Ngone51

SparkQA · 2020-05-29T03:26:02Z

Test build #123250 has finished for PR 28669 at commit 3ed0f63.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2020-05-29T04:30:04Z

This makes sense to me. e.g. for partitions [1, 1, 1, ..., 1, 10], which are coalesced to [9, 10], I don't think there are skewed partitions.

@JkSelf what do you think?

JkSelf · 2020-05-29T07:54:10Z

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewedJoin.scala

        if supportedJoinTypes.contains(joinType) =>
      assert(left.partitionsWithSizes.length == right.partitionsWithSizes.length)
      val numPartitions = left.partitionsWithSizes.length
      // We use the median size of the original shuffle partitions to detect skewed partitions.


We may also need to adjust the comment here.

JkSelf · 2020-05-29T07:59:03Z

Good improvement. Except one small comment. LGTM.

SparkQA · 2020-05-29T20:11:52Z

Test build #123293 has finished for PR 28669 at commit 10a468e.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2020-05-30T07:47:28Z

thanks, merging to master/3.0!

This PR makes a minor change in deciding whether a partition is skewed by comparing the partition size to the median size of coalesced partitions instead of median size of raw partitions before coalescing. This change is line with target size criteria of splitting skew join partitions and can also cope with situations of extra empty partitions caused by over-partitioning. This PR has also improved skew join tests in AQE tests. No. Updated UTs. Closes apache#28669 from maryannxue/spark-31864. Authored-by: Maryann Xue <[email protected]> Signed-off-by: Wenchen Fan <[email protected]> (cherry picked from commit b9737c3) Signed-off-by: Wenchen Fan <[email protected]>

This PR makes a minor change in deciding whether a partition is skewed by comparing the partition size to the median size of coalesced partitions instead of median size of raw partitions before coalescing. This change is line with target size criteria of splitting skew join partitions and can also cope with situations of extra empty partitions caused by over-partitioning. This PR has also improved skew join tests in AQE tests. No. Updated UTs. Closes #28669 from maryannxue/spark-31864. Authored-by: Maryann Xue <[email protected]> Signed-off-by: Wenchen Fan <[email protected]> (cherry picked from commit b9737c3) Signed-off-by: Wenchen Fan <[email protected]>

manuzhang · 2020-06-07T15:24:31Z

@cloud-fan @maryannxue @JkSelf I'm seeing a case where partitions [0,0,0,...,13GB] were coalesced to [13GB] and took 17 min for a SortMergeJoin. With coalescing disabled, partitions would be split into [0,0,0,..., 256MB, 256MB,...,256MB] by OptimizeSkewedJoin and only took 38s. WDYT ?

cloud-fan · 2020-06-08T12:34:03Z

I think a single partition should be a special case. The median size doesn't make sense anymore and we should use the target partition size. Can you open a PR for it?

maryannxue · 2020-06-08T15:32:15Z

I assume there's one only key in that 13G partition. You can always run into this when the number of distinct keys is smaller than the partition number. It is not a good idea to use median whether it's pre-coalescing size or post-coalescing size. Suppose, you have 100 keys, and the partition number is set to 200, even with pre-coalescing size, you won't get the skew handling you want, but with partition number being 201 you can. The skew behavior now depends on the partition number. However, it's probably the best we can do for now given we don't have a more advanced strategy here.
With something like 13G, we should probably always split it.

So think we can:

revert this PR, so for extremely clustered keys, skew can work
change the spark.sql.adaptive.skewJoin.skewedPartitionFactor check condition to ">= 0" instead of ">0" so we can force split large partitions even if there isn't an uneven distribution.

manuzhang · 2020-06-08T23:11:41Z

@cloud-fan
I think there are two cases here. The first is only one big partition before coalescing. I don't think it fits into skew join optimization but more general splitting-big-partition optimization. The second is, like in my example, skew before coalescing but no skew afterwards.

Do you want to target both cases or just the second ?

@maryannxue
In terms of performance, I don't think there will be a big difference between the case of "not skew join" 200 and optimizing "little skew join" 201. However, as in my case, the difference of optimizing "very skew join" before coalescing and "not skew join" after coalescing would be significant.

I prefer option 1 since we can have a more general configuration that works good enough.

cloud-fan · 2020-06-09T05:49:42Z

I think the current status for 3.0 is good enough, as it's conservative to trigger the skew join optimization.

@manuzhang can you send a PR to master branch to do the revert? We can still keep the test change though.

manuzhang · 2020-06-09T11:54:17Z

@cloud-fan #28770 has been created. Please help review.

maryannxue · 2020-06-09T15:37:32Z

@manuzhang
I was actually suggesting both :)

manuzhang · 2020-06-09T15:57:17Z

@maryannxue spark.sql.adaptive.skewJoin.skewedPartitionFactor=0 doesn't look meaningful to me. Even spark.sql.adaptive.skewJoin.skewedPartitionFactor=1 feels counter-intuitive as we are optimizing skew joins when there aren't any skewness.

JkSelf · 2020-06-10T07:15:39Z

@manuzhang If you adjust the configuration of spark.sql.adaptive.skewJoin.skewedPartitionFactor between 0 and 1, can the 13GB partition be split? If so, can we set the skewed related configurations to trigger the skew optimization not revert this PR? I think this PR is meaningful when the rule of OptimizeSkewedJoin is behind CoalesceShufflePartitions in the queryStageOptimizerRules of AdaptiveSparkPlanExec.

manuzhang · 2020-06-10T07:35:49Z

@JkSelf That will work but the value doesn't make sense to me. In addition, I don't want to tune the configuration for every specific case but come up with a suite of default configurations that work out of box for most cases.

I think this PR is meaningful when the rule of OptimizeSkewedJoin is behind CoalesceShufflePartitions in the queryStageOptimizerRules of AdaptiveSparkPlanExec

I don't see too much "meaning" here as we split the skewed partition towards the average size of coalesced non-skew partitions or advisory target size whichever is larger. We'll end up with evenly distributed partitions with or without this PR. However, we lose the chance to optimize skew partitions in some cases with this change.

… condition ### What changes were proposed in this pull request? This reverts commit b9737c3 while keeping following changes * set default value of `spark.sql.adaptive.skewJoin.skewedPartitionFactor` to 5 * improve tests * remove unused imports ### Why are the changes needed? As discussed in #28669 (comment), revert SPARK-31864 for optimizing skew join to work for extremely clustered keys. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing tests. Closes #28770 from manuzhang/spark-31942. Authored-by: manuzhang <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

[SPARK-31864][SQL] Adjust AQE skew join trigger condition

3ed0f63

probot-autolabeler bot added the SQL label May 28, 2020

JkSelf reviewed May 29, 2020

View reviewed changes

address review comments

10a468e

cloud-fan closed this in b9737c3 May 30, 2020

manuzhang mentioned this pull request Jun 9, 2020

[SPARK-31942] Partially revert "[SPARK-31864][SQL] Adjust AQE skew join trigger condition #28770

Closed

[SPARK-31864][SQL] Adjust AQE skew join trigger condition #28669

[SPARK-31864][SQL] Adjust AQE skew join trigger condition #28669

Uh oh!

Conversation

maryannxue commented May 28, 2020

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

maryannxue commented May 28, 2020

Uh oh!

SparkQA commented May 29, 2020

Uh oh!

cloud-fan commented May 29, 2020

Uh oh!

JkSelf May 29, 2020

Choose a reason for hiding this comment

Uh oh!

JkSelf commented May 29, 2020

Uh oh!

SparkQA commented May 29, 2020

Uh oh!

cloud-fan commented May 30, 2020

Uh oh!

manuzhang commented Jun 7, 2020

Uh oh!

cloud-fan commented Jun 8, 2020

Uh oh!

maryannxue commented Jun 8, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

manuzhang commented Jun 8, 2020

Uh oh!

cloud-fan commented Jun 9, 2020

Uh oh!

manuzhang commented Jun 9, 2020

Uh oh!

maryannxue commented Jun 9, 2020

Uh oh!

manuzhang commented Jun 9, 2020

Uh oh!

JkSelf commented Jun 10, 2020

Uh oh!

manuzhang commented Jun 10, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

maryannxue commented Jun 8, 2020 •

edited

Loading