comphead commented on issue #2216:
URL:
https://github.com/apache/datafusion-comet/issues/2216#issuecomment-6643381118
I measured both assumptions on `main` (`7bf572efb7`).
**Setup.** A throwaway Scala probe on `local-cluster[2,4,12288]` (two
executor JVMs with 4 cores each), Spark 4.1.3,
`JavaSerializer` and AQE off. The query joins a 4M-row fact table to a
dimension with a `BROADCAST` hint. Each row of
the table is the median of 4 runs. "On" is Spark's default
`spark.broadcast.compress=true`, and "off" is `false`, which
leaves only Comet's own compression.
| Build side (size after Comet's compression) | Codec | Bytes each executor
fetches, on vs off | Time to broadcast, on → off | Executor read, on → off |
Wall, on → off |
| --- | --- | --- | --- | --- | --- |
| 12M rows: sequential key, random long, random string (444 MiB) | `lz4` |
401.5 vs 445.9 MiB (−10%) | 376 → 139 ms | 410 → 331 ms | 2.87 → 2.10 s |
| same rows (273 MiB) | `zstd` level 1 | 247.5 vs 274.7 MiB (−10%) | 146 →
96 ms | 236 → 206 ms | 1.57 → 1.53 s |
| same rows (284 MiB) | `zstd` level 5 | 262.8 vs 285.6 MiB (−8%) | 923 → 87
ms | 237 ms both | 4.54 → 3.72 s |
| TPC-H-like customer, 3M rows (298 MiB) | `lz4` | 271.9 vs 299.2 MiB (−9%)
| 263 → 96 ms | 289 → 220 ms | 2.02 → 1.76 s |
| the 12M rows in random order (456 MiB) | `lz4` | 457.8 vs 458.2 MiB
(−0.1%) | 361 → 139 ms | 644 → 498 ms | 2.59 → 2.11 s |
"Executor read" is how long an executor takes to get the value of a fresh
broadcast of the same bytes: fetching,
decompressing and deserializing. Running only the torrent layer on the
driver over the same bytes, for `lz4`, `snappy`
and `zstd` levels 1, 3 and 5, gives the same picture. The second pass saves
6 to 11% with every codec when the build
side has a sequential column, and 0% (`lz4`, `snappy`) to 6% (`zstd`) when
it does not.
**Conclusions**
1. The regression is real. The second pass runs on the driver before the
join can start. It costs about 50 ms per
100 MiB with `lz4` and about 300 ms per 100 MiB with `zstd` at level 5,
plus up to 150 ms on each executor with
`lz4`.
2. The second pass is not useless. It shrinks broadcasts with sorted or
sequential columns by about 10%.
Since only the first assumption holds, I don't think a Comet change is worth
it. The natural one is to write the
coalesced broadcast uncompressed and let the torrent layer compress it once.
From the same probe, I estimate that saves
0.2 to 0.25 s of driver time per 300 to 460 MiB broadcast with `lz4` and at
most about 0.6 s with `zstd` level 5. In
exchange, each executor would keep the uncompressed Arrow IPC bytes on heap,
1.1 to 3 times as much as now depending on
the data and codec. Users who want to skip the second pass can already set
`spark.broadcast.compress=false`. The
caveat is that it also stops compressing Spark's own broadcasts. #7492
documents this in the tuning guide and closes
this issue.
--
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]
