[GitHub] [spark] cloud-fan commented on a change in pull request #33172: [SPARK-35968][SQL] Make sure partitions are not too small in AQE partition coalescing
cloud-fan commented on a change in pull request #33172:
URL: https://github.com/apache/spark/pull/33172#discussion_r662102426
##
File path:
sql/core/src/test/scala/org/apache/spark/sql/execution/ShufflePartitionsUtilSuite.scala
##
@@ -459,9 +466,36 @@ class ShufflePartitionsUtilSuite extends SparkFunSuite {
Seq(
Some(specs1),
Some(specs2)),
-targetSize, 1)
+targetSize)
assert(coalesced == Seq(expected1, expected2))
}
+
+{
+ // No actual coalescing happened, return empty result.
Review comment:
not related to this PR. Just found we missed testing this case when
reading the code.
--
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-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #33172: [SPARK-35968][SQL] Make sure partitions are not too small in AQE partition coalescing
cloud-fan commented on a change in pull request #33172:
URL: https://github.com/apache/spark/pull/33172#discussion_r662152677
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ShufflePartitionsUtil.scala
##
@@ -61,7 +63,9 @@ object ShufflePartitionsUtil extends Logging {
// `maxTargetSize` from being set to 0.
val maxTargetSize = math.max(
math.ceil(totalPostShuffleInputSize / minNumPartitions.toDouble).toLong,
16)
-val targetSize = math.min(maxTargetSize, advisoryTargetSize)
+// The target size can be very small if minNumPartitions is super big.
Here we use
+// minPartitionSize as the lower bound to avoid too small partitions after
coalescing.
+val targetSize = math.max(math.min(maxTargetSize, advisoryTargetSize),
minPartitionSize)
val shuffleIds =
mapOutputStatistics.flatMap(_.map(_.shuffleId)).mkString(", ")
logInfo(s"For shuffle($shuffleIds), advisory target size:
$advisoryTargetSize, " +
Review comment:
sure
--
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-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #33172: [SPARK-35968][SQL] Make sure partitions are not too small in AQE partition coalescing
cloud-fan commented on a change in pull request #33172:
URL: https://github.com/apache/spark/pull/33172#discussion_r662151008
##
File path:
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##
@@ -525,18 +525,34 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ private val MIN_PARTITION_NUM_KEY =
"spark.sql.adaptive.coalescePartitions.minPartitionNum"
+ private val MIN_PARTITION_SIZE_KEY =
"spark.sql.adaptive.coalescePartitions.minPartitionSize"
+
val COALESCE_PARTITIONS_MIN_PARTITION_NUM =
-buildConf("spark.sql.adaptive.coalescePartitions.minPartitionNum")
+buildConf(MIN_PARTITION_NUM_KEY)
.doc("The suggested (not guaranteed) minimum number of shuffle
partitions after " +
"coalescing. If not set, the default value is the default parallelism
of the " +
-"Spark cluster. This configuration only has an effect when " +
-s"'${ADAPTIVE_EXECUTION_ENABLED.key}' and " +
+s"Spark cluster, w.r.t. the minimum partition size set by
'$MIN_PARTITION_SIZE_KEY'. " +
+s"This configuration only has an effect when
'${ADAPTIVE_EXECUTION_ENABLED.key}' and " +
s"'${COALESCE_PARTITIONS_ENABLED.key}' are both true.")
.version("3.0.0")
.intConf
.checkValue(_ > 0, "The minimum number of partitions must be positive.")
.createOptional
+ val COALESCE_PARTITIONS_MIN_PARTITION_SIZE =
+buildConf(MIN_PARTITION_SIZE_KEY)
+ .doc(s"The minimum size of shuffle partitions after coalescing if
'$MIN_PARTITION_NUM_KEY' " +
+s"is not set. When '$MIN_PARTITION_NUM_KEY' is not set, the minimum
number of " +
+"shuffle partitions after coalescing will fall back to the default
parallelism " +
+"of the Spark cluster, sometimes causing partition sizes to be too
small. This config " +
+"aims to prevent this situation. Note that this config does not take
effect if " +
+s"'${COALESCE_PARTITIONS_MIN_PARTITION_NUM.key}' is set by the user.")
+ .version("3.2.0")
+ .bytesConf(ByteUnit.BYTE)
+ .checkValue(_ > 0, "The minimum partition size must be positive.")
+ .createWithDefaultString("1MB")
Review comment:
We tried some values internally and 1MB gives the best perf in TPCDS. Of
curse, this should depend on cluster resources, and 1MB is good enough to avoid
the worse cases: many partitions of several KBs.
--
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-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #33172: [SPARK-35968][SQL] Make sure partitions are not too small in AQE partition coalescing
cloud-fan commented on a change in pull request #33172:
URL: https://github.com/apache/spark/pull/33172#discussion_r662151008
##
File path:
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##
@@ -525,18 +525,34 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ private val MIN_PARTITION_NUM_KEY =
"spark.sql.adaptive.coalescePartitions.minPartitionNum"
+ private val MIN_PARTITION_SIZE_KEY =
"spark.sql.adaptive.coalescePartitions.minPartitionSize"
+
val COALESCE_PARTITIONS_MIN_PARTITION_NUM =
-buildConf("spark.sql.adaptive.coalescePartitions.minPartitionNum")
+buildConf(MIN_PARTITION_NUM_KEY)
.doc("The suggested (not guaranteed) minimum number of shuffle
partitions after " +
"coalescing. If not set, the default value is the default parallelism
of the " +
-"Spark cluster. This configuration only has an effect when " +
-s"'${ADAPTIVE_EXECUTION_ENABLED.key}' and " +
+s"Spark cluster, w.r.t. the minimum partition size set by
'$MIN_PARTITION_SIZE_KEY'. " +
+s"This configuration only has an effect when
'${ADAPTIVE_EXECUTION_ENABLED.key}' and " +
s"'${COALESCE_PARTITIONS_ENABLED.key}' are both true.")
.version("3.0.0")
.intConf
.checkValue(_ > 0, "The minimum number of partitions must be positive.")
.createOptional
+ val COALESCE_PARTITIONS_MIN_PARTITION_SIZE =
+buildConf(MIN_PARTITION_SIZE_KEY)
+ .doc(s"The minimum size of shuffle partitions after coalescing if
'$MIN_PARTITION_NUM_KEY' " +
+s"is not set. When '$MIN_PARTITION_NUM_KEY' is not set, the minimum
number of " +
+"shuffle partitions after coalescing will fall back to the default
parallelism " +
+"of the Spark cluster, sometimes causing partition sizes to be too
small. This config " +
+"aims to prevent this situation. Note that this config does not take
effect if " +
+s"'${COALESCE_PARTITIONS_MIN_PARTITION_NUM.key}' is set by the user.")
+ .version("3.2.0")
+ .bytesConf(ByteUnit.BYTE)
+ .checkValue(_ > 0, "The minimum partition size must be positive.")
+ .createWithDefaultString("1MB")
Review comment:
We tried some values internally and 1MB gives the best perf in TPCDS. Of
curse, this should depend on cluster resources, and 1MB is good enough to avoid
the worse case: many partitions of several KBs.
--
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-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #33172: [SPARK-35968][SQL] Make sure partitions are not too small in AQE partition coalescing
cloud-fan commented on a change in pull request #33172:
URL: https://github.com/apache/spark/pull/33172#discussion_r662152464
##
File path:
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ShufflePartitionsUtil.scala
##
@@ -61,7 +63,9 @@ object ShufflePartitionsUtil extends Logging {
// `maxTargetSize` from being set to 0.
val maxTargetSize = math.max(
math.ceil(totalPostShuffleInputSize / minNumPartitions.toDouble).toLong,
16)
-val targetSize = math.min(maxTargetSize, advisoryTargetSize)
+// The target size can be very small if minNumPartitions is super big.
Here we use
+// minPartitionSize as the lower bound to avoid too small partitions after
coalescing.
+val targetSize = math.max(math.min(maxTargetSize, advisoryTargetSize),
minPartitionSize)
Review comment:
I don't have a strong opinion here. We can say that `minPartitionSize`
larger than `advisoryTargetSize` is an ill pattern and we should forbid, or we
just respect whatever user sets in for `minPartitionSize`
--
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-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #33172: [SPARK-35968][SQL] Make sure partitions are not too small in AQE partition coalescing
cloud-fan commented on a change in pull request #33172: URL: https://github.com/apache/spark/pull/33172#discussion_r662328867 ## File path: sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala ## @@ -1813,7 +1813,7 @@ class DataFrameSuite extends QueryTest (1 to 100).map(i => TestData2(i % 10, i))).toDF() // Distribute and order by. -val df4 = data.repartition($"a").sortWithinPartitions($"b".desc) +val df4 = data.repartition(5, $"a").sortWithinPartitions($"b".desc) Review comment: now partition coalescing applies here and then we are sorting one single partition, which fails this test. Adding the partition number of skip partition coalescing -- 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-unsubscr...@spark.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #33172: [SPARK-35968][SQL] Make sure partitions are not too small in AQE partition coalescing
cloud-fan commented on a change in pull request #33172: URL: https://github.com/apache/spark/pull/33172#discussion_r662328867 ## File path: sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala ## @@ -1813,7 +1813,7 @@ class DataFrameSuite extends QueryTest (1 to 100).map(i => TestData2(i % 10, i))).toDF() // Distribute and order by. -val df4 = data.repartition($"a").sortWithinPartitions($"b".desc) +val df4 = data.repartition(5, $"a").sortWithinPartitions($"b".desc) Review comment: now partition coalescing applies here and then we are sorting one single partition, which fails this test. Adding the partition number to skip partition coalescing -- 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-unsubscr...@spark.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #33172: [SPARK-35968][SQL] Make sure partitions are not too small in AQE partition coalescing
cloud-fan commented on a change in pull request #33172:
URL: https://github.com/apache/spark/pull/33172#discussion_r662362583
##
File path:
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##
@@ -525,8 +525,20 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val COALESCE_PARTITIONS_PARALLELISM_FIRST =
+buildConf("spark.sql.adaptive.coalescePartitions.parallelismFirst")
Review comment:
`ADVISORY_PARTITION_SIZE_IN_BYTES` doesn't mention
`COALESCE_PARTITIONS_MIN_PARTITION_NUM` either. I think it's fine to only refer
to "big" confs in the "small" confs, not from the other side.
--
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-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #33172: [SPARK-35968][SQL] Make sure partitions are not too small in AQE partition coalescing
cloud-fan commented on a change in pull request #33172:
URL: https://github.com/apache/spark/pull/33172#discussion_r662468622
##
File path:
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##
@@ -525,10 +525,35 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ private val MIN_PARTITION_SIZE_KEY =
"spark.sql.adaptive.coalescePartitions.minPartitionSize"
+
+ val COALESCE_PARTITIONS_PARALLELISM_FIRST =
+buildConf("spark.sql.adaptive.coalescePartitions.parallelismFirst")
+ .doc("When true, Spark ignores the target size specified by " +
+s"'${ADVISORY_PARTITION_SIZE_IN_BYTES.key}' (default 64MB) when
coalescing contiguous " +
+"shuffle partitions, and only respect the minimum partition size
specified by " +
+s"'$MIN_PARTITION_SIZE_KEY' (default 1MB), to maximize the
parallelism. " +
+"This is to avoid performance regression when enabling adaptive query
execution. " +
+"It's recommended to set this config to false and respect the target
size specified by " +
+s"'${ADVISORY_PARTITION_SIZE_IN_BYTES.key}'.")
Review comment:
@viirya is it clear to you now?
##
File path:
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
##
@@ -525,10 +525,35 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ private val MIN_PARTITION_SIZE_KEY =
"spark.sql.adaptive.coalescePartitions.minPartitionSize"
+
+ val COALESCE_PARTITIONS_PARALLELISM_FIRST =
+buildConf("spark.sql.adaptive.coalescePartitions.parallelismFirst")
+ .doc("When true, Spark ignores the target size specified by " +
+s"'${ADVISORY_PARTITION_SIZE_IN_BYTES.key}' (default 64MB) when
coalescing contiguous " +
+"shuffle partitions, and only respect the minimum partition size
specified by " +
+s"'$MIN_PARTITION_SIZE_KEY' (default 1MB), to maximize the
parallelism. " +
+"This is to avoid performance regression when enabling adaptive query
execution. " +
+"It's recommended to set this config to false and respect the target
size specified by " +
+s"'${ADVISORY_PARTITION_SIZE_IN_BYTES.key}'.")
Review comment:
@viirya is it clearer to you now?
--
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-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
