Skip to content

[SPARK-49386][CORE][SQL] Add memory based thresholds for shuffle spill #47856

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
cxzl25 wants to merge 12 commits into from
Closed

[SPARK-49386][CORE][SQL] Add memory based thresholds for shuffle spill #47856

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
647afb8
[SPARK-27734][CORE][SQL] Add memory based thresholds for shuffle spill
amuraru Sep 15, 2019
044a232
spill by size
cxzl25 Aug 23, 2024
27c6478
config
cxzl25 Aug 23, 2024
a3f5d16
codegen
cxzl25 Aug 23, 2024
14a7f2d
add test
cxzl25 Aug 27, 2024
8374d18
use _elementsRead
cxzl25 Apr 21, 2025
57f0711
checkValue fallbackConf
cxzl25 Apr 21, 2025
d734bfa
log
cxzl25 May 3, 2025
7f78264
shouldSpill val
cxzl25 May 6, 2025
b28bd7b
config version 4.1.0
cxzl25 May 26, 2025
d083773
extra line
cxzl25 Jun 26, 2025
8225e94
test
cxzl25 Jun 26, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
log
  • Loading branch information
@cxzl25
cxzl25 committed Jun 26, 2025
commit d734bfa62e03fd8735de7901d3e44d1747978ca7
2 changes: 2 additions & 0 deletions common/utils/src/main/scala/org/apache/spark/internal/LogKey.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -769,6 +769,8 @@ private[spark] object LogKeys {
case object SPARK_REPO_URL extends LogKey
case object SPARK_REVISION extends LogKey
case object SPARK_VERSION extends LogKey
case object SPILL_RECORDS_SIZE extends LogKey
case object SPILL_RECORDS_SIZE_THRESHOLD extends LogKey
case object SPILL_TIMES extends LogKey
case object SQL_TEXT extends LogKey
case object SRC_PATH extends LogKey
Expand Down
8 changes: 4 additions & 4 deletions core/src/main/java/org/apache/spark/shuffle/sort/ShuffleExternalSorter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -429,13 +429,13 @@ public void insertRecord(Object recordBase, long recordOffset, int length, int p
assert(inMemSorter != null);
if (inMemSorter.numRecords() >= numElementsForSpillThreshold) {
logger.info("Spilling data because number of spilledRecords ({}) crossed the threshold {}",
MDC.of(LogKeys.NUM_ELEMENTS_SPILL_RECORDS$.MODULE$,inMemSorter.numRecords()),
MDC.of(LogKeys.NUM_ELEMENTS_SPILL_RECORDS$.MODULE$, inMemSorter.numRecords()),
MDC.of(LogKeys.NUM_ELEMENTS_SPILL_THRESHOLD$.MODULE$, numElementsForSpillThreshold));
spill();
} else if (inMemRecordsSize >= recordsSizeForSpillThreshold) {
logger.info("Spilling data because size of spilledRecords ({}) crossed the threshold {}",
MDC.of(LogKeys.NUM_ELEMENTS_SPILL_RECORDS$.MODULE$,inMemSorter.numRecords()),
MDC.of(LogKeys.NUM_ELEMENTS_SPILL_THRESHOLD$.MODULE$, recordsSizeForSpillThreshold));
logger.info("Spilling data because size of spilledRecords ({}) crossed the size threshold {}",
MDC.of(LogKeys.SPILL_RECORDS_SIZE$.MODULE$, inMemRecordsSize),
MDC.of(LogKeys.SPILL_RECORDS_SIZE_THRESHOLD$.MODULE$, recordsSizeForSpillThreshold));
spill();
}

Expand Down
8 changes: 4 additions & 4 deletions core/src/main/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -493,13 +493,13 @@ public void insertRecord(
assert(inMemSorter != null);
if (inMemSorter.numRecords() >= numElementsForSpillThreshold) {
logger.info("Spilling data because number of spilledRecords ({}) crossed the threshold {}",
MDC.of(LogKeys.NUM_ELEMENTS_SPILL_RECORDS$.MODULE$,inMemSorter.numRecords()),
MDC.of(LogKeys.NUM_ELEMENTS_SPILL_RECORDS$.MODULE$, inMemSorter.numRecords()),
MDC.of(LogKeys.NUM_ELEMENTS_SPILL_THRESHOLD$.MODULE$, numElementsForSpillThreshold));
spill();
} else if (inMemRecordsSize >= recordsSizeForSpillThreshold) {
logger.info("Spilling data because size of spilledRecords ({}) crossed the threshold {}",
MDC.of(LogKeys.NUM_ELEMENTS_SPILL_RECORDS$.MODULE$,inMemSorter.numRecords()),
MDC.of(LogKeys.NUM_ELEMENTS_SPILL_THRESHOLD$.MODULE$, recordsSizeForSpillThreshold));
logger.info("Spilling data because size of spilledRecords ({}) crossed the size threshold {}",
MDC.of(LogKeys.SPILL_RECORDS_SIZE$.MODULE$, inMemRecordsSize),
MDC.of(LogKeys.SPILL_RECORDS_SIZE_THRESHOLD$.MODULE$, recordsSizeForSpillThreshold));
spill();
}

Expand Down