You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "marin-ma (via GitHub)" <gi...@apache.org> on 2025/11/13 17:24:37 UTC
[PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
marin-ma opened a new pull request, #53040:
URL: https://github.com/apache/spark/pull/53040
<!--
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
'common/utils/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.
-->
Currently, when Spark partitions the input files in a table scan, it first sorts the input splits, then adjacent splits are coalesced into a single partition. If the input split size distribution is uneven, some partitions will have only a few splits while others will have many.
We observed that this file partitioning strategy can slow down the reading process if some tasks are reading more files than others, especially in Gluten’s native Parquet reader. To address this performance issue, we are proposing a new partitioning strategy that takes both partition size and file count into account and distributes the small files across different partitions to avoid skew.
The strategy is designed with following steps:
1. Get the number of output partition number from Spark's original logic FilePartition.getFilePartitions. If `spark.sql.files.maxPartitionNum` is set, use the smaller one as the output partition number.
2. Assign small files starting from the smallest to the partitions with the minimum file count + total file size strategy
3. Assign the remaining files from the largest into the partition with the minimum total file size + file count strategy
The total size of small files can be configured using spark.gluten.sql.columnar.smallFileThreshold, which specifies the percentage of the total input file size represented by small files.
### 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.
-->
As described in the previous section. End users and projects like Apache Gluten can benefit from this change.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as new features, bug fixes, or other behavior changes. Documentation-only updates are not considered user-facing changes.
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'.
-->
new configurations are added for this enhancement:
- spark.sql.files.partitionStrategy
- spark.sql.files.smallFileThreshold
### 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
### Was this patch authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this patch, please include the
phrase: 'Generated-by: ' followed by the name of the tool and its version.
If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
No
--
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #53040:
URL: https://github.com/apache/spark/pull/53040#discussion_r2555902959
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala:
##########
@@ -87,28 +87,116 @@ object FilePartition extends SessionStateHelper with Logging {
partitions.toSeq
}
+ private def getFilePartitionsByFileNum(
+ partitionedFiles: Seq[PartitionedFile],
+ outputPartitions: Int,
+ smallFileThreshold: Double): Seq[FilePartition] = {
+ // Flatten and sort descending by file size.
+ val filesSorted: Seq[(PartitionedFile, Long)] =
+ partitionedFiles
+ .map(f => (f, f.length))
+ .sortBy(_._2)(Ordering.Long.reverse)
+
+ val partitions = Seq.fill(outputPartitions)(mutable.ArrayBuffer.empty[PartitionedFile])
+
+ def addToBucket(
+ heap: mutable.PriorityQueue[(Long, Int, Int)],
+ file: PartitionedFile,
+ sz: Long): Unit = {
+ val (load, numFiles, idx) = heap.dequeue()
+ partitions(idx) += file
+ heap.enqueue((load + sz, numFiles + 1, idx))
+ }
+
+ // First by load, then by numFiles.
+ val heapByFileSize =
+ mutable.PriorityQueue.empty[(Long, Int, Int)](
+ Ordering
+ .by[(Long, Int, Int), (Long, Int)] {
+ case (load, numFiles, _) =>
+ (load, numFiles)
+ }
+ .reverse
+ )
+
+ if (smallFileThreshold > 0) {
+ val smallFileTotalSize = filesSorted.map(_._2).sum * smallFileThreshold
+ // First by numFiles, then by load.
+ val heapByFileNum =
+ mutable.PriorityQueue.empty[(Long, Int, Int)](
+ Ordering
+ .by[(Long, Int, Int), (Int, Long)] {
+ case (load, numFiles, _) =>
+ (numFiles, load)
+ }
+ .reverse
+ )
+
+ (0 until outputPartitions).foreach(i => heapByFileNum.enqueue((0L, 0, i)))
+
+ var numSmallFiles = 0
+ var smallFileSize = 0L
+ // Enqueue small files to the least number of files and the least load.
+ filesSorted.reverse.takeWhile(f => f._2 + smallFileSize <= smallFileTotalSize).foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileNum, file, sz)
+ numSmallFiles += 1
+ smallFileSize += sz
+ }
+
+ // Move buckets from heapByFileNum to heapByFileSize.
+ while (heapByFileNum.nonEmpty) {
+ heapByFileSize.enqueue(heapByFileNum.dequeue())
+ }
+
+ // Finally, enqueue remaining files.
+ filesSorted.take(filesSorted.size - numSmallFiles).foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileSize, file, sz)
+ }
+ } else {
+ (0 until outputPartitions).foreach(i => heapByFileSize.enqueue((0L, 0, i)))
+
+ filesSorted.foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileSize, file, sz)
Review Comment:
Yes. It's a trade-off between balancing the number of splits per partition and the size per partition.
--
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "abhiips07 (via GitHub)" <gi...@apache.org>.
abhiips07 commented on code in PR #53040:
URL: https://github.com/apache/spark/pull/53040#discussion_r2549170799
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2430,6 +2430,28 @@ object SQLConf {
.checkValue(v => v > 0, "The maximum number of partitions must be a positive integer.")
.createOptional
+ val FILES_PARTITION_STRATEGY = buildConf("spark.sql.files.partitionStrategy")
+ .doc("The strategy to coalesce small files into larger partitions when reading files. " +
+ "Options are `size_based` (coalesce based on size of files), and `file_based` "
+ + "(coalesce based on number of files). The number of output partitions depends on " +
+ "`spark.sql.files.maxPartitionBytes` and `spark.sql.files.maxPartitionNum`. " +
+ "This configuration is effective only when using file-based sources such as " +
+ "Parquet, JSON and ORC.")
+ .version("3.5.0")
+ .stringConf
+ .checkValues(Set("size_based", "file_based"))
+ .createWithDefault("size_based")
+
+ val SMALL_FILE_THRESHOLD =
+ buildConf("spark.sql.files.smallFileThreshold")
+ .doc(
+ "Defines the total size threshold for small files in a table scan. If the cumulative size " +
+ "of small files falls below this threshold, they are distributed across multiple " +
+ "partitions to avoid concentrating them in a single partition. This configuration is " +
+ "used when `spark.sql.files.coalesceStrategy` is set to `file_based`.")
+ .doubleConf
+ .createWithDefault(0.5)
Review Comment:
@marin-ma If i understand correctly, does 0.5 means "**if the total small files are less than 50% of a partition, distribute them across multiple partitions**"??
--
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #53040: [SPARK-54335][SQL] Reducing skew in the number of file splits per partition
URL: https://github.com/apache/spark/pull/53040
--
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "VindhyaG (via GitHub)" <gi...@apache.org>.
VindhyaG commented on code in PR #53040:
URL: https://github.com/apache/spark/pull/53040#discussion_r2554564269
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala:
##########
@@ -75,7 +75,7 @@ object FilePartition extends SessionStateHelper with Logging {
}
// Assign files to partitions using "Next Fit Decreasing"
- partitionedFiles.foreach { file =>
+ partitionedFiles.sortBy(_.length)(implicitly[Ordering[Long]].reverse).foreach { file =>
Review Comment:
The PR description mentions currently the partition files are already sorted then why do we need to sort Seq[PartitionedFile] again?
--
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "VindhyaG (via GitHub)" <gi...@apache.org>.
VindhyaG commented on code in PR #53040:
URL: https://github.com/apache/spark/pull/53040#discussion_r2554564269
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala:
##########
@@ -75,7 +75,7 @@ object FilePartition extends SessionStateHelper with Logging {
}
// Assign files to partitions using "Next Fit Decreasing"
- partitionedFiles.foreach { file =>
+ partitionedFiles.sortBy(_.length)(implicitly[Ordering[Long]].reverse).foreach { file =>
Review Comment:
The PR description mentions currently the partition files are already sorted then why do we need to sort Seq[PartitionedFile] again? As far as I understand https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala#L888 does that right?
--
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "VindhyaG (via GitHub)" <gi...@apache.org>.
VindhyaG commented on code in PR #53040:
URL: https://github.com/apache/spark/pull/53040#discussion_r2555454615
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala:
##########
@@ -87,28 +87,116 @@ object FilePartition extends SessionStateHelper with Logging {
partitions.toSeq
}
+ private def getFilePartitionsByFileNum(
+ partitionedFiles: Seq[PartitionedFile],
+ outputPartitions: Int,
+ smallFileThreshold: Double): Seq[FilePartition] = {
+ // Flatten and sort descending by file size.
+ val filesSorted: Seq[(PartitionedFile, Long)] =
+ partitionedFiles
+ .map(f => (f, f.length))
+ .sortBy(_._2)(Ordering.Long.reverse)
+
+ val partitions = Seq.fill(outputPartitions)(mutable.ArrayBuffer.empty[PartitionedFile])
+
+ def addToBucket(
+ heap: mutable.PriorityQueue[(Long, Int, Int)],
+ file: PartitionedFile,
+ sz: Long): Unit = {
+ val (load, numFiles, idx) = heap.dequeue()
+ partitions(idx) += file
+ heap.enqueue((load + sz, numFiles + 1, idx))
+ }
+
+ // First by load, then by numFiles.
+ val heapByFileSize =
+ mutable.PriorityQueue.empty[(Long, Int, Int)](
+ Ordering
+ .by[(Long, Int, Int), (Long, Int)] {
+ case (load, numFiles, _) =>
+ (load, numFiles)
+ }
+ .reverse
+ )
+
+ if (smallFileThreshold > 0) {
+ val smallFileTotalSize = filesSorted.map(_._2).sum * smallFileThreshold
+ // First by numFiles, then by load.
+ val heapByFileNum =
+ mutable.PriorityQueue.empty[(Long, Int, Int)](
+ Ordering
+ .by[(Long, Int, Int), (Int, Long)] {
+ case (load, numFiles, _) =>
+ (numFiles, load)
+ }
+ .reverse
+ )
+
+ (0 until outputPartitions).foreach(i => heapByFileNum.enqueue((0L, 0, i)))
+
+ var numSmallFiles = 0
+ var smallFileSize = 0L
+ // Enqueue small files to the least number of files and the least load.
+ filesSorted.reverse.takeWhile(f => f._2 + smallFileSize <= smallFileTotalSize).foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileNum, file, sz)
+ numSmallFiles += 1
+ smallFileSize += sz
+ }
+
+ // Move buckets from heapByFileNum to heapByFileSize.
+ while (heapByFileNum.nonEmpty) {
+ heapByFileSize.enqueue(heapByFileNum.dequeue())
+ }
+
+ // Finally, enqueue remaining files.
+ filesSorted.take(filesSorted.size - numSmallFiles).foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileSize, file, sz)
+ }
+ } else {
+ (0 until outputPartitions).foreach(i => heapByFileSize.enqueue((0L, 0, i)))
+
+ filesSorted.foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileSize, file, sz)
+ }
+ }
+
+ partitions.zipWithIndex.map {
+ case (p, idx) => FilePartition(idx, p.toArray)
+ }
+ }
+
def getFilePartitions(
sparkSession: SparkSession,
partitionedFiles: Seq[PartitionedFile],
maxSplitBytes: Long): Seq[FilePartition] = {
val conf = getSqlConf(sparkSession)
val openCostBytes = conf.filesOpenCostInBytes
val maxPartNum = conf.filesMaxPartitionNum
- val partitions = getFilePartitions(partitionedFiles, maxSplitBytes, openCostBytes)
- if (maxPartNum.exists(partitions.size > _)) {
- val totalSizeInBytes =
- partitionedFiles.map(_.length + openCostBytes).map(BigDecimal(_)).sum[BigDecimal]
- val desiredSplitBytes =
- (totalSizeInBytes / BigDecimal(maxPartNum.get)).setScale(0, RoundingMode.UP).longValue
- val desiredPartitions = getFilePartitions(partitionedFiles, desiredSplitBytes, openCostBytes)
- logWarning(log"The number of partitions is ${MDC(NUM_PARTITIONS, partitions.size)}, " +
- log"which exceeds the maximum number configured: " +
- log"${MDC(MAX_NUM_PARTITIONS, maxPartNum.get)}. Spark rescales it to " +
- log"${MDC(DESIRED_NUM_PARTITIONS, desiredPartitions.size)} by ignoring the " +
- log"configuration of ${MDC(CONFIG, SQLConf.FILES_MAX_PARTITION_BYTES.key)}.")
- desiredPartitions
- } else {
- partitions
+ val partitions = getFilePartitionsBySize(partitionedFiles, maxSplitBytes, openCostBytes)
Review Comment:
since we are first getting the original partition size although file per partition could be well distributed , can it end up with wide mismatch between size of each partition considering no.of partitions provided by original algo will be way above than ideal no. of partitions for the kind of uneven distribution this new algo is trying to solve?
--
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #53040:
URL: https://github.com/apache/spark/pull/53040#discussion_r2555858014
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala:
##########
@@ -87,28 +87,116 @@ object FilePartition extends SessionStateHelper with Logging {
partitions.toSeq
}
+ private def getFilePartitionsByFileNum(
+ partitionedFiles: Seq[PartitionedFile],
+ outputPartitions: Int,
+ smallFileThreshold: Double): Seq[FilePartition] = {
+ // Flatten and sort descending by file size.
+ val filesSorted: Seq[(PartitionedFile, Long)] =
+ partitionedFiles
+ .map(f => (f, f.length))
+ .sortBy(_._2)(Ordering.Long.reverse)
+
+ val partitions = Seq.fill(outputPartitions)(mutable.ArrayBuffer.empty[PartitionedFile])
+
+ def addToBucket(
+ heap: mutable.PriorityQueue[(Long, Int, Int)],
+ file: PartitionedFile,
+ sz: Long): Unit = {
+ val (load, numFiles, idx) = heap.dequeue()
+ partitions(idx) += file
+ heap.enqueue((load + sz, numFiles + 1, idx))
+ }
+
+ // First by load, then by numFiles.
+ val heapByFileSize =
+ mutable.PriorityQueue.empty[(Long, Int, Int)](
+ Ordering
+ .by[(Long, Int, Int), (Long, Int)] {
+ case (load, numFiles, _) =>
+ (load, numFiles)
+ }
+ .reverse
+ )
+
+ if (smallFileThreshold > 0) {
+ val smallFileTotalSize = filesSorted.map(_._2).sum * smallFileThreshold
+ // First by numFiles, then by load.
+ val heapByFileNum =
+ mutable.PriorityQueue.empty[(Long, Int, Int)](
+ Ordering
+ .by[(Long, Int, Int), (Int, Long)] {
+ case (load, numFiles, _) =>
+ (numFiles, load)
+ }
+ .reverse
+ )
+
+ (0 until outputPartitions).foreach(i => heapByFileNum.enqueue((0L, 0, i)))
+
+ var numSmallFiles = 0
+ var smallFileSize = 0L
+ // Enqueue small files to the least number of files and the least load.
+ filesSorted.reverse.takeWhile(f => f._2 + smallFileSize <= smallFileTotalSize).foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileNum, file, sz)
+ numSmallFiles += 1
+ smallFileSize += sz
+ }
+
+ // Move buckets from heapByFileNum to heapByFileSize.
+ while (heapByFileNum.nonEmpty) {
+ heapByFileSize.enqueue(heapByFileNum.dequeue())
+ }
+
+ // Finally, enqueue remaining files.
+ filesSorted.take(filesSorted.size - numSmallFiles).foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileSize, file, sz)
+ }
+ } else {
+ (0 until outputPartitions).foreach(i => heapByFileSize.enqueue((0L, 0, i)))
+
+ filesSorted.foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileSize, file, sz)
+ }
+ }
+
+ partitions.zipWithIndex.map {
+ case (p, idx) => FilePartition(idx, p.toArray)
+ }
+ }
+
def getFilePartitions(
sparkSession: SparkSession,
partitionedFiles: Seq[PartitionedFile],
maxSplitBytes: Long): Seq[FilePartition] = {
val conf = getSqlConf(sparkSession)
val openCostBytes = conf.filesOpenCostInBytes
val maxPartNum = conf.filesMaxPartitionNum
- val partitions = getFilePartitions(partitionedFiles, maxSplitBytes, openCostBytes)
- if (maxPartNum.exists(partitions.size > _)) {
- val totalSizeInBytes =
- partitionedFiles.map(_.length + openCostBytes).map(BigDecimal(_)).sum[BigDecimal]
- val desiredSplitBytes =
- (totalSizeInBytes / BigDecimal(maxPartNum.get)).setScale(0, RoundingMode.UP).longValue
- val desiredPartitions = getFilePartitions(partitionedFiles, desiredSplitBytes, openCostBytes)
- logWarning(log"The number of partitions is ${MDC(NUM_PARTITIONS, partitions.size)}, " +
- log"which exceeds the maximum number configured: " +
- log"${MDC(MAX_NUM_PARTITIONS, maxPartNum.get)}. Spark rescales it to " +
- log"${MDC(DESIRED_NUM_PARTITIONS, desiredPartitions.size)} by ignoring the " +
- log"configuration of ${MDC(CONFIG, SQLConf.FILES_MAX_PARTITION_BYTES.key)}.")
- desiredPartitions
- } else {
- partitions
+ val partitions = getFilePartitionsBySize(partitionedFiles, maxSplitBytes, openCostBytes)
Review Comment:
In very extreme cases it might happen. If it does, the user should not enable this algorithm. This algorithm is particularly beneficial in the common scenario we observed in the user's production environment, where the input contains many small files that are much smaller than most of the input splits. In that case, there won’t be a wide mismatch.
--
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #53040:
URL: https://github.com/apache/spark/pull/53040#discussion_r2555836695
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala:
##########
@@ -87,28 +87,116 @@ object FilePartition extends SessionStateHelper with Logging {
partitions.toSeq
}
+ private def getFilePartitionsByFileNum(
+ partitionedFiles: Seq[PartitionedFile],
+ outputPartitions: Int,
+ smallFileThreshold: Double): Seq[FilePartition] = {
+ // Flatten and sort descending by file size.
+ val filesSorted: Seq[(PartitionedFile, Long)] =
+ partitionedFiles
+ .map(f => (f, f.length))
+ .sortBy(_._2)(Ordering.Long.reverse)
+
+ val partitions = Seq.fill(outputPartitions)(mutable.ArrayBuffer.empty[PartitionedFile])
+
+ def addToBucket(
+ heap: mutable.PriorityQueue[(Long, Int, Int)],
+ file: PartitionedFile,
+ sz: Long): Unit = {
+ val (load, numFiles, idx) = heap.dequeue()
+ partitions(idx) += file
+ heap.enqueue((load + sz, numFiles + 1, idx))
+ }
+
+ // First by load, then by numFiles.
+ val heapByFileSize =
+ mutable.PriorityQueue.empty[(Long, Int, Int)](
+ Ordering
+ .by[(Long, Int, Int), (Long, Int)] {
+ case (load, numFiles, _) =>
+ (load, numFiles)
+ }
+ .reverse
+ )
+
+ if (smallFileThreshold > 0) {
+ val smallFileTotalSize = filesSorted.map(_._2).sum * smallFileThreshold
+ // First by numFiles, then by load.
+ val heapByFileNum =
+ mutable.PriorityQueue.empty[(Long, Int, Int)](
+ Ordering
+ .by[(Long, Int, Int), (Int, Long)] {
+ case (load, numFiles, _) =>
+ (numFiles, load)
+ }
+ .reverse
+ )
+
+ (0 until outputPartitions).foreach(i => heapByFileNum.enqueue((0L, 0, i)))
+
+ var numSmallFiles = 0
+ var smallFileSize = 0L
+ // Enqueue small files to the least number of files and the least load.
+ filesSorted.reverse.takeWhile(f => f._2 + smallFileSize <= smallFileTotalSize).foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileNum, file, sz)
+ numSmallFiles += 1
+ smallFileSize += sz
+ }
+
+ // Move buckets from heapByFileNum to heapByFileSize.
+ while (heapByFileNum.nonEmpty) {
+ heapByFileSize.enqueue(heapByFileNum.dequeue())
+ }
+
+ // Finally, enqueue remaining files.
+ filesSorted.take(filesSorted.size - numSmallFiles).foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileSize, file, sz)
+ }
+ } else {
+ (0 until outputPartitions).foreach(i => heapByFileSize.enqueue((0L, 0, i)))
+
+ filesSorted.foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileSize, file, sz)
Review Comment:
If the partition number is fixed, we cannot guarantee the max split size doesn't exceed the configured value. See https://github.com/apache/spark/blob/0ba9a9bd8d6dad71a5938c5edfc2687aba7ef4f5/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala#L104-L108
--
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "VindhyaG (via GitHub)" <gi...@apache.org>.
VindhyaG commented on code in PR #53040:
URL: https://github.com/apache/spark/pull/53040#discussion_r2555465035
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala:
##########
@@ -87,28 +87,116 @@ object FilePartition extends SessionStateHelper with Logging {
partitions.toSeq
}
+ private def getFilePartitionsByFileNum(
+ partitionedFiles: Seq[PartitionedFile],
+ outputPartitions: Int,
+ smallFileThreshold: Double): Seq[FilePartition] = {
+ // Flatten and sort descending by file size.
+ val filesSorted: Seq[(PartitionedFile, Long)] =
+ partitionedFiles
+ .map(f => (f, f.length))
+ .sortBy(_._2)(Ordering.Long.reverse)
+
+ val partitions = Seq.fill(outputPartitions)(mutable.ArrayBuffer.empty[PartitionedFile])
+
+ def addToBucket(
+ heap: mutable.PriorityQueue[(Long, Int, Int)],
+ file: PartitionedFile,
+ sz: Long): Unit = {
+ val (load, numFiles, idx) = heap.dequeue()
+ partitions(idx) += file
+ heap.enqueue((load + sz, numFiles + 1, idx))
+ }
+
+ // First by load, then by numFiles.
+ val heapByFileSize =
+ mutable.PriorityQueue.empty[(Long, Int, Int)](
+ Ordering
+ .by[(Long, Int, Int), (Long, Int)] {
+ case (load, numFiles, _) =>
+ (load, numFiles)
+ }
+ .reverse
+ )
+
+ if (smallFileThreshold > 0) {
+ val smallFileTotalSize = filesSorted.map(_._2).sum * smallFileThreshold
+ // First by numFiles, then by load.
+ val heapByFileNum =
+ mutable.PriorityQueue.empty[(Long, Int, Int)](
+ Ordering
+ .by[(Long, Int, Int), (Int, Long)] {
+ case (load, numFiles, _) =>
+ (numFiles, load)
+ }
+ .reverse
+ )
+
+ (0 until outputPartitions).foreach(i => heapByFileNum.enqueue((0L, 0, i)))
+
+ var numSmallFiles = 0
+ var smallFileSize = 0L
+ // Enqueue small files to the least number of files and the least load.
+ filesSorted.reverse.takeWhile(f => f._2 + smallFileSize <= smallFileTotalSize).foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileNum, file, sz)
+ numSmallFiles += 1
+ smallFileSize += sz
+ }
+
+ // Move buckets from heapByFileNum to heapByFileSize.
+ while (heapByFileNum.nonEmpty) {
+ heapByFileSize.enqueue(heapByFileNum.dequeue())
+ }
+
+ // Finally, enqueue remaining files.
+ filesSorted.take(filesSorted.size - numSmallFiles).foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileSize, file, sz)
+ }
+ } else {
+ (0 until outputPartitions).foreach(i => heapByFileSize.enqueue((0L, 0, i)))
+
+ filesSorted.foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileSize, file, sz)
Review Comment:
what happens if the size exceeds maxSplitSize while adding to one of the buckets? Shouldn't there be a check even in this new logic since the size of small files + large files during handing files above smallfilethreshold could end up greater than maxsplitsize?
--
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "VindhyaG (via GitHub)" <gi...@apache.org>.
VindhyaG commented on code in PR #53040:
URL: https://github.com/apache/spark/pull/53040#discussion_r2555878973
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala:
##########
@@ -87,28 +87,116 @@ object FilePartition extends SessionStateHelper with Logging {
partitions.toSeq
}
+ private def getFilePartitionsByFileNum(
+ partitionedFiles: Seq[PartitionedFile],
+ outputPartitions: Int,
+ smallFileThreshold: Double): Seq[FilePartition] = {
+ // Flatten and sort descending by file size.
+ val filesSorted: Seq[(PartitionedFile, Long)] =
+ partitionedFiles
+ .map(f => (f, f.length))
+ .sortBy(_._2)(Ordering.Long.reverse)
+
+ val partitions = Seq.fill(outputPartitions)(mutable.ArrayBuffer.empty[PartitionedFile])
+
+ def addToBucket(
+ heap: mutable.PriorityQueue[(Long, Int, Int)],
+ file: PartitionedFile,
+ sz: Long): Unit = {
+ val (load, numFiles, idx) = heap.dequeue()
+ partitions(idx) += file
+ heap.enqueue((load + sz, numFiles + 1, idx))
+ }
+
+ // First by load, then by numFiles.
+ val heapByFileSize =
+ mutable.PriorityQueue.empty[(Long, Int, Int)](
+ Ordering
+ .by[(Long, Int, Int), (Long, Int)] {
+ case (load, numFiles, _) =>
+ (load, numFiles)
+ }
+ .reverse
+ )
+
+ if (smallFileThreshold > 0) {
+ val smallFileTotalSize = filesSorted.map(_._2).sum * smallFileThreshold
+ // First by numFiles, then by load.
+ val heapByFileNum =
+ mutable.PriorityQueue.empty[(Long, Int, Int)](
+ Ordering
+ .by[(Long, Int, Int), (Int, Long)] {
+ case (load, numFiles, _) =>
+ (numFiles, load)
+ }
+ .reverse
+ )
+
+ (0 until outputPartitions).foreach(i => heapByFileNum.enqueue((0L, 0, i)))
+
+ var numSmallFiles = 0
+ var smallFileSize = 0L
+ // Enqueue small files to the least number of files and the least load.
+ filesSorted.reverse.takeWhile(f => f._2 + smallFileSize <= smallFileTotalSize).foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileNum, file, sz)
+ numSmallFiles += 1
+ smallFileSize += sz
+ }
+
+ // Move buckets from heapByFileNum to heapByFileSize.
+ while (heapByFileNum.nonEmpty) {
+ heapByFileSize.enqueue(heapByFileNum.dequeue())
+ }
+
+ // Finally, enqueue remaining files.
+ filesSorted.take(filesSorted.size - numSmallFiles).foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileSize, file, sz)
+ }
+ } else {
+ (0 until outputPartitions).foreach(i => heapByFileSize.enqueue((0L, 0, i)))
+
+ filesSorted.foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileSize, file, sz)
Review Comment:
Got it. But in case of the new logic, the chances of crossing maxSplitsize per partition is high even when max partition is not set right?
--
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "VindhyaG (via GitHub)" <gi...@apache.org>.
VindhyaG commented on code in PR #53040:
URL: https://github.com/apache/spark/pull/53040#discussion_r2555878973
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala:
##########
@@ -87,28 +87,116 @@ object FilePartition extends SessionStateHelper with Logging {
partitions.toSeq
}
+ private def getFilePartitionsByFileNum(
+ partitionedFiles: Seq[PartitionedFile],
+ outputPartitions: Int,
+ smallFileThreshold: Double): Seq[FilePartition] = {
+ // Flatten and sort descending by file size.
+ val filesSorted: Seq[(PartitionedFile, Long)] =
+ partitionedFiles
+ .map(f => (f, f.length))
+ .sortBy(_._2)(Ordering.Long.reverse)
+
+ val partitions = Seq.fill(outputPartitions)(mutable.ArrayBuffer.empty[PartitionedFile])
+
+ def addToBucket(
+ heap: mutable.PriorityQueue[(Long, Int, Int)],
+ file: PartitionedFile,
+ sz: Long): Unit = {
+ val (load, numFiles, idx) = heap.dequeue()
+ partitions(idx) += file
+ heap.enqueue((load + sz, numFiles + 1, idx))
+ }
+
+ // First by load, then by numFiles.
+ val heapByFileSize =
+ mutable.PriorityQueue.empty[(Long, Int, Int)](
+ Ordering
+ .by[(Long, Int, Int), (Long, Int)] {
+ case (load, numFiles, _) =>
+ (load, numFiles)
+ }
+ .reverse
+ )
+
+ if (smallFileThreshold > 0) {
+ val smallFileTotalSize = filesSorted.map(_._2).sum * smallFileThreshold
+ // First by numFiles, then by load.
+ val heapByFileNum =
+ mutable.PriorityQueue.empty[(Long, Int, Int)](
+ Ordering
+ .by[(Long, Int, Int), (Int, Long)] {
+ case (load, numFiles, _) =>
+ (numFiles, load)
+ }
+ .reverse
+ )
+
+ (0 until outputPartitions).foreach(i => heapByFileNum.enqueue((0L, 0, i)))
+
+ var numSmallFiles = 0
+ var smallFileSize = 0L
+ // Enqueue small files to the least number of files and the least load.
+ filesSorted.reverse.takeWhile(f => f._2 + smallFileSize <= smallFileTotalSize).foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileNum, file, sz)
+ numSmallFiles += 1
+ smallFileSize += sz
+ }
+
+ // Move buckets from heapByFileNum to heapByFileSize.
+ while (heapByFileNum.nonEmpty) {
+ heapByFileSize.enqueue(heapByFileNum.dequeue())
+ }
+
+ // Finally, enqueue remaining files.
+ filesSorted.take(filesSorted.size - numSmallFiles).foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileSize, file, sz)
+ }
+ } else {
+ (0 until outputPartitions).foreach(i => heapByFileSize.enqueue((0L, 0, i)))
+
+ filesSorted.foreach {
+ case (file, sz) =>
+ addToBucket(heapByFileSize, file, sz)
Review Comment:
Got it. But in case of the new logic, the chances of crossing maxSplitsize per partition is high even when max partition is not set, isn't 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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #53040:
URL: https://github.com/apache/spark/pull/53040#discussion_r2555812941
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2430,6 +2430,28 @@ object SQLConf {
.checkValue(v => v > 0, "The maximum number of partitions must be a positive integer.")
.createOptional
+ val FILES_PARTITION_STRATEGY = buildConf("spark.sql.files.partitionStrategy")
+ .doc("The strategy to coalesce small files into larger partitions when reading files. " +
+ "Options are `size_based` (coalesce based on size of files), and `file_based` "
+ + "(coalesce based on number of files). The number of output partitions depends on " +
+ "`spark.sql.files.maxPartitionBytes` and `spark.sql.files.maxPartitionNum`. " +
+ "This configuration is effective only when using file-based sources such as " +
+ "Parquet, JSON and ORC.")
+ .version("3.5.0")
+ .stringConf
+ .checkValues(Set("size_based", "file_based"))
+ .createWithDefault("size_based")
+
+ val SMALL_FILE_THRESHOLD =
+ buildConf("spark.sql.files.smallFileThreshold")
+ .doc(
+ "Defines the total size threshold for small files in a table scan. If the cumulative size " +
+ "of small files falls below this threshold, they are distributed across multiple " +
+ "partitions to avoid concentrating them in a single partition. This configuration is " +
+ "used when `spark.sql.files.coalesceStrategy` is set to `file_based`.")
+ .doubleConf
+ .createWithDefault(0.5)
Review Comment:
0.5 represents half of the total file size. For example, with 1 TB to scan, sort files from smallest to largest and classify those as small files until the cumulative size reaches 500 GB.
--
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #53040:
URL: https://github.com/apache/spark/pull/53040#issuecomment-4001220177
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "VindhyaG (via GitHub)" <gi...@apache.org>.
VindhyaG commented on code in PR #53040:
URL: https://github.com/apache/spark/pull/53040#discussion_r2554564269
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala:
##########
@@ -75,7 +75,7 @@ object FilePartition extends SessionStateHelper with Logging {
}
// Assign files to partitions using "Next Fit Decreasing"
- partitionedFiles.foreach { file =>
+ partitionedFiles.sortBy(_.length)(implicitly[Ordering[Long]].reverse).foreach { file =>
Review Comment:
The PR description mentions Seq[PartitionedFile] are alerady sorted then why do we need to sort again?
--
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "marin-ma (via GitHub)" <gi...@apache.org>.
marin-ma commented on code in PR #53040:
URL: https://github.com/apache/spark/pull/53040#discussion_r2555794369
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/FilePartition.scala:
##########
@@ -75,7 +75,7 @@ object FilePartition extends SessionStateHelper with Logging {
}
// Assign files to partitions using "Next Fit Decreasing"
- partitionedFiles.foreach { file =>
+ partitionedFiles.sortBy(_.length)(implicitly[Ordering[Long]].reverse).foreach { file =>
Review Comment:
I planned to remove the sort from the caller like the one you mentioned. Seems like this part of change is missing from this patch. Thanks for catching.
--
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
Re: [PR] [SPARK-54335][SQL] Reducing skew in the number of file splits per partition [spark]
Posted by "VindhyaG (via GitHub)" <gi...@apache.org>.
VindhyaG commented on code in PR #53040:
URL: https://github.com/apache/spark/pull/53040#discussion_r2540570386
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -2430,6 +2430,28 @@ object SQLConf {
.checkValue(v => v > 0, "The maximum number of partitions must be a positive integer.")
.createOptional
+ val FILES_PARTITION_STRATEGY = buildConf("spark.sql.files.partitionStrategy")
+ .doc("The strategy to coalesce small files into larger partitions when reading files. " +
+ "Options are `size_based` (coalesce based on size of files), and `file_based` "
+ + "(coalesce based on number of files). The number of output partitions depends on " +
+ "`spark.sql.files.maxPartitionBytes` and `spark.sql.files.maxPartitionNum`. " +
+ "This configuration is effective only when using file-based sources such as " +
+ "Parquet, JSON and ORC.")
+ .version("3.5.0")
+ .stringConf
+ .checkValues(Set("size_based", "file_based"))
+ .createWithDefault("size_based")
+
+ val SMALL_FILE_THRESHOLD =
+ buildConf("spark.sql.files.smallFileThreshold")
+ .doc(
+ "Defines the total size threshold for small files in a table scan. If the cumulative size " +
+ "of small files falls below this threshold, they are distributed across multiple " +
+ "partitions to avoid concentrating them in a single partition. This configuration is " +
+ "used when `spark.sql.files.coalesceStrategy` is set to `file_based`.")
+ .doubleConf
+ .createWithDefault(0.5)
Review Comment:
may I ask how did we arrive with 0.5 as the default smallfilethreshold?
--
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