Vivek1106-04 commented on PR #57528:
URL: https://github.com/apache/spark/pull/57528#issuecomment-5170718701
Both reproductions confirmed at d3be2ba389d, and they land on your numbers
exactly. A reports a median of 57,393,120 and a threshold of 286,965,600, with
19 of the 120 tied reducers under it. B reports 636,320,000 against
754,350,000, all 200 missed.
I don't think this can be fixed by rotating better, so rather than send
another window schedule I want to make the case that the approach is wrong.
In B the tie group is exactly twice the cap, so coverage is already a
perfect 50% and stays 50% under any rotation policy, including a full-window
stride. The miss doesn't come from coverage. It comes from avgSize: the 100
ties each map task drops are folded into totalSmallBlockSize, lifting avgSize
to 15,088, lifting the median reducer to 150,870,000, lifting the threshold to
754,350,000. Recording more ties per map task raises the reported size of the
tied reducers and the threshold they're compared against at the same time. No
schedule closes that gap. A is a genuine coverage bug on top of it — a stride
of one reaches only 100 + M - 1 ties — but fixing the stride alone leaves B
failing.
So I've stopped rationing the ties. They're all exactly skewCutoff, so the
missing information is which reducers are at that size, not what their sizes
are. That's one bitmap plus one byte: HighlyCompressedMapStatus now carries a
run-optimized RoaringBitmap of the tied blocks alongside the existing
emptyBlocks one, and the shared compressed cutoff size. Ties are excluded from
the avgSize average, since they're recorded exactly and folding them in would
only inflate the size reported for the blocks that really are average. The
rotation, the tie counting pass and the mapTaskId-derived offset are gone; the
id and size arrays still hold only the blocks strictly above the cutoff, which
nthSmallest bounds.
This does drop the per-status cap on tied blocks, which your P1 asked for
and which you asked me to preserve here. That's deliberate — the cap is the
thing that makes this unsolvable — but it's your call, so the cost is measured
below rather than asserted.
| | reproduction A | reproduction B |
|----------------------------|----------------|----------------|
| tied reducers / map tasks | 120 / 10 | 200 / 10,000 |
| real size per tied reducer | 524,288,000 | 1,024,000,000 |
| reported, before | 57,393,120 | 636,320,000 |
| AQE threshold, before | 286,965,600 | 754,350,000 |
| detected, before | 101 / 120 | 0 / 200 |
| reported, after | 550,087,940 | 1,121,770,000 |
| AQE threshold, after | 268,435,456 | 512,000,000 |
| detected, after | 120 / 120 | 200 / 200 |
Retained driver heap, SizeEstimator over the 10,000 deserialized map
statuses of a 2,001 partition stage, which is what your earlier P2 asked for:
| | bytes per map task | serialized |
|----------------------------------------|--------------------|------------|
| factor -1.0, before | 164 | 13 KiB |
| factor 5.0, before (100 ties recorded) | 668 | 27 KiB |
| factor -1.0, after | 172 | 15 KiB |
| factor 5.0, after (every tie recorded) | 308 | 15 KiB |
Recording all 200 ties now costs less than half of what recording 100 of
them did. Two caveats. The disabled case grows 8 bytes per map task for the
extra field; statuses with no ties share one never-mutated empty bitmap, which
is what keeps it to that. And the bitmap's cost tracks partition count, not tie
count: your 50,000-partition case from the first round, with 24,999 ties, is a
single run and retains 448 bytes per status, but a maximally scattered tie set
at that width would be closer to 6.25 KiB, which is worse than the capped
array. Bounded by the partition count either way, which the id array was not.
HighlyCompressedMapStatusBenchmark, back to back on the same machine, ns per
block to build one status with sizes recorded:
| partitions | skewed, before | skewed, after | organ pipe, before | organ
pipe, after |
|------------|----------------|---------------|--------------------|-------------------|
| 2,048 | 17,875 | 18,250 | 24,875 | 24,917
|
| 10,000 | 56,750 | 55,833 | 105,208 | 100,333
|
| 50,000 | 346,417 | 310,750 | 555,458 | 531,458
|
Construction is flat to slightly faster — the tie counting pass is gone,
which pays for the bitmap insertions — and the -1.0 cases are unchanged.
getSizeForBlock gets faster, 4.7 to 3.9 ns per block at 2,048 partitions and
3.9 to 3.0 at 50,000, since the ties leave the binary-searched array. Laptop,
single run per case, so treat small differences as noise and the direction as
the claim.
On the tests, the two rounds of misses share a cause worth naming: they
asserted the mechanism — coverage, and per-reducer error against a fixed target
— rather than the property. Both sides of AQE's comparison come out of the map
statuses, so a test that fixes the target can't see the threshold move. The
replacement computes max(skewedPartitionThresholdInBytes, skewedPartitionFactor
* medianReportedSize) from the statuses themselves and requires every skewed
reducer to clear it, over three shapes: your A, your B, and the
101-tie/many-mapper case. It fails on d3be2ba389d with
106662602 was not greater than 286965600 reducer 1881 holds 524288000 bytes
but is reported as 106662602, below the AQE skew threshold of 286965600, so it
would never be split
One thing that fell out of writing it: the previous rotation test used
2,000 map tasks and 100 KiB tied blocks, so its tied reducers held 204,800,000
bytes — below the 256 MiB floor. They'd never be split even if reported
perfectly, so that test was never measuring AQE visibility at all. The new case
uses 5,000 map tasks.
--
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]
