Apache Celeborn: a shuffle service separated from the compute engine

What Celeborn moves off the cluster's local disks

Celeborn is an Apache project for the shuffle stage of distributed computation. The README describes it as a service for improving the efficiency and elasticity of map-reduce engines, providing an elastic, high-efficient management service for intermediate data including shuffle data, spilled data and result data, with the current focus on shuffle data.

The architectural claim in its feature list is disaggregated computing and storage, paired with push-based shuffle write and merged shuffle read, and high availability and high fault tolerance. Read together, those three mean the storage for intermediate data is separated from the executors that produce and consume it, the writers push rather than pull, readers merge from multiple workers instead of fetching one file, and the service is expected to survive node loss.

That separation is the whole design decision. It buys elasticity, since storage capacity is not tied to the executor that happens to hold the data, and it costs a network hop and a service to operate. A cluster whose shuffle is not the bottleneck will not get much from it.

Master, Worker, Client and the LifecycleManager

Celeborn has three primary components plus one piece that lives inside the engine driver.

The Master manages all resources and syncs shared state based on Raft, so the cluster metadata itself has a consensus mechanism behind it. Workers process read and write requests and merge data for each reducer, which is where the shuffle files live. The Client is the interface the engine uses to reach those workers.

The fourth piece is the LifecycleManager, which maintains the metadata of each shuffle and runs within the Spark driver. Keeping that state in the driver rather than in the Master is a meaningful choice: the driver knows which shuffles exist for the job it is running, and the Master stays a cluster-level resource manager rather than a per-job metadata store.

The high availability claim therefore rests on the Master syncing shared state through Raft and on workers replicating data to peers. When a mapper finishes, the workers commit their files, and only then do reducers ask for file locations, which is what keeps a reducer from reading a file that is still being written.

The shuffle, from registration to reducer read

The README lists the shuffle sequence in eleven steps, and the order explains the design.

Mappers lazily ask the LifecycleManager to registerShuffle, which means nothing is allocated until there is data to shuffle. The LifecycleManager then requests slots from the Master, workers reserve those slots and create the corresponding files, and the mappers receive worker locations from the LifecycleManager. The mappers push data to the specified workers, the workers merge and replicate data to their peer, and they flush to disk periodically.

When the mapper tasks are done they trigger the MapperEnd event, and once all mapper tasks are complete the workers commit the files. Only then do reducers ask for file locations and read the shuffle data.

Two details deserve attention. The write path is push-based, so a mapper writes to a remote worker rather than pulling later, which moves cost to the write side and lets the worker merge several mappers' output into one file. And the commit step is a barrier: reducers cannot read locations before the files are committed, so a slow final mapper delays the reducer stage rather than producing partial reads.

Slots: how Celeborn decides where partitions land

Load balancing in Celeborn is expressed in slots. The Slot is a logical concept in a Celeborn Worker that represents how many partitions can be allocated to each worker, and partitions are distributed equally on every worker by tracking slot usage.

The slot count is derived from disk rather than from CPU, and the README gives the formula as total usable disk size divided by average shuffle file size. The count decreases when a partition is allocated and increments when one is freed, which makes the accounting a counter rather than a recomputation.

This is a deliberate simplification. A worker with faster disks or faster network is treated the same as one with slower ones, and the model assumes shuffle files are of similar size across the job. In practice that assumption holds for evenly sized partitions, and it stops holding for skewed data, where one partition may be many times the average and can consume a worker's disk without consuming a proportionate number of slots. If your job has skewed keys, that is the first thing to check after a cluster lands a partition on a worker that runs out of space.

Building a distribution and the engine profiles

Celeborn is built with a single script that takes a profile per engine. The README lists support for Spark 3.0 through 4.2, Flink 1.18 through 2.3 and Hadoop MapReduce 3, tested under Scala 2.11, 2.12 and 2.13 and Java 8, 11 and 17.

The build command selects the engine profile, and the output is a compressed package named apache-celeborn-${project.version}-bin.tgz. For example, compiling for Spark 3.5 with Java 21 uses:

./build/make-distribution.sh -Pspark-3.5 -Pjdk-21

A second variant adds --sbt-enabled, which selects the sbt build path. Storage backends have their own flags: AWS S3 multipart upload support is enabled with the aws profile and Aliyun OSS with the aliyun profile, in both cases with an explicit engine and JDK combination:

./build/make-distribution.sh --sbt-enabled -Pspark-3.4 -Pjdk-8 -Paws

There is also an aarch64 profile for Linux on ARM, which the README qualifies with a specific requirement: it needs glibc version 3.4.21, and other glibc versions such as 2.x can hit a problematic frame described as C [libc.so.6+0x8412a]. That is the kind of detail that only matters on ARM infrastructure, and the kind that costs an afternoon if nobody has written it down.

The compatibility matrix is the real specification

The table of engine, Java and Scala combinations is the most operationally useful part of the README, and it narrows faster than the headline support list suggests.

For Spark 3.0 to 3.5, the supported combinations run from Java 8 with Scala 2.12 through Java 17 with Scala 2.13, with a few cells marked unsupported, such as Spark 3.0 with Java 17 and Scala 2.13. For Spark 4.0, 4.1 and 4.2, only the Java 17 with Scala 2.13 column is available. The Flink rows show the same narrowing for Flink 2.0 and above, where Java 11 and Java 17 with Scala 2.13 are supported and the Java 8 column is not.

Two conclusions follow. A Spark 4 cluster must run on Java 17 and Scala 2.13, so the upgrade path for the engine also constrains the runtime. And Flink 2.x requires Java 11 or later, so a Flink 1.x deployment on Java 8 is a different build target from a Flink 2.x deployment on the same hardware.

The client side follows the same versioning logic. The repository has separate client-spark, client-flink, client-mr and client-tez directories alongside the master, worker, common and lifecycle-manager modules, so each engine integration is a separate artefact that has to be matched to the engine you run.

A codebase that reaches beyond the JVM

The tree shows a much wider project than the Java modules imply. Alongside client/, common/, master/, worker/, lifecycle-manager/ and the four engine clients, there is a rust/ directory, a cpp/ directory, a web/ directory, an openapi/ directory, a cli/, a charts/ directory with deployment manifests, a docker/ directory, a multipart-uploader/ for object storage transfers, a toolkit/, dev/, service/, spi/ and tests/.

The multipart-uploader/ and the aws and aliyun build profiles are two views of the same concern: shuffle data ultimately has to land in object storage for long-lived jobs, and doing that efficiently needs a component that can push parts in parallel. The rust/ and cpp/ directories point at the same idea from the other direction, native code where the JVM is the bottleneck.

Documentation has its own toolchain. The root Makefile builds the MkDocs site inside a python:3.11-slim container, with a pinned requirements.txt holding mkdocs, mkdocs-material and the plugin set, and a docs target that builds the site:

docs: check-docker ## Build the MkDocs site inside a Python 3.11 container

The project is Apache-2.0 licensed with the usual LICENSE, NOTICE and LICENSE-binary files, which is what a foundation project expects when it ships third-party binary dependencies.

Editorial conclusion

Celeborn suits a cluster team whose shuffle stage is the bottleneck and who can afford to operate three extra services, a Master, a set of Workers and a client-side lifecycle manager. It is the wrong choice if shuffle data is not your problem, since the project itself says the current focus is shuffle data, with spilled and result data still ahead of it. It is also the wrong choice if your cluster runs an unsupported combination, because the compatibility matrix narrows sharply for Spark 4.0 and above, which require Scala 2.13 and Java 17. Verify first that your Spark, Flink, Scala and Java versions appear in the table, that the client you deploy matches the engine version, and that the disk space you give each worker matches the slot count the load balancer will assign.

Frequently asked questions

Official sources

1. apache/celeborn on GitHub
2. License: Apache-2.0
3. Project website
4. README
5. Releases