Apache Celeborn Documentation

Celeborn provides two primary methods for monitoring the cluster state and performance:

1. Prometheus metrics: For time-series data collection and alerting.
2. REST API: For programmatic access to cluster information and status.

When fetching chunks from a data file, the ShuffleClient manages failures through a retry mechanism:

• Retry Limit: The ShuffleClient has a maximum number of retries per replica (defaults to 3).
• Replica Switching: If a fetch chunk fails, the ShuffleClient attempts to try another replica. If replication is disabled, it retries the same replica.
• Failure State: If the maximum number of retries is exceeded, the ShuffleClient stops retrying and throws an Exception.

Celeborn Workers perform periodic checks on disk health and usage to ensure stability:

• Health Checks: If a disk health check fails, the Worker isolates that disk and will not allocate slots on it until it returns to a healthy state.
• Capacity Threshold: If usable space falls below a specific threshold (defaults to 5GiB), the Worker will stop allocating slots on that disk.
• Preventing Overflow: To prevent exceeding available space, the Worker triggers a HARD_SPLIT for all PartitionLocations on the disk to prevent further file size growth.

Dynamic configurations are applied at different levels. When multiple levels define the same key, the following order of precedence applies (from highest to lowest):

1. TENANT_USER: Specific to a tenantId and a username. Overrides TENANT and SYSTEM levels.
2. TENANT: Specific to a tenantId. Overrides SYSTEM level.
3. SYSTEM: Global system-wide configuration. Overrides static CelebornConf.
4. Static Configuration: The base configuration defined in CelebornConf.

The shuffle lifecycle follows these steps:

1. Mappers lazily ask the LifecycleManager to registerShuffle.
2. LifecycleManager requests slots from the Master.
3. Workers reserve slots and create corresponding files.
4. Mappers retrieve worker locations from the LifecycleManager.
5. Mappers push data to the specified workers.
6. Workers merge and replicate data to their peers.
7. Workers periodically flush data to disk.
8. Mapper tasks complete and trigger a MapperEnd event.
9. Once all mapper tasks are complete, workers commit the files.
10. Reducers request file locations.
11. Reducers read the shuffle data.

Celeborn uses a logical concept called a Slot to achieve load balancing across workers. A Slot represents the capacity of a Celeborn Worker to hold partitions.

• Slot Count Calculation: The number of slots per worker is determined by total usable disk size / average shuffle file size.
• Lifecycle: A worker's slot count decreases when a partition is allocated and increments when a partition is freed.

TagsQL provides enhanced flexibility for selecting workers based on key-value pairs. To use it, set celeborn.tags.useTagsQL to true in the Master configuration.

Syntax Rules:

• Match single value: key:value
• Negate single value: key:!value
• Match list of values: key:{value1,value2}
• Negate list of values: key:!{value1,value2}

Note: TagsQL only supports tags where key-value pairs are separated by an equal sign (=) in the underlying data.

Example: env:production region:{us-east,us-west} env:!sandbox selects workers where env is production, region is either us-east or us-west, and env is NOT sandbox.

celeborn.tags.useTagsQL=true
# Example expression:
env:production region:{us-east,us-west} env:!sandbox

The Celeborn Master is the central coordinator of the cluster. Its primary responsibilities include:

• Cluster Status Management: Tracking the health and availability of all Worker nodes.
• Shuffle Lifecycle Management: Maintaining active shuffles and cleaning up resources when applications fail.
• High Availability (HA): Ensuring the Master component remains resilient using the Raft consensus protocol.
• Slot Allocation: Distributing shuffle partition locations to available disks across the cluster using load-balancing strategies.

While the Celeborn server is compatible with various engine clients, the Celeborn client must match the version of the engine you are running.

For example, if you are using Spark 3.2, you must compile the Celeborn client using the -Pspark-3.2 flag.

Celeborn supports asynchronous pushing via the DataPusher component to ensure that the compute engine's execution is not blocked by shuffle I/O.

Workflow:

1. The compute engine calls DataPusher#addTask.
2. A PushTask containing the data is created and added to a non-blocking queue.
3. DataPusher continuously polls the queue and invokes ShuffleClient#pushData to perform the actual transfer.