You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "wForget (via GitHub)" <gi...@apache.org> on 2023/05/31 13:18:28 UTC
[GitHub] [spark] wForget opened a new pull request, #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
wForget opened a new pull request, #41407:
URL: https://github.com/apache/spark/pull/41407
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
+ move the rule OptimizeSkewInRebalancePartitions from stage optimization phase to stage preparation phase.
+ run the rule EnsureRequirements one more time after the OptimizeSkewInRebalancePartitions rule in the stage preparation phase.
+ consider skewed AQEShuffleReadExec in SimpleCostEvaluator
+ add new config to decide if force optimize skewed partitions
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
In some cases, the aggregation operation has serious data skew, and I hope to insert Rebalance before the aggregation operation to avoid skew.
A common case (data skew in RebalancePartition):
```
HashAggregate(keys=[c1#226], functions=[count(1)], output=[c1#226, count(1)#231L])
+- *(2) HashAggregate(keys=[c1#226], functions=[partial_count(1)], output=[c1#226, count#235L])
+- AQEShuffleRead coalesced
+- ShuffleQueryStage 0
+- Exchange hashpartitioning(c1#226, 5), REBALANCE_PARTITIONS_BY_COL, [plan_id=106]
+- *(1) Project [key#221 AS c1#226]
+- *(1) SerializeFromObject [knownnotnull(assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$TestData, true])).key AS key#221]
+- Scan[obj#220]
```
and after this PR, the plan looks like:
```
HashAggregate(keys=[c1#226], functions=[count(1)], output=[c1#226, count(1)#231L])
+- AQEShuffleRead coalesced
+- ShuffleQueryStage 1
+- Exchange hashpartitioning(c1#226, 5), ENSURE_REQUIREMENTS, [plan_id=140]
+- *(2) HashAggregate(keys=[c1#226], functions=[partial_count(1)], output=[c1#226, count#235L])
+- AQEShuffleRead coalesced and skewed
+- ShuffleQueryStage 0
+- Exchange hashpartitioning(c1#226, 5), REBALANCE_PARTITIONS_BY_COL, [plan_id=106]
+- *(1) Project [key#221 AS c1#226]
+- *(1) SerializeFromObject [knownnotnull(assertnotnull(input[0, org.apache.spark.sql.test.SQLTestData$TestData, true])).key AS key#221]
+- Scan[obj#220]
```
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
Yes, a new config.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
unit test
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #41407:
URL: https://github.com/apache/spark/pull/41407#discussion_r1217406788
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala:
##########
@@ -104,7 +104,10 @@ case class AdaptiveSparkPlanExec(
@transient private val costEvaluator =
conf.getConf(SQLConf.ADAPTIVE_CUSTOM_COST_EVALUATOR_CLASS) match {
case Some(className) => CostEvaluator.instantiate(className, session.sparkContext.getConf)
- case _ => SimpleCostEvaluator(conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN))
+ case _ =>
+ val forceOptimizeSkewed = conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN) ||
+ conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_IN_REBALANCE_PARTITIONS)
Review Comment:
it seems fragile, what happens if user enable one and disable the other ? the result behavior is uncontrolled.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wForget commented on a diff in pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #41407:
URL: https://github.com/apache/spark/pull/41407#discussion_r1218791056
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/simpleCosting.scala:
##########
@@ -36,22 +36,26 @@ case class SimpleCost(value: Long) extends Cost {
}
/**
- * A skew join aware implementation of [[CostEvaluator]], which counts the number of
- * [[ShuffleExchangeLike]] nodes and skew join nodes in the plan.
+ * A skew aware implementation of [[CostEvaluator]], which counts the number of
+ * skew join nodes and skew [[AQEShuffleReadExec]] nodes in the plan.
*/
-case class SimpleCostEvaluator(forceOptimizeSkewedJoin: Boolean) extends CostEvaluator {
+case class SimpleCostEvaluator(
+ forceOptimizeSkewedJoin: Boolean,
+ forceOptimizeSkewedShuffleRead: Boolean)
+ extends CostEvaluator {
override def evaluateCost(plan: SparkPlan): Cost = {
val numShuffles = plan.collect {
case s: ShuffleExchangeLike => s
}.size
- if (forceOptimizeSkewedJoin) {
- val numSkewJoins = plan.collect {
- case j: ShuffledJoin if j.isSkewJoin => j
+ if (forceOptimizeSkewedJoin || forceOptimizeSkewedShuffleRead) {
+ val numSkews = plan.collect {
+ case j: ShuffledJoin if forceOptimizeSkewedJoin && j.isSkewJoin => j
+ case r: AQEShuffleReadExec if forceOptimizeSkewedShuffleRead && r.hasSkewedPartition => r
Review Comment:
Makes sense, I will add relevant test cases
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wForget commented on a diff in pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #41407:
URL: https://github.com/apache/spark/pull/41407#discussion_r1218765788
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala:
##########
@@ -104,7 +104,10 @@ case class AdaptiveSparkPlanExec(
@transient private val costEvaluator =
conf.getConf(SQLConf.ADAPTIVE_CUSTOM_COST_EVALUATOR_CLASS) match {
case Some(className) => CostEvaluator.instantiate(className, session.sparkContext.getConf)
- case _ => SimpleCostEvaluator(conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN))
+ case _ =>
+ val forceOptimizeSkewed = conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN) ||
+ conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_IN_REBALANCE_PARTITIONS)
Review Comment:
I have made it a new argument in SimpleCostEvaluator.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wForget commented on pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on PR #41407:
URL: https://github.com/apache/spark/pull/41407#issuecomment-1571297406
`Rebalance(groupingExpressions + constant)` achieves the same effect.
```
SELECT c1, count(1) FROM (SELECT /*+ REBALANCE(c1, n) */ c1, 1 as n FROM v) t group by c1
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jaceklaskowski commented on a diff in pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "jaceklaskowski (via GitHub)" <gi...@apache.org>.
jaceklaskowski commented on code in PR #41407:
URL: https://github.com/apache/spark/pull/41407#discussion_r1229261305
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewInRebalancePartitions.scala:
##########
@@ -92,9 +97,33 @@ object OptimizeSkewInRebalancePartitions extends AQEShuffleReadRule {
return plan
}
- plan transformUp {
- case stage: ShuffleQueryStageExec if isSupported(stage.shuffle) =>
+ val optimized = plan transformUp {
+ case stage: ShuffleQueryStageExec if stage.isMaterialized &&
+ stage.mapStats.isDefined && isSupported(stage.shuffle) =>
tryOptimizeSkewedPartitions(stage)
}
+ val requirementSatisfied = if (ensureRequirements.requiredDistribution.isDefined) {
+ ValidateRequirements.validate(optimized, ensureRequirements.requiredDistribution.get)
+ } else {
+ ValidateRequirements.validate(optimized)
+ }
+ if (requirementSatisfied) {
+ optimized.transform {
+ case RebalanceChildWrapper(child) => child
+ }
+ } else if (conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS)) {
Review Comment:
Can I ask for a `SQLConf.isAdaptiveForceOptimize...`?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewInRebalancePartitions.scala:
##########
@@ -92,9 +97,33 @@ object OptimizeSkewInRebalancePartitions extends AQEShuffleReadRule {
return plan
}
- plan transformUp {
- case stage: ShuffleQueryStageExec if isSupported(stage.shuffle) =>
+ val optimized = plan transformUp {
+ case stage: ShuffleQueryStageExec if stage.isMaterialized &&
+ stage.mapStats.isDefined && isSupported(stage.shuffle) =>
tryOptimizeSkewedPartitions(stage)
}
+ val requirementSatisfied = if (ensureRequirements.requiredDistribution.isDefined) {
+ ValidateRequirements.validate(optimized, ensureRequirements.requiredDistribution.get)
+ } else {
+ ValidateRequirements.validate(optimized)
+ }
+ if (requirementSatisfied) {
+ optimized.transform {
+ case RebalanceChildWrapper(child) => child
+ }
+ } else if (conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS)) {
+ ensureRequirements.apply(optimized).transform {
+ case RebalanceChildWrapper(child) => child
+ }
+ } else {
+ plan
+ }
}
}
+
+case class RebalanceChildWrapper(plan: SparkPlan) extends LeafExecNode {
Review Comment:
It'd be so helpful to have some description of this new class. I'd start by answering the question of why this class is really required 🙏
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewInRebalancePartitions.scala:
##########
@@ -92,9 +97,33 @@ object OptimizeSkewInRebalancePartitions extends AQEShuffleReadRule {
return plan
}
- plan transformUp {
- case stage: ShuffleQueryStageExec if isSupported(stage.shuffle) =>
+ val optimized = plan transformUp {
+ case stage: ShuffleQueryStageExec if stage.isMaterialized &&
+ stage.mapStats.isDefined && isSupported(stage.shuffle) =>
tryOptimizeSkewedPartitions(stage)
}
+ val requirementSatisfied = if (ensureRequirements.requiredDistribution.isDefined) {
+ ValidateRequirements.validate(optimized, ensureRequirements.requiredDistribution.get)
+ } else {
+ ValidateRequirements.validate(optimized)
+ }
Review Comment:
Had hard times to understand this change and thought I've got mine that could make it easier for future readers 😉
```scala
val requirementSatisfied = ensureRequirements.requiredDistribution.map { rd =>
ValidateRequirements.validate(optimized, rd)
}.getOrElse {
ValidateRequirements.validate(optimized)
}
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -823,6 +823,14 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val ADAPTIVE_FORCE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS =
+ buildConf("spark.sql.adaptive.forceOptimizeSkewsInRebalancePartitions")
+ .doc("When true, force enable OptimizeSkewsInRebalancePartitions" +
Review Comment:
Think `Enables OptimizeSkewsInRebalancePartitions even if it introduces an extra shuffle` is equally forceful yes less wordy due to `Enables...even if` (that implies some force).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #41407:
URL: https://github.com/apache/spark/pull/41407#discussion_r1218787943
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewInRebalancePartitions.scala:
##########
@@ -92,9 +93,21 @@ object OptimizeSkewInRebalancePartitions extends AQEShuffleReadRule {
return plan
}
- plan transformUp {
+ val optimized = plan transformUp {
case stage: ShuffleQueryStageExec if isSupported(stage.shuffle) =>
Review Comment:
since we move the rule to prepare execution, we should make sure the stage is materialized and fast return if it's not materialized.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/simpleCosting.scala:
##########
@@ -36,22 +36,26 @@ case class SimpleCost(value: Long) extends Cost {
}
/**
- * A skew join aware implementation of [[CostEvaluator]], which counts the number of
- * [[ShuffleExchangeLike]] nodes and skew join nodes in the plan.
+ * A skew aware implementation of [[CostEvaluator]], which counts the number of
+ * skew join nodes and skew [[AQEShuffleReadExec]] nodes in the plan.
*/
-case class SimpleCostEvaluator(forceOptimizeSkewedJoin: Boolean) extends CostEvaluator {
+case class SimpleCostEvaluator(
+ forceOptimizeSkewedJoin: Boolean,
+ forceOptimizeSkewedShuffleRead: Boolean)
+ extends CostEvaluator {
override def evaluateCost(plan: SparkPlan): Cost = {
val numShuffles = plan.collect {
case s: ShuffleExchangeLike => s
}.size
- if (forceOptimizeSkewedJoin) {
- val numSkewJoins = plan.collect {
- case j: ShuffledJoin if j.isSkewJoin => j
+ if (forceOptimizeSkewedJoin || forceOptimizeSkewedShuffleRead) {
+ val numSkews = plan.collect {
+ case j: ShuffledJoin if forceOptimizeSkewedJoin && j.isSkewJoin => j
+ case r: AQEShuffleReadExec if forceOptimizeSkewedShuffleRead && r.hasSkewedPartition => r
Review Comment:
if ShuffledJoin is skewed, it should always have AQEShuffleReadExec with skewed partition, right ? Then we will go into this pattern even if people disable force optimize skewed join..
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wForget commented on pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on PR #41407:
URL: https://github.com/apache/spark/pull/41407#issuecomment-1571392124
> Rebalance(groupingExpressions + constant) achieves the same effect.
It can make OptimizeSkewInRebalancePartitions effective, but will introduce additional shuffle when there is no data skew. So I think this fix will still bring benefits.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wForget commented on pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on PR #41407:
URL: https://github.com/apache/spark/pull/41407#issuecomment-1571389874
>
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wForget commented on pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on PR #41407:
URL: https://github.com/apache/spark/pull/41407#issuecomment-1581796572
> I'm still a bit concern about the use case of when foce optimizing skewed partition in rebalance that if it really has benefits, i.e, one more shuffle vs. map side data skew with aggregation. cc @cloud-fan @viirya if you have other thoughts.
Thanks @ulysses-you for your patient review, in my case it did bring good results.
before:
after:
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wForget commented on a diff in pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #41407:
URL: https://github.com/apache/spark/pull/41407#discussion_r1230371929
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/OptimizeSkewInRebalancePartitions.scala:
##########
@@ -92,9 +97,33 @@ object OptimizeSkewInRebalancePartitions extends AQEShuffleReadRule {
return plan
}
- plan transformUp {
- case stage: ShuffleQueryStageExec if isSupported(stage.shuffle) =>
+ val optimized = plan transformUp {
+ case stage: ShuffleQueryStageExec if stage.isMaterialized &&
+ stage.mapStats.isDefined && isSupported(stage.shuffle) =>
tryOptimizeSkewedPartitions(stage)
}
+ val requirementSatisfied = if (ensureRequirements.requiredDistribution.isDefined) {
+ ValidateRequirements.validate(optimized, ensureRequirements.requiredDistribution.get)
+ } else {
+ ValidateRequirements.validate(optimized)
+ }
+ if (requirementSatisfied) {
+ optimized.transform {
+ case RebalanceChildWrapper(child) => child
+ }
+ } else if (conf.getConf(SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS)) {
Review Comment:
I want to be consistent with `SQLConf#ADAPTIVE_FORCE_OPTIMIZE_SKEWED_JOIN`, can I add it in a follow-up PR if needed?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wForget commented on pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on PR #41407:
URL: https://github.com/apache/spark/pull/41407#issuecomment-1588355074
@cloud-fan @viirya Could you please take a look?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wForget commented on a diff in pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on code in PR #41407:
URL: https://github.com/apache/spark/pull/41407#discussion_r1221016028
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/simpleCosting.scala:
##########
@@ -36,22 +36,26 @@ case class SimpleCost(value: Long) extends Cost {
}
/**
- * A skew join aware implementation of [[CostEvaluator]], which counts the number of
- * [[ShuffleExchangeLike]] nodes and skew join nodes in the plan.
+ * A skew aware implementation of [[CostEvaluator]], which counts the number of
+ * skew join nodes and skew [[AQEShuffleReadExec]] nodes in the plan.
*/
-case class SimpleCostEvaluator(forceOptimizeSkewedJoin: Boolean) extends CostEvaluator {
+case class SimpleCostEvaluator(
+ forceOptimizeSkewedJoin: Boolean,
+ forceOptimizeSkewedShuffleRead: Boolean)
+ extends CostEvaluator {
override def evaluateCost(plan: SparkPlan): Cost = {
val numShuffles = plan.collect {
case s: ShuffleExchangeLike => s
}.size
- if (forceOptimizeSkewedJoin) {
- val numSkewJoins = plan.collect {
- case j: ShuffledJoin if j.isSkewJoin => j
+ if (forceOptimizeSkewedJoin || forceOptimizeSkewedShuffleRead) {
+ val numSkews = plan.collect {
+ case j: ShuffledJoin if forceOptimizeSkewedJoin && j.isSkewJoin => j
+ case r: AQEShuffleReadExec if forceOptimizeSkewedShuffleRead && r.hasSkewedPartition => r
Review Comment:
I made the fix, but couldn't think of a suitable test case to cover it.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] github-actions[bot] commented on pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #41407:
URL: https://github.com/apache/spark/pull/41407#issuecomment-1732714193
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #41407:
URL: https://github.com/apache/spark/pull/41407#issuecomment-1571197826
I think the more simpler way is to inject a repartition hint for this case, or do rebalance on a different key with gourp by. Optimize skew can not handle some cases which keys have high compression ratio.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wForget commented on pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on PR #41407:
URL: https://github.com/apache/spark/pull/41407#issuecomment-1571387439
> I think the more simpler way is to inject a repartition hint for this case, or do rebalance on a different key with gourp by.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] github-actions[bot] closed pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
URL: https://github.com/apache/spark/pull/41407
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wForget commented on pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on PR #41407:
URL: https://github.com/apache/spark/pull/41407#issuecomment-1571228458
> I think the more simpler way is to inject a repartition hint for this case,
There seems to be data skew, although relatively minor.
> or do rebalance on a different key with gourp by.
This requires analyzing the skewed key and rebalance on it.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] wForget commented on pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "wForget (via GitHub)" <gi...@apache.org>.
wForget commented on PR #41407:
URL: https://github.com/apache/spark/pull/41407#issuecomment-1571217222
> I think the more simpler way is to inject a repartition hint for this case, or do rebalance on a different key with gourp by. Optimize skew can not handle some cases which keys have high compression ratio.
Thanks @ulysses-you ,it sounds simpler, is this change still needed?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on a diff in pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on code in PR #41407:
URL: https://github.com/apache/spark/pull/41407#discussion_r1222359355
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/simpleCosting.scala:
##########
@@ -36,22 +36,37 @@ case class SimpleCost(value: Long) extends Cost {
}
/**
- * A skew join aware implementation of [[CostEvaluator]], which counts the number of
- * [[ShuffleExchangeLike]] nodes and skew join nodes in the plan.
+ * A skew aware implementation of [[CostEvaluator]], which counts the number of
+ * skew join nodes and skew [[AQEShuffleReadExec]] nodes in the plan.
*/
-case class SimpleCostEvaluator(forceOptimizeSkewedJoin: Boolean) extends CostEvaluator {
+case class SimpleCostEvaluator(
+ forceOptimizeSkewedJoin: Boolean,
+ forceOptimizeSkewsInRebalancePartitions: Boolean)
+ extends CostEvaluator {
Review Comment:
nit: 2 indentation
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -2841,6 +2841,55 @@ class AdaptiveQueryExecSuite
}
}
}
+
+ test("SPARK-35725: Support optimize skewed partitions even if introduce extra shuffle") {
+ withTempView("v") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+ SQLConf.COALESCE_PARTITIONS_ENABLED.key -> "true",
+ SQLConf.ADAPTIVE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS_ENABLED.key -> "true",
+ SQLConf.ADAPTIVE_FORCE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "5",
+ SQLConf.COALESCE_PARTITIONS_MIN_PARTITION_NUM.key -> "1") {
+
+ spark.sparkContext.parallelize(
+ (1 to 10).map(i => TestData(if (i > 4) 5 else i, i.toString)), 3)
+ .toDF("c1", "c2").createOrReplaceTempView("v")
+
+ def checkPartitionNumber(
+ query: String, skewedPartitionNumber: Int, totalNumber: Int): Unit = {
Review Comment:
`skewedPartitionNumber` -> `numSkewedPartition`, `totalNumber` -> `numPartition`
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] ulysses-you commented on pull request #41407: [SPARK-43900][SQL] Support optimize skewed partitions even if introduce extra shuffle
Posted by "ulysses-you (via GitHub)" <gi...@apache.org>.
ulysses-you commented on PR #41407:
URL: https://github.com/apache/spark/pull/41407#issuecomment-1581777331
I'm still a bit concern about the use case of when foce optimizing skewed partition in rebalance that if it really has benefits, i.e, one more shuffle vs. map side data skew with aggregation. cc @cloud-fan @viirya if you have other thoughts.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org