ftong2010 opened a new issue, #3849:
URL: https://github.com/apache/celeborn/issues/3849
### Is your feature request related to a problem? Please describe.
Shuffle data volume across production workloads is heavily **bimodal** in
our production environment:
- **~99% of workloads shuffle less than 10TB** — modest, short-lived staging
needs;
- **~1% of workloads shuffle more than 40TB** — monthly/yearly aggregation,
large joins,
training intermediates.
Provisioning shuffle storage for that distribution with today's options
forces one of three bad trades:
| Option | What happens |
|---|---|
| **Pure SSD/HDD sized for the tail** | Cluster dimensioned for the 1%; the
other 99% of the time the capacity sits idle. Over-provisioning at this ratio
is very high financial waste, and local disk provides **no elasticity at all**
when a larger-than-planned job lands anyway. |
| **S3/HDFS-only** | Every write pays remote-store latency in the
mapper-critical path — sub-optimal performance — and on public cloud the
sustained concurrent-PUT shuffle pattern is **heavily throttled** by
object-store rate limits. |
| **Mixed tiering (status quo)** | Files routed to the remote tier still
write synchronously through it; throttling and latency land directly on
mappers, and capacity planning doesn't actually improve. |
Official document states that celeborn supports s3 as storage backend, and
user can specify s3 in "availableStorageType" following memory and ssd/hdd
disks, which looks like a sequence based tiering solution. But in fact that is
**misleading**: s3/hdfs can only be combined with memory, and it is
exclusive againinst local disks. Only SSD/HDD exists in the list, DFS tiering
no longer works.
What the distribution calls for is **tiering by time, not by file**: write
local first (fast, unthrottled, failure-isolated), then asynchronously move
*sealed* (immutable, map-side-complete) files to object storage and free the
local disk under pressure. Local SSD becomes a cache whose high-water mark —
not its size —
defines worker capacity: the 99% run entirely on fast local disk; the 1%
spill cold data to S3 with no operator intervention.
This design also protects against data is written unbound to disk, as
currently "disk capacity" configuration in "celeborn.worker.storage.dirs" does
not stop client from writing when capacity is reached. It will force partition
splitting but big job will eventually find a way keeping writing data to
celeborn workers, until disk is full or node is evicted by disk pressure in
kubernetes environment.
### Describe the solution you'd like
**1. Seal barrier — hard split as the immutability point.** A file is
sealable when no mapper can ever append again; hard split already provides
this (epoch bump, mappers revived away, late pushes rejected). Enforce with
`trySealForOffload()` (write-lock; post-seal pushes get HARD_SPLIT) plus
`sealFlushAndWait()` (drain flush buffer and wait for in-flight flushes so
the bytes are final — we hit and fixed a flushBuffer-tail truncation bug
here). Sealed files enter an offload queue; the unsealed write path is
untouched.
**2. Budget-controlled background offloader.** A ~1.5s scanner submits
sealed files under:
gate = hysteresis latch on local usage (start / stop watermarks)
budget = (usage - stopWatermark) × totalBytes - inFlightBytes
(stop=0 → drain-while-open)
selection= FIFO by seal time, submit until budget exhausted
(one overshoot file allowed — files are atomic)
Uploads are multipart with part-level parallelism (`partConcurrency`,
default 4; memory bounded by threads×concurrency×8m = 64 MiB at defaults),
running on a small bounded pool — so **public-cloud throttling slows
draining but can never stall pushes**, the inverse of the S3-tier behavior.
Failures retry with backoff and ultimately **fall back to keeping the file
local** (data is never at risk). On success, an atomic registry maps
committed file → S3 path, the local file is deleted, and a shuffle-unregister
hook cleans both local dirs and the S3 prefix.
**3. Worker-proxied reads.** Reducers keep reading from the worker whether
the file is local or already in S3, via `S3ChunkBuffers` in the normal
fetch/OpenStream path, supporting **full-chunk and map-range reads** (AQE
skew-join sub-tasks read map ranges; S3 range GETs make those cheap). One
upstream fix needed: `S3AInputStream` can't back the sorter's ByteBuffer
reads, requiring a heap-array fallback. **Clients need zero S3
credentials, config, or jars** — object-store topology stays worker-side.
**4. Configuration.**
| key | default | meaning |
|---|---|---|
| `celeborn.worker.storage.offload.enabled` | false | master switch |
| `...offload.threads` | 2 | concurrent file uploads |
| `...offload.partConcurrency` | 4 | MPU parts per file (mem bound
threads×c×8m) |
| `...offload.settleDelay` | 3s | quiescence wait after seal |
| `...offload.diskUsageThreshold` | 0.0 | start watermark (0 = always
active) |
| `...offload.diskUsageStopThreshold` | 0.0 | stop watermark → byte budget
(0 = drain) |
| `...offload.maxAttempts` / `retryBackoff` | 5 / 30s | per-file retry
policy |
**Measured** (single worker, 10G SSD cache, S3-compatible store):
20G shuffle drained through a 10G cache (20 uploads, 0 failures, 0 fetch
failures; usage oscillated near the watermark instead of hitting
"No available disks"); gate-open/gate-closed drills show the control plane
never interferes with the data plane (152 splits, 0 offloads when closed);
AQE skew-join range reads over offloaded files verified end-to-end; all via
`FetchS3Chunk*` metrics with zero client-side configuration.
### Describe alternatives you've considered
- **Pure SSD/HDD, over-provisioned**: no elasticity; sizing for 40TB
outliers that appear 1% of the time is financially untenable.
- **S3/HDFS-only / DFS-as-tier (status quo)**: synchronous remote writes in
the push path put latency and throttling on mappers; every file pays
remote costs rather than only cold ones. The docs should also be amended
to state object-store throughput/rate-limit realities.
- **Client-side direct S3 reads**: ships credentials and storage topology
into every executor's config/jars, and requires per-engine integration;
rejected.
- **Mount S3 as a local filesystem (s3fs / goofys / Mountpoint for S3) and
register it alongside real SSDs, i.e. mixing SSD/HDD disk types in
`celeborn.worker.storage.dirs`**: the S3 mount is declared as an
HDD-type dir and Celeborn's existing multi-disk allocation spreads
writers across both. This does **not** satisfy tiering: dir selection is
capacity/slot-based, not temperature-based — new writers and
hard-split generations are placed on whichever dir has slots, so hot
in-flight files routinely land on the S3 mount and inherit FUSE context
switches, network round-trips, and object-store throttling directly in
the push/flush path, with non-POSIX fsync/rename semantics adding jitter
to commit. Reads fare no better: millions of small chunk fetches over
the mount each pay object-store GET latency. It also adds real
**operational overhead**: per-node mount lifecycle management, FUSE
daemon monitoring, local cache-dir sizing on the very SSDs you were
trying to offload, credential rotation on every worker host, and
misleading observability (`df`-based usage checks and quota math become
fiction on a mount that reports PBs of free space). Tiering must remain
an explicit, worker-managed boundary — hot data local, cold data remote —
rather than an accident of slot allocation.
- **External lifecycle jobs (cron upload/compaction)**: race with active
shuffles and need cluster-global coordination; the in-worker seal barrier
makes offload safe and simple.
### Future improvement
* Better dedupe handling when replication is enabled.
* Support more evict policy
* Read data back from s3 when data is loaded and disk pressure is gone
* Throttle client push using current mechanism when disk pressure is high,
since s3 write throughput is generally slower than ssds aggregated on worker
cluster.
I am wondering if community is interested in this feature. we can submit PR
for this feature.
It is currently tested in our production environment and will report and new
finding if anything comes up.
/cc @who-need-to-know
/assign @who-can-help-you
--
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]
