E-MapReduce:Celeborn

How Celeborn works

The standard shuffle implementation has several limitations:

• Write amplification: Large shuffle write tasks can cause data overflow, leading to write amplification.

• Connection reset: Shuffle read tasks generate many small network packets, which can trigger connection resets.

• High disk and CPU load: Shuffle read tasks produce many small I/O requests and random reads, increasing disk and CPU load.

• Connection explosion: With thousands of mappers (M) and reducers (N), the number of connections grows to M x N, making jobs difficult to run.

• YARN instability: When shuffle data volume is extremely large, NodeManager restarts, disrupting YARN-based task scheduling.

Celeborn addresses these issues by using a push-based shuffle model:

• Reduces mapper memory pressure by using push-based shuffle instead of pull-based shuffle.

• Supports I/O aggregation, reducing read-side connections from M x N to N and converting random reads to sequential reads.

• Uses a two-replica mechanism to reduce fetch failures.

• Supports compute-storage separation, so the shuffle service can run on dedicated hardware.

• Eliminates dependency on local disks for Spark on Kubernetes workloads.

The following figure shows the Celeborn architecture.

Prerequisites

Before you begin, ensure that you have:

• An E-MapReduce (EMR) DataLake cluster or a custom cluster with the Celeborn service selected. For more information, see Create a cluster.

Supported versions

Celeborn is supported on the following cluster types and versions:

Enable Celeborn for Spark

You can enable Celeborn for Spark in two ways: through the EMR console with one click (recommended), or by manually setting Spark parameters.

# Shuffle manager — class name changed in Celeborn 0.4.0:
# Celeborn 0.3.x or earlier: org.apache.spark.shuffle.celeborn.RssShuffleManager
# Celeborn 0.4.x or later:  org.apache.spark.shuffle.celeborn.SparkShuffleManager
spark.shuffle.manager org.apache.spark.shuffle.celeborn.SparkShuffleManager

# Kryo serializer is required — Java serializer does not support relocation
spark.serializer org.apache.spark.serializer.KryoSerializer

# Disable the external shuffle service — required to use Celeborn
# Note: Setting this to true means Celeborn is NOT used.
# Celeborn does not affect Spark dynamic allocation.
spark.shuffle.service.enabled false

# Shuffle writer mode:
#   hash (default): higher memory usage when partition count is large
#   sort: fixed memory usage, better for workloads with many partitions
spark.celeborn.shuffle.writer hash

# Two-replica replication — reduces fetch failures
spark.celeborn.push.replicate.enabled true

# Celeborn master endpoint
# Format: <celeborn-master-ip>:9097 (use the public IP address of the master node)
# For high-availability clusters, specify all master node IPs
spark.celeborn.master.endpoints <celeborn-master-ip>:9097

# Adaptive Query Execution (AQE) settings
# Disable local shuffle reader for better Celeborn performance
spark.sql.adaptive.enabled true
spark.sql.adaptive.localShuffleReader.enabled false
spark.sql.adaptive.skewJoin.enabled true

Configure Celeborn parameters

View and modify Celeborn parameters on the Configure tab of the Celeborn service page.

Restart the CelebornMaster component

After modifying Celeborn parameters, restart the CelebornMaster component to apply the changes.

1. On the Status tab of the Celeborn service page, find the CelebornMaster component. In the Actions column, move the pointer over the icon and select restart_clean_meta.

   Note

   For non-high-availability clusters, click Restart in the Actions column instead.

2. In the dialog box, turn off Rolling Execution, set the Execution Reason, and click OK.

3. In the confirmation dialog box, click OK.