cloud-fan commented on code in PR #59823:
URL: https://github.com/apache/spark/pull/59823#discussion_r4395121456
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/benchmark/AdaptivePartialAggregationBenchmark.scala:
##########
@@ -135,6 +135,55 @@ object AdaptivePartialAggregationBenchmark extends
SqlBasedBenchmark {
extraConf = spillCheckConf)
benchmark.run()
}
+
+ // Inputs that compact by about 1.5 within each map epoch, read from four
partitions so an
+ // `Exchange` separates the Partial from the Final, and with a map forced
to fill every few
+ // hundred thousand rows. Whether keeping the aggregation through the
spills pays off depends
+ // on whether the rest of the input keeps repeating keys:
+ // - each key's rows arrive together, so later epochs share no keys and
passing the rest
+ // through at the first fill avoids spilling every epoch;
+ // - keys recur uniformly across the task (10 rows per key), so the
final merge collapses
+ // them and keeping the aggregation shuffles a tenth of the rows;
+ // - 10 hot keys hold 30% of the rows and the rest are unique, so
passing through avoids the
+ // spills but sends every hot-key row to the reducer of its key.
+ // `minCompactionToSpill = 1.05` judges the spill check by
`minCompaction`; the default, 2.0,
+ // demands that the recent rows compact by 2.0.
+ val spillCheckCases = Seq(
+ ("keys arrive together", "cast(id * 2 / 3 as bigint)", 300000),
+ ("keys recur", "pmod(xxhash64(id), 200000)", 220000),
+ ("hot keys", "if(id % 100 < 30, id % 10, id + 1000)", 300000))
+ spillCheckCases.foreach { case (name, key, fillRows) =>
+ runBenchmark(s"spill check with an Exchange, $name") {
Review Comment:
Done. d74b8d0 extends the class doc, 75d7d40 adds the 40% cases with the
fast map off and on, and 6a5684c has the three result files from the "Run
benchmarks" workflow.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregationIterator.scala:
##########
@@ -225,10 +230,16 @@ class TungstenAggregationIterator(
// The spill check waits for `minRows` rows of the current epoch, and
for at least one row
// when `minRows = 0`.
val spillCheckMinRows = math.max(minRows, 1L)
- // The partial aggregation is ineffective when it does not collapse
`minCompaction` rows into
- // one key. There is no fast map on this path, so the map's keys are all
the operator holds.
- def ineffective(): Boolean =
- processedRows < hashMap.getNumKeys().toDouble * adaptiveMinCompaction
+ // The epoch's processed rows and keys at its last periodic check, and
at the check before
+ // that, where the window of recent rows that the spill check judges
starts.
+ var checkRows = 0L
+ var checkKeys = 0L
+ var windowRows = 0L
Review Comment:
Done in d74b8d0. Both paths keep only the two key snapshots and derive the
window's row start as `max(0, nextCheckRow - 2 * minRows)`.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
