(celeborn) branch main updated: [CELEBORN-2146] Setting the DFS replication factor for balanced fault tolerance and storage efficiency

[nicholasjiang]

This is an automated email from the ASF dual-hosted git repository.

nicholasjiang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git


The following commit(s) were added to refs/heads/main by this push:
     new 7fc4f241f [CELEBORN-2146] Setting the DFS replication factor for 
balanced fault tolerance and storage efficiency
7fc4f241f is described below

commit 7fc4f241f792774190c7724ae716345603c65d35
Author: xxx <953396...@qq.com>
AuthorDate: Sat Sep 20 15:20:50 2025 +0800

    [CELEBORN-2146] Setting the DFS replication factor for balanced fault 
tolerance and storage efficiency
    
    …erance and storage efficiency
    
    ### What changes were proposed in this pull request?
    
    Set the DFS replication factor for balanced fault tolerance and storage 
efficiency.
    
    ### Why are the changes needed?
    
    Setting replication factor is to determine the number of redundant copies 
of data that should be maintained across multiple nodes, ensuring fault 
tolerance and data availability in case of node failures.
    
    ### Does this PR introduce _any_ user-facing change?
    
    NO
    
    ### How was this patch tested?
    
    CI
    
    Closes #3472 from xy2953396112/CELEBORN-2146.
    
    Authored-by: xxx <953396...@qq.com>
    Signed-off-by: SteNicholas <programg...@163.com>
---
 .../main/scala/org/apache/celeborn/common/CelebornConf.scala  | 11 +++++++++++
 docs/configuration/worker.md                                  |  1 +
 .../celeborn/service/deploy/worker/storage/TierWriter.scala   |  4 ++++
 3 files changed, 16 insertions(+)

diff --git 
a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala 
b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 19cc15982..9b2d044aa 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -690,6 +690,9 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable 
with Logging with Se
     get(MASTER_RESOURCE_CONSUMPTION_METRICS_ENABLED)
   def workerFlushReuseCopyBufferEnabled: Boolean =
     get(WORKER_FLUSH_REUSE_COPY_BUFFER_ENABLED)
+  def workerDfsReplicationFactor: Int =
+    get(WORKER_DFS_REPLICATION_FACTOR)
+
   def clusterName: String = get(CLUSTER_NAME)
 
   // //////////////////////////////////////////////////////
@@ -6687,4 +6690,12 @@ object CelebornConf extends Logging {
       .booleanConf
       .createWithDefaultString("true")
 
+  val WORKER_DFS_REPLICATION_FACTOR: ConfigEntry[Int] =
+    buildConf("celeborn.worker.hdfs.replication.factor")
+      .categories("worker")
+      .version("0.7.0")
+      .doc("HDFS replication factor for shuffle files.")
+      .intConf
+      .createWithDefault(2)
+
 }
diff --git a/docs/configuration/worker.md b/docs/configuration/worker.md
index 72343249e..4c688fff3 100644
--- a/docs/configuration/worker.md
+++ b/docs/configuration/worker.md
@@ -106,6 +106,7 @@ license: |
 | celeborn.worker.graceful.shutdown.saveCommittedFileInfo.interval | 5s | 
false | Interval for a Celeborn worker to flush committed file infos into Level 
DB. | 0.3.1 |  | 
 | celeborn.worker.graceful.shutdown.saveCommittedFileInfo.sync | false | false 
| Whether to call sync method to save committed file infos into Level DB to 
handle OS crash. | 0.3.1 |  | 
 | celeborn.worker.graceful.shutdown.timeout | 600s | false | The worker's 
graceful shutdown timeout time. | 0.2.0 |  | 
+| celeborn.worker.hdfs.replication.factor | 2 | false | HDFS replication 
factor for shuffle files. | 0.7.0 |  | 
 | celeborn.worker.http.auth.administers |  | false | A comma-separated list of 
users who have admin privileges, Note, when 
celeborn.worker.http.auth.supportedSchemes is not set, everyone is treated as 
administrator. | 0.6.0 |  | 
 | celeborn.worker.http.auth.basic.provider | 
org.apache.celeborn.common.authentication.AnonymousAuthenticationProviderImpl | 
false | User-defined password authentication implementation of 
org.apache.celeborn.common.authentication.PasswdAuthenticationProvider | 0.6.0 
|  | 
 | celeborn.worker.http.auth.bearer.provider | 
org.apache.celeborn.common.authentication.AnonymousAuthenticationProviderImpl | 
false | User-defined token authentication implementation of 
org.apache.celeborn.common.authentication.TokenAuthenticationProvider | 0.6.0 | 
 | 
diff --git 
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/TierWriter.scala
 
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/TierWriter.scala
index f4b5e19f9..93a6c2989 100644
--- 
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/TierWriter.scala
+++ 
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/TierWriter.scala
@@ -536,6 +536,7 @@ class DfsTierWriter(
 
   try {
     hadoopFs.create(dfsFileInfo.getDfsPath, true).close()
+    hadoopFs.setReplication(dfsFileInfo.getDfsPath, 
conf.workerDfsReplicationFactor.toShort);
     if (dfsFileInfo.isS3) {
       val uri = hadoopFs.getUri
       val bucketName = uri.getHost
@@ -654,6 +655,9 @@ class DfsTierWriter(
       hadoopFs.create(dfsFileInfo.getDfsWriterSuccessPath).close()
       if (dfsFileInfo.isReduceFileMeta) {
         val indexOutputStream = hadoopFs.create(dfsFileInfo.getDfsIndexPath)
+        hadoopFs.setReplication(
+          dfsFileInfo.getDfsIndexPath,
+          conf.workerDfsReplicationFactor.toShort)
         val byteStream: ByteArrayOutputStream = new ByteArrayOutputStream()
         val dataStream = new DataOutputStream(byteStream)
         try {