apache / apache/datafusion-ballista
[Feature] Support external Remote Shuffle Service (e.g., Apache Celeborn / Apache Uniffle)
- Dominant language
- Rust
- Stars
- 2.1k
- Forks
- 320
- Avg merge
- 1d 22h
- Merged PRs (30d)
- 66
Description
## Is your feature request related to a problem or challenge?
Ballista currently stores shuffle data on local executor disks and serves it via Arrow Flight between executors. This creates several limitations:
- **Fault tolerance**: If an executor crashes, its shuffle data is lost and the entire job must restart from the beginning.
- **Resource coupling**: Compute (executor CPU/memory) and shuffle storage are tightly coupled. Executors must remain alive for the full duration of the job to serve shuffle data to downstream stages.
- **Scalability**: With M map tasks and N reduce tasks, executors must hold shuffle data for all downstream consumers, which limits independent scaling of compute vs. shuffle storage.
- **Elasticity**: Executors cannot be reclaimed or scaled down between stages because they are still serving shuffle data.
## Describe the solution you'd like
Add support for an **external Remote Shuffle Service (RSS)** as a pluggable shuffle storage and serving layer. Instead of writing shuffle data to local executor disks and serving it via Flight, executors would push shuffle data to a dedicated external service, and reduce tasks would pull from it.
Popular open-source RSS implementations that could serve as integration targets:
- **[Apache Celeborn](https://celeborn.apache.org/)** (formerly Alibaba RSS) — push-based, supports Spark/Flink/MR
- **[Apache Uniffle](https://uniffle.apache.org/)** (formerly Tencent RSS) — supports multiple compute engines
- **[Riffle](https://github.com/zuston/riffle)** — a Rust-native RSS implementation that may be particularly interesting given Ballista's Rust ecosystem
### High-level design ideas
1. **Pluggable shuffle writer**: Introduce a `ShuffleWriter` trait/abstraction so the current local-disk writer and a new RSS writer can be swapped via configuration.
2. **RSS push on map completion**: After a map stage task finishes, shuffle blocks are pushed to the RSS instead of being left on local disk.
3. **RSS pull on reduce**: The shuffle reader fetches partitions from the RSS endpoint rather than connecting back to the producing executor via Flight.
4. **Scheduler awareness**: The scheduler should not need to track partition locations per-executor; instead, it queries the RSS for partition locations.
### Benefits
- **Better fault tolerance**: Shuffle data survives executor failures.
- **Executor elasticity**: Executors can be released after map stages complete.
- **Decoupled scaling**: Shuffle storage and compute can scale independently.
- **Reduced network pressure on executors**: Dedicated shuffle servers can handle merging and serving more efficiently.
Is there community interest in this direction? Happy to help with design, prototyping, or integration work.
Contributor guide
Research direction
Start by tracing Ballista’s current local-disk shuffle path, Arrow Flight serving between executors, and scheduler handling of partition locations. Define the scope and acceptance criteria for a pluggable writer, RSS push and pull flows, and executor release after map stages; validate the design against a concrete target such as Celeborn, Uniffle, or Riffle.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100