Lobo2008 opened a new pull request, #3762:
URL: https://github.com/apache/celeborn/pull/3762
### What is this PR for?
Add configurable HDFS remote spill support for MR reduce merge tasks,
allowing
the reduce merge phase to spill intermediate data to HDFS instead of NM local
disk, enabling storage-compute separation.
### New config keys
| Key | Default | Description |
|-----|---------|-------------|
| `celeborn.client.mr.remote.spill.enabled` | `false` | Enable HDFS remote
spill for MR reduce merge |
| `celeborn.client.mr.remote.spill.path` | (none) | HDFS directory for spill
files. Must be set on both client and worker side. |
### Key changes
- `CelebornRemoteSpillMergeManager` — orchestrates HDFS merge with fallback
- `CelebornInMemoryRemoteMerger` — writes in-memory map outputs to HDFS
- `CelebornShuffleConsumer` — integrates remote spill into reduce shuffle
path
- `HadoopShuffleDeleteHandler` — worker-side cleanup of spilled files
- Security/HDFS utilities — Kerberos-aware HDFS access
- Worker startup validation — fails fast if enabled but path not set
- Unit tests: `HadoopShuffleDeleteHandlerSuiteJ` (4 test cases)
### Design decisions
1. **Why not merge on server side?**
Uniffle already implemented server-side merge for MR ([PR
#2109](https://github.com/apache/uniffle/pull/2109),
part of the umbrella issue
[#1239](https://github.com/apache/uniffle/issues/1239)).
We tested this in production and found significant performance
degradation.
The root cause: `RMRssShuffle` requires all data for a partition to be on
a
**single** shuffle server (`serverInfoSet.size() != 1` throws, see
`client-mr/core/.../RMRssShuffle.java:114`), creating a serial bottleneck
where one server must deserialize, sort, and merge all records.
In contrast, our HDFS spill (approach 1) lets each reduce task
independently
merge its own data in parallel across all NM nodes.
2. **Why `celeborn.client.mr.remote.spill.path` instead of
`celeborn.worker.remote.spill.path`?**
This config is used by both the MR client (write spill files) and the
Celeborn worker (cleanup expired files). A neutral key name accurately
reflects its shared nature.
3. **Why fail-fast on worker startup when enabled but path not set?**
Without this validation, the worker would silently accept disabled cleanup
while the MR client still writes HDFS files, leading to orphaned data.
### Known limitations
1. **Configuration propagation**: Unlike Uniffle which supports automatic
config
propagation between client and server, Celeborn currently requires
`celeborn.client.mr.remote.spill.path` to be manually set on both the MR
client side and the Celeborn worker side. A future improvement would be
to enable the client to read this config from the worker/master, so that
users only need to set `celeborn.client.mr.remote.spill.enabled=true` on
the client side while the path is managed centrally on the worker.
### How should this be tested?
1. Set `celeborn.client.mr.remote.spill.enabled=true`
2. Set `celeborn.client.mr.remote.spill.path=hdfs://namenode/path`
3. Run a MR job with Celeborn shuffle
4. Verify reduce merge spill files go to HDFS (not NM local disk)
5. Verify worker cleanup deletes expired HDFS files after job completion
6. Run `mvn test -Pl common -Dtest=HadoopShuffleDeleteHandlerSuiteJ`
--
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]
