Engine Integration

Apache Spark

• Deploy Client JAR

cp $CELEBORN_HOME/spark/celeborn-client-spark-3-shaded_2.12-0.6.2.jar $SPARK_HOME/jars/

Core Spark Configuration

# Core settings

spark.shuffle.manager=org.apache.spark.shuffle.celeborn.SparkShuffleManager
spark.celeborn.master.endpoints=master1:9097,master2:9097,master3:9097
spark.shuffle.service.enabled=false
spark.serializer=org.apache.spark.serializer.KryoSerializer

# Performance tuning

spark.celeborn.client.push.replicate.enabled=true
spark.celeborn.client.spark.shuffle.writer=hash
spark.sql.adaptive.localShuffleReader.enabled=false
# Dynamic allocation (Spark 3.5+)

spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.shuffleTracking.enabled=false
spark.shuffle.sort.io.plugin.class=org.apache.spark.shuffle.celeborn.CelebornShuffleDataIO

• Quick Test with spark-shell

/usr/odp/3.3.6.4-1/spark3/bin/spark-shell --conf spark.shuffle.manager=org.apache.spark.shuffle.celeborn.SparkShuffleManager --conf spark.shuffle.service.enabled=false --conf spark.sql.adaptive.localShuffleReader.enabled=false --conf spark.dynamicAllocation.shuffleTracking.enabled=false

---

Apache Flink

• Deploy Client JAR.

cp $CELEBORN_HOME/flink/celeborn-client-flink-1.19-shaded_2.12-0.6.2.jar $FLINK_HOME/lib/

• Configure Flink.

shuffle-service-factory.class=org.apache.celeborn.plugin.flink.RemoteShuffleServiceFactory
execution.batch-shuffle-mode=ALL_EXCHANGES_BLOCKING
celeborn.master.endpoints=master1:9097,master2:9097,master3:9097

# Performance

celeborn.client.shuffle.batchHandleReleasePartition.enabled=true
celeborn.client.push.maxReqsInFlight=128
celeborn.data.io.numConnectionsPerPeer=16
celeborn.data.io.threads=32

---

Apache Tez

Append following property in yarn-site xml under yarn.application.classpath and under Mapreduce mapreduce.application.classpath.

/usr/odp/3.3.6.4-1/celeborn/tez/celeborn-client-tez-shaded_2.12-0.6.2.3.3.6.4-1.jar:/usr/odp/3.3.6.4-1/celeborn/mr/celeborn-client-mr-shaded_2.12-0.6.2.3.3.6.4-1.jar

# In tez-site.xml

tez.am.launch.cluster-default.cmd-opts=org.apache.tez.dag.app.CelebornDagAppMaster

tez.celeborn.master.endpoints=master1:9097,master2:9097,master3:9097

tez.cluster.additional.classpath.prefix append /usr/odp/3.3.6.3-101/celeborn/tez/celeborn-client-tez-shaded_2.12-0.6.2.3.3.6.3-101.jar

---

MapReduce

Need to add above tez configuration for classpath addition and then following additional properties in Mapreduce configuration.

yarn.app.mapreduce.am.command-opts=org.apache.celeborn.mapreduce.v2.app.MRAppMasterWithCeleborn

mapreduce.job.reduce.slowstart.completedmaps=1

mapreduce.celeborn.master.endpoints=master1:9097,master2:9097,master3:9097
mapreduce.job.map.output.collector.class=org.apache.hadoop.mapred.CelebornMapOutputCollector
mapreduce.job.reduce.shuffle.consumer.plugin.class=org.apache.hadoop.mapreduce.task.reduce.CelebornShuffleConsumer
yarn.app.mapreduce.am.job.recovery.enable=false

---

High Availability

Master HA Overview

Celeborn Master achieves HA using Apache Ratis (Raft consensus protocol). A minimum of 3 Master nodes is required. Odd numbers (3, 5, or 7) are required so the cluster can elect a leader by majority vote.

HA Requirement

Minimum: 3 Master nodes. Always use odd numbers (3, 5, 7).

Ratis storage: /var/lib/celeborn/ratis — ensure this path has adequate I/O performance.

Worker fault tolerance: Enable replication (spark.celeborn.client.push.replicate.enabled=true).

Master HA Configuration

# Enable HA mode
celeborn.master.ha.enabled=true
# Node 1
celeborn.master.ha.node.1.host=master1.example.com
celeborn.master.ha.node.1.port=9097
celeborn.master.ha.node.1.ratis.port=9872
# Node 2
celeborn.master.ha.node.2.host=master2.example.com
celeborn.master.ha.node.2.port=9097
celeborn.master.ha.node.2.ratis.port=9872
# Node 3
celeborn.master.ha.node.3.host=master3.example.com
celeborn.master.ha.node.3.port=9097
celeborn.master.ha.node.3.ratis.port=9872
# Ratis metadata storage
celeborn.master.ha.ratis.raft.server.storage.dir=/var/lib/celeborn/ratis

---

Ambari Mpack HA Settings

---

Worker Fault Tolerance

Enable data replication on the Spark client to ensure shuffle data survives a worker node failure:

spark.celeborn.client.push.replicate.enabled=true