[
https://issues.apache.org/jira/browse/SPARK-60601?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Wenchen Fan updated SPARK-60601:
--------------------------------
Description:
When the aggregation map of adaptive partial aggregation is about to spill, the
spill check (SPARK-60561) decides whether to keep aggregating through spills or
pass the rest of the input through. It judges the recent rows of the current
map epoch, which cannot show whether keys come back after the map fills: with
keys spread evenly over a task, the recent ratio reflects how much of the
task's distinct keys the map holds, not how often each key repeats. So at the
default {{minCompactionToSpill}}, a task whose map holds well under half of its
keys passes the rest through even when each key repeats many times. In the
"keys recur, map holds 40% of them" cases of
{{AdaptivePartialAggregationBenchmark}} that is within about 8% of keeping the
aggregation on most runners, but up to about 20% slower on one.
Add a signal for keys that recur across epochs, at no per-row cost beyond a bit
test. For example, keep a small sample of the first epoch's keys, chosen by
bits of the hash the map already computes, and count how often later epochs hit
it. With the two-level map on, the fast map's keys, which outlive every spill,
can serve as that sample. Since no later epoch exists at the first fill, using
the signal may mean deferring the decision past the first spill.
The same change could cover inputs where one or two keys, such as a NULL or
default value, hold a large share of the rows over many distinct ones. Once
pass-through starts, every later row of such a key reaches its one reducer as
its own partial buffer, and AQE does not split skewed aggregation shuffles.
Keeping a few heavy keys of the frozen map aggregating during pass-through
would avoid that, at the cost of a probe for every passed-through row, which
needs measuring.
was:
When the aggregation map of adaptive partial aggregation is about to spill, the
spill check (SPARK-60561) decides whether to keep aggregating through spills or
pass the rest of the input through. It judges the recent rows of the current
map epoch, which cannot show whether keys come back after the map fills: with
keys spread evenly over a task, the recent ratio reflects how much of the
task's distinct keys the map holds, not how often each key repeats. So at the
default {{minCompactionToSpill}}, a task whose map holds well under half of its
keys passes the rest through even when each key repeats many times. In the
"keys recur, map holds 40% of them" cases of
{{AdaptivePartialAggregationBenchmark}} that is within about 8% of keeping the
aggregation on most runners, but up to about 20% slower on one.
Add a signal for keys that recur across epochs, at no per-row cost beyond a bit
test. For example, keep a small sample of the first epoch's keys, chosen by
bits of the hash the map already computes, and count how often later epochs hit
it. With the two-level map on, the fast map's keys, which outlive every spill,
can serve as that sample. Since no later epoch exists at the first fill, using
the signal may mean deferring the decision past the first spill.
> Detect keys that recur across epochs in the adaptive partial aggregation
> spill check
> ------------------------------------------------------------------------------------
>
> Key: SPARK-60601
> URL: https://issues.apache.org/jira/browse/SPARK-60601
> Project: Spark
> Issue Type: Improvement
> Components: SQL
> Affects Versions: 4.4.0
> Reporter: Wenchen Fan
> Priority: Major
>
> When the aggregation map of adaptive partial aggregation is about to spill,
> the spill check (SPARK-60561) decides whether to keep aggregating through
> spills or pass the rest of the input through. It judges the recent rows of
> the current map epoch, which cannot show whether keys come back after the map
> fills: with keys spread evenly over a task, the recent ratio reflects how
> much of the task's distinct keys the map holds, not how often each key
> repeats. So at the default {{minCompactionToSpill}}, a task whose map holds
> well under half of its keys passes the rest through even when each key
> repeats many times. In the "keys recur, map holds 40% of them" cases of
> {{AdaptivePartialAggregationBenchmark}} that is within about 8% of keeping
> the aggregation on most runners, but up to about 20% slower on one.
> Add a signal for keys that recur across epochs, at no per-row cost beyond a
> bit test. For example, keep a small sample of the first epoch's keys, chosen
> by bits of the hash the map already computes, and count how often later
> epochs hit it. With the two-level map on, the fast map's keys, which outlive
> every spill, can serve as that sample. Since no later epoch exists at the
> first fill, using the signal may mean deferring the decision past the first
> spill.
> The same change could cover inputs where one or two keys, such as a NULL or
> default value, hold a large share of the rows over many distinct ones. Once
> pass-through starts, every later row of such a key reaches its one reducer as
> its own partial buffer, and AQE does not split skewed aggregation shuffles.
> Keeping a few heavy keys of the frozen map aggregating during pass-through
> would avoid that, at the cost of a probe for every passed-through row, which
> needs measuring.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
