This is an automated email from the ASF dual-hosted git repository. dongjoon pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push: new b174aaa42e6 [SPARK-44021][SQL] Add spark.sql.files.maxPartitionNum b174aaa42e6 is described below commit b174aaa42e6d405a1d157534eec9623436c3232b Author: Yuming Wang <yumw...@ebay.com> AuthorDate: Tue Jun 13 18:13:24 2023 -0700 [SPARK-44021][SQL] Add spark.sql.files.maxPartitionNum ### What changes were proposed in this pull request? This PR add a new SQL config: `spark.sql.files.maxPartitionNum`. User can set it to avoid generating too many partitions when reading file-based sources. Too many partitions will increase the various overheads of the driver and cause Shuffle service OOM. The following is the GC log of the Shuffle service: ``` 2023-06-08T01:41:01.871-0700: 7303.965: [Full GC (Allocation Failure) 2023-06-08T01:41:01.871-0700: 7303.965: [CMS: 4194304K->4194304K(4194304K), 7.4010107 secs]2023-06-08T01:41:09.272-0700: 7311.366: [Class Histogram (after full gc): num #instances #bytes class name ---------------------------------------------- 1: 7110660 2334927400 [C 2: 19465810 467514416 [I 3: 6754570 270182800 org.apache.spark.network.protocol.ChunkFetchRequest 4: 6661155 266446200 org.sparkproject.io.netty.channel.DefaultChannelPromise 5: 6639056 265562240 org.apache.spark.network.buffer.FileSegmentManagedBuffer 6: 6639055 265562200 org.apache.spark.network.protocol.RequestTraceInfo 7: 6663764 213240448 org.sparkproject.io.netty.util.Recycler$DefaultHandle 8: 6659382 213100224 org.sparkproject.io.netty.channel.AbstractChannelHandlerContext$WriteTask 9: 6659218 213094976 org.apache.spark.network.server.ChunkFetchRequestHandler$$Lambda$156/886274988 10: 6640444 212494208 java.io.File ... ``` ### Why are the changes needed? 1. The PR aims to selectively rescale large RDDs only, while keeping the existing behavior in the same way in small RDDs. So directly increasing `spark.sql.files.maxPartitionBytes` is not acceptable: 1. There are multiple data sources in one SQL, setting `spark.sql.files.maxPartitionBytes` will affect all data sources. 2. We don't know how much `spark.sql.files.maxPartitionBytes` should be set to, sometimes it may be very large(More than 20GiB). 2. To make it do not generate too many partitions if it is very large partitioned and bucketed table as it is not always use bucket scan since [SPARK-32859](https://issues.apache.org/jira/browse/SPARK-32859). Before SPARK-32859 | After SPARK-32859 --- | --- <img width="400" src="https://github.com/apache/spark/assets/5399861/5e14932b-aa3d-4b14-b80c-e3ff348958c4"> | <img width="400" src="https://github.com/apache/spark/assets/5399861/170311a0-c086-408a-9d95-17031e21e16a"> 3. Avoid generating too many partitions if these are lots of small files. ### Does this PR introduce _any_ user-facing change? No. Unless the user sets `spark.sql.files.maxPartitionNum`. ### How was this patch tested? Unit test and manual testing: Before this PR | After this PR and `set spark.sql.files.maxPartitionNum=20000` -- | -- <img width="400" src="https://github.com/apache/spark/assets/5399861/ffda1850-cd4a-4970-a4e5-e1e43177135a"> | <img width="330" src="https://github.com/apache/spark/assets/5399861/1df7cac7-fe82-4af3-b3ec-91aa23c79a8b"> Closes #41545 from wangyum/SPARK-44021. Authored-by: Yuming Wang <yumw...@ebay.com> Signed-off-by: Dongjoon Hyun <dongj...@apache.org> --- docs/sql-performance-tuning.md | 11 ++++++++ .../org/apache/spark/sql/internal/SQLConf.scala | 12 +++++++++ .../sql/execution/datasources/FilePartition.scala | 31 +++++++++++++++++++--- .../datasources/FileSourceStrategySuite.scala | 23 ++++++++++++++++ 4 files changed, 73 insertions(+), 4 deletions(-) diff --git a/docs/sql-performance-tuning.md b/docs/sql-performance-tuning.md index 36b02d122bd..1467409bb50 100644 --- a/docs/sql-performance-tuning.md +++ b/docs/sql-performance-tuning.md @@ -95,6 +95,17 @@ that these options will be deprecated in future release as more optimizations ar </td> <td>3.1.0</td> </tr> + <tr> + <td><code>spark.sql.files.maxPartitionNum</code></td> + <td>None</td> + <td> + The suggested (not guaranteed) maximum number of split file partitions. If it is set, + Spark will rescale each partition to make the number of partitions is close to this + value if the initial number of partitions exceeds this value. This configuration is + effective only when using file-based sources such as Parquet, JSON and ORC. + </td> + <td>3.5.0</td> + </tr> <tr> <td><code>spark.sql.broadcastTimeout</code></td> <td>300</td> diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index 47b8474953b..2fb3f17b725 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -1749,6 +1749,16 @@ object SQLConf { .checkValue(v => v > 0, "The min partition number must be a positive integer.") .createOptional + val FILES_MAX_PARTITION_NUM = buildConf("spark.sql.files.maxPartitionNum") + .doc("The suggested (not guaranteed) maximum number of split file partitions. If it is set, " + + "Spark will rescale each partition to make the number of partitions is close to this " + + "value if the initial number of partitions exceeds this value. This configuration is " + + "effective only when using file-based sources such as Parquet, JSON and ORC.") + .version("3.5.0") + .intConf + .checkValue(v => v > 0, "The maximum number of partitions must be a positive integer.") + .createOptional + val IGNORE_CORRUPT_FILES = buildConf("spark.sql.files.ignoreCorruptFiles") .doc("Whether to ignore corrupt files. If true, the Spark jobs will continue to run when " + "encountering corrupted files and the contents that have been read will still be returned. " + @@ -4509,6 +4519,8 @@ class SQLConf extends Serializable with Logging { def filesMinPartitionNum: Option[Int] = getConf(FILES_MIN_PARTITION_NUM) + def filesMaxPartitionNum: Option[Int] = getConf(FILES_MAX_PARTITION_NUM) + def ignoreCorruptFiles: Boolean = getConf(IGNORE_CORRUPT_FILES) def ignoreMissingFiles: Boolean = getConf(IGNORE_MISSING_FILES) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala index a4d16a0fd2b..8169652c6f0 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala @@ -18,11 +18,13 @@ package org.apache.spark.sql.execution.datasources import scala.collection.mutable import scala.collection.mutable.ArrayBuffer +import scala.math.BigDecimal.RoundingMode import org.apache.spark.Partition import org.apache.spark.internal.Logging import org.apache.spark.sql.SparkSession import org.apache.spark.sql.connector.read.InputPartition +import org.apache.spark.sql.internal.SQLConf /** * A collection of file blocks that should be read as a single task @@ -50,10 +52,10 @@ case class FilePartition(index: Int, files: Array[PartitionedFile]) object FilePartition extends Logging { - def getFilePartitions( - sparkSession: SparkSession, + private def getFilePartitions( partitionedFiles: Seq[PartitionedFile], - maxSplitBytes: Long): Seq[FilePartition] = { + maxSplitBytes: Long, + openCostInBytes: Long): Seq[FilePartition] = { val partitions = new ArrayBuffer[FilePartition] val currentFiles = new ArrayBuffer[PartitionedFile] var currentSize = 0L @@ -69,7 +71,6 @@ object FilePartition extends Logging { currentSize = 0 } - val openCostInBytes = sparkSession.sessionState.conf.filesOpenCostInBytes // Assign files to partitions using "Next Fit Decreasing" partitionedFiles.foreach { file => if (currentSize + file.length > maxSplitBytes) { @@ -83,6 +84,28 @@ object FilePartition extends Logging { partitions.toSeq } + def getFilePartitions( + sparkSession: SparkSession, + partitionedFiles: Seq[PartitionedFile], + maxSplitBytes: Long): Seq[FilePartition] = { + val openCostBytes = sparkSession.sessionState.conf.filesOpenCostInBytes + val maxPartitionNum = sparkSession.sessionState.conf.filesMaxPartitionNum + val partitions = getFilePartitions(partitionedFiles, maxSplitBytes, openCostBytes) + if (maxPartitionNum.exists(partitions.size > _)) { + val totalSizeInBytes = + partitionedFiles.map(_.length + openCostBytes).map(BigDecimal(_)).sum[BigDecimal] + val desiredSplitBytes = + (totalSizeInBytes / BigDecimal(maxPartitionNum.get)).setScale(0, RoundingMode.UP).longValue + val desiredPartitions = getFilePartitions(partitionedFiles, desiredSplitBytes, openCostBytes) + logWarning(s"The number of partitions is ${partitions.size}, which exceeds the maximum " + + s"number configured: $maxPartitionNum. Spark rescales it to ${desiredPartitions.size} " + + s"by ignoring the configuration of ${SQLConf.FILES_MAX_PARTITION_BYTES.key}.") + desiredPartitions + } else { + partitions + } + } + def maxSplitBytes( sparkSession: SparkSession, selectedPartitions: Seq[PartitionDirectory]): Long = { diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala index 26655c2d95a..91182f6473d 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/FileSourceStrategySuite.scala @@ -602,6 +602,29 @@ class FileSourceStrategySuite extends QueryTest with SharedSparkSession { checkDataFilters(Set.empty) } + test(s"SPARK-44021: Test ${SQLConf.FILES_MAX_PARTITION_NUM.key} works as expected") { + val files = + Range(0, 300000).map(p => PartitionedFile(InternalRow.empty, sp(s"$p"), 0, 50000000)) + val maxPartitionBytes = conf.filesMaxPartitionBytes + val defaultPartitions = FilePartition.getFilePartitions(spark, files, maxPartitionBytes) + assert(defaultPartitions.size === 150000) + + withSQLConf(SQLConf.FILES_MAX_PARTITION_NUM.key -> "20000") { + val partitions = FilePartition.getFilePartitions(spark, files, maxPartitionBytes) + assert(partitions.size === 20000) + } + + withSQLConf(SQLConf.FILES_MAX_PARTITION_NUM.key -> "50000") { + val partitions = FilePartition.getFilePartitions(spark, files, maxPartitionBytes) + assert(partitions.size === 50000) + } + + withSQLConf(SQLConf.FILES_MAX_PARTITION_NUM.key -> "200000") { + val partitions = FilePartition.getFilePartitions(spark, files, maxPartitionBytes) + assert(partitions.size === defaultPartitions.size) + } + } + // Helpers for checking the arguments passed to the FileFormat. protected val checkPartitionSchema = --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org