[
https://issues.apache.org/jira/browse/SPARK-59510?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Qing Fu updated SPARK-59510:
----------------------------
Description:
SPARK-56410 added bounded multi-round merging to {{UnsafeExternalSorter}},
controlled by {{spark.unsafe.sorter.spill.merge.factor}}. It shipped disabled:
the config defaults to {{-1}}.
h3. The code already disagrees with itself
{{UnsafeSorterBoundedSpillMerger}} documents a default of 64:
{quote}A smaller merge factor requires more rounds (and thus more I/O), while a
larger factor uses more memory. *The default factor of 64* typically requires
at most one intermediate round for up to ~4000 spill files.{quote}
but {{core/src/main/scala/org/apache/spark/internal/config/package.scala}} has:
{code:scala}
private[spark] val UNSAFE_SORTER_SPILL_MERGE_FACTOR =
ConfigBuilder("spark.unsafe.sorter.spill.merge.factor")
...
.createWithDefault(-1)
{code}
So either the class documentation is stale, or the default is not what was
intended. Either way the two should be reconciled, and this ticket proposes
reconciling them in favour of the documented value.
h3. Why enabling by default is the right resolution
The failure this guards against is not gradual. {{UnsafeSorterSpillMerger}}
opens every spill file at once, and each open {{UnsafeSorterSpillReader}} holds
a read-ahead buffer, a decompression buffer and a buffer sized for the largest
record it has returned. The memory the merge phase needs therefore grows
linearly with the number of spills, bounded by nothing. A task that spills a
few hundred times can write every spill successfully and then OOM in the merge,
with no recovery available to the user other than giving the executor more
memory. The workloads that hit this are precisely the ones that need the sorter
most.
The cost of enabling it is bounded and conditional:
* the multi-round path only engages when the spill count exceeds the factor, so
jobs that spill less than 64 times are completely unaffected;
* when it does engage, the extra cost is disk I/O for the intermediate rounds,
which is what the user is trading for not failing;
* consumed files are deleted eagerly after each group merge, so peak disk
overhead stays around one group's worth above the original spill total.
In other words, the default of {{-1}} preserves an unbounded-memory merge for
everyone in order to avoid I/O for the minority of jobs that spill heavily --
and those are the jobs the feature exists for.
h3. Evidence
We run bounded multi-round merging enabled by default across a large production
Spark deployment. We see a substantial reduction in merge-phase OOMs and a
meaningful reduction in the compute cost of the affected workloads, with no
correctness regression and no performance regression observed across a broad
benchmark set. I am not in a position to publish the specific figures, so
please weigh this as qualitative operational experience rather than as data you
can audit; the argument above stands on the code alone.
h3. Proposed change
Change the default of {{spark.unsafe.sorter.spill.merge.factor}} from {{-1}} to
{{64}}, matching the value {{UnsafeSorterBoundedSpillMerger}} already
documents. {{-1}} remains available to restore the previous single-round
behaviour.
Open questions for whoever owns this area:
* Is 64 the right value, or was {{-1}} a deliberate soak period for a then-new
feature (SPARK-56410 landed in 4.2.0, with a race fixed in SPARK-56873)? If the
latter, is there a release at which enabling it becomes reasonable?
* Should this be treated as a behaviour change requiring a migration-guide
note, given it is an internal config and the observable difference is disk I/O
rather than results?
h3. Note
This ticket previously proposed deriving the merge factor from a memory budget
rather than a fixed count. That is a larger change and a separate argument; it
has been dropped from this ticket in favour of the much smaller default change.
Happy to raise it separately if there is interest.
was:
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.
Summary: Enable bounded UnsafeExternalSorter spill merge by default
(was: Derive UnsafeExternalSorter spill merge factor from a memory budget)
> Enable bounded UnsafeExternalSorter spill merge by default
> ----------------------------------------------------------
>
> 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
> Priority: Major
>
> SPARK-56410 added bounded multi-round merging to {{UnsafeExternalSorter}},
> controlled by {{spark.unsafe.sorter.spill.merge.factor}}. It shipped
> disabled: the config defaults to {{-1}}.
> h3. The code already disagrees with itself
> {{UnsafeSorterBoundedSpillMerger}} documents a default of 64:
> {quote}A smaller merge factor requires more rounds (and thus more I/O), while
> a larger factor uses more memory. *The default factor of 64* typically
> requires at most one intermediate round for up to ~4000 spill files.{quote}
> but {{core/src/main/scala/org/apache/spark/internal/config/package.scala}}
> has:
> {code:scala}
> private[spark] val UNSAFE_SORTER_SPILL_MERGE_FACTOR =
> ConfigBuilder("spark.unsafe.sorter.spill.merge.factor")
> ...
> .createWithDefault(-1)
> {code}
> So either the class documentation is stale, or the default is not what was
> intended. Either way the two should be reconciled, and this ticket proposes
> reconciling them in favour of the documented value.
> h3. Why enabling by default is the right resolution
> The failure this guards against is not gradual. {{UnsafeSorterSpillMerger}}
> opens every spill file at once, and each open {{UnsafeSorterSpillReader}}
> holds a read-ahead buffer, a decompression buffer and a buffer sized for the
> largest record it has returned. The memory the merge phase needs therefore
> grows linearly with the number of spills, bounded by nothing. A task that
> spills a few hundred times can write every spill successfully and then OOM in
> the merge, with no recovery available to the user other than giving the
> executor more memory. The workloads that hit this are precisely the ones that
> need the sorter most.
> The cost of enabling it is bounded and conditional:
> * the multi-round path only engages when the spill count exceeds the factor,
> so jobs that spill less than 64 times are completely unaffected;
> * when it does engage, the extra cost is disk I/O for the intermediate
> rounds, which is what the user is trading for not failing;
> * consumed files are deleted eagerly after each group merge, so peak disk
> overhead stays around one group's worth above the original spill total.
> In other words, the default of {{-1}} preserves an unbounded-memory merge for
> everyone in order to avoid I/O for the minority of jobs that spill heavily --
> and those are the jobs the feature exists for.
> h3. Evidence
> We run bounded multi-round merging enabled by default across a large
> production Spark deployment. We see a substantial reduction in merge-phase
> OOMs and a meaningful reduction in the compute cost of the affected
> workloads, with no correctness regression and no performance regression
> observed across a broad benchmark set. I am not in a position to publish the
> specific figures, so please weigh this as qualitative operational experience
> rather than as data you can audit; the argument above stands on the code
> alone.
> h3. Proposed change
> Change the default of {{spark.unsafe.sorter.spill.merge.factor}} from {{-1}}
> to {{64}}, matching the value {{UnsafeSorterBoundedSpillMerger}} already
> documents. {{-1}} remains available to restore the previous single-round
> behaviour.
> Open questions for whoever owns this area:
> * Is 64 the right value, or was {{-1}} a deliberate soak period for a
> then-new feature (SPARK-56410 landed in 4.2.0, with a race fixed in
> SPARK-56873)? If the latter, is there a release at which enabling it becomes
> reasonable?
> * Should this be treated as a behaviour change requiring a migration-guide
> note, given it is an internal config and the observable difference is disk
> I/O rather than results?
> h3. Note
> This ticket previously proposed deriving the merge factor from a memory
> budget rather than a fixed count. That is a larger change and a separate
> argument; it has been dropped from this ticket in favour of the much smaller
> default change. Happy to raise it separately if there is interest.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
