Qing Fu created SPARK-59510:
-------------------------------
Summary: Derive UnsafeExternalSorter spill merge factor from a
memory budget
Key: SPARK-59510
URL: https://issues.apache.org/jira/browse/SPARK-59510
Project: Spark
Issue Type: Improvement
Components: Spark Core
Affects Versions: 5.0.0
Reporter: Qing Fu
SPARK-56410 added bounded multi-round merging to {{UnsafeExternalSorter}},
controlled by {{spark.unsafe.sorter.spill.merge.factor}}: the maximum number of
spill files merged in a single round.
The quantity the merge phase actually needs to bound is memory, not a file
count. Each open {{UnsafeSorterSpillReader}} holds a read-ahead buffer, a
decompression buffer and a buffer sized for the largest record it has returned,
so the memory a round needs is roughly {{factor * perReaderBytes}}. Choosing a
good {{factor}} therefore means knowing {{perReaderBytes}}, which is a function
of settings the user has already configured.
Proposal: allow the factor to be derived from a memory budget. Spark already
exposes the inputs:
* {{spark.unsafe.sorter.spill.reader.buffer.size}} (default 1m) - the reader's
I/O buffer
* {{spark.unsafe.sorter.spill.read.ahead.enabled}} (default true) - read-ahead
holds a second buffer
so the derivation needs only one new config, the budget itself:
{code}
perReader = readerBufferSize * (readAheadEnabled ? 2 : 1) + recordAllowance
factor = max(2, maxMemory / perReader)
{code}
An explicitly configured {{spark.unsafe.sorter.spill.merge.factor}} would
continue to win; the derivation would only apply when the user asks for it.
Two related questions, worth settling here rather than in review:
1. Should bounded merging be enabled by default?
{{spark.unsafe.sorter.spill.merge.factor}} currently defaults to {{-1}}
(disabled), while the {{UnsafeSorterBoundedSpillMerger}} class javadoc already
describes "the default factor of 64". From operating a large Spark deployment
with multi-round merging enabled by default, we have seen a substantial
reduction in merge-phase OOMs and in the compute cost of the affected
workloads, with no correctness or performance regression observed across a
broad benchmark set. Tasks that spill hundreds of times are exactly the ones
that most need the sorter, and today such a task can write every spill
successfully and then OOM in the merge, with no recovery other than a larger
executor.
2. {{UnsafeSorterIterator.getNumRecords()}} returns {{int}}. Widening it is the
real fix for the record-count ceiling that
{{UnsafeSorterBoundedSpillMerger.partitionWriters}} currently works around, but
it changes the spill file header format, so it likely deserves its own
discussion.
Happy to put up a PR for the memory-budget derivation if the approach sounds
reasonable.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
