Skip to content

[CELEBORN-2151] Support multiple dfs mount point #3481

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
xy2953396112 wants to merge 2 commits into from
Closed

[CELEBORN-2151] Support multiple dfs mount point #3481

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4bc6fff
[CELEBORN-2151] Support multiple dfs mount point
xy2953396112 Aug 26, 2025
171f7a1
fix
xy2953396112 Nov 4, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions client/src/main/java/org/apache/celeborn/client/ShuffleClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
Expand Down Expand Up @@ -56,7 +57,7 @@ public abstract class ShuffleClient {
private static Logger logger = LoggerFactory.getLogger(ShuffleClient.class);
private static volatile ShuffleClient _instance;
private static volatile boolean initialized = false;
private static volatile Map<StorageInfo.Type, FileSystem> hadoopFs;
private static volatile Map<StorageInfo.Type, List<Tuple2<String, FileSystem>>> hadoopFs;
private static LongAdder totalReadCounter = new LongAdder();
private static LongAdder localShuffleReadCounter = new LongAdder();

Expand Down Expand Up @@ -114,7 +115,8 @@ public static ShuffleClient get(
return _instance;
}

public static Map<StorageInfo.Type, FileSystem> getHadoopFs(CelebornConf conf) {
public static Map<StorageInfo.Type, List<Tuple2<String, FileSystem>>> getHadoopFs(
CelebornConf conf) {
if (null == hadoopFs) {
synchronized (ShuffleClient.class) {
if (null == hadoopFs) {
Expand Down
10 changes: 7 additions & 3 deletions client/src/main/java/org/apache/celeborn/client/read/DfsPartitionReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import scala.Tuple2;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import org.apache.commons.lang3.tuple.Pair;
Expand Down Expand Up @@ -101,15 +103,17 @@ public DfsPartitionReader(

this.metricsCallback = metricsCallback;
this.location = location;
List<Tuple2<String, FileSystem>> list = null;
if (location.getStorageInfo() != null
&& location.getStorageInfo().getType() == StorageInfo.Type.S3) {
this.hadoopFs = ShuffleClient.getHadoopFs(conf).get(StorageInfo.Type.S3);
list = ShuffleClient.getHadoopFs(conf).get(StorageInfo.Type.S3);
} else if (location.getStorageInfo() != null
&& location.getStorageInfo().getType() == StorageInfo.Type.OSS) {
this.hadoopFs = ShuffleClient.getHadoopFs(conf).get(StorageInfo.Type.OSS);
list = ShuffleClient.getHadoopFs(conf).get(StorageInfo.Type.OSS);
} else {
this.hadoopFs = ShuffleClient.getHadoopFs(conf).get(StorageInfo.Type.HDFS);
list = ShuffleClient.getHadoopFs(conf).get(StorageInfo.Type.HDFS);
}
this.hadoopFs = list.get(Math.floorMod(location.getUniqueId().hashCode(), list.size()))._2;

long fetchTimeoutMs = conf.clientFetchTimeoutMs();
try {
Expand Down
108 changes: 92 additions & 16 deletions common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ import scala.util.matching.Regex

import io.netty.channel.epoll.Epoll
import io.netty.channel.kqueue.KQueue
import org.apache.commons.lang3.StringUtils

import org.apache.celeborn.common.authentication.AnonymousAuthenticationProviderImpl
import org.apache.celeborn.common.client.{ApplicationInfoProvider, DefaultApplicationInfoProvider}
Expand Down Expand Up @@ -1256,6 +1257,25 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
}.getOrElse("")
}

def s3DirList: Option[Seq[String]] = {
get(S3_MULTI_DIR).map {
s3MultiDir =>
StringUtils.split(s3MultiDir, ",")
val dirList: Seq[String] = StringUtils.split(s3MultiDir, ",").toSeq
val validated = dirList.forall(s3Dir =>
if (!Utils.isS3Path(s3Dir)) {
log.error(s"${S3_MULTI_DIR.key} configuration is wrong $s3Dir. Disable S3 support.")
false
} else {
true
})
(validated, dirList)
} match {
case Some((true, dirList)) => Some(dirList)
case _ => None
}
}

def ossDir: String = {
get(OSS_DIR).map {
ossDir =>
Expand All @@ -1267,28 +1287,55 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
}
}.getOrElse("")
}

def ossDirList: Option[Seq[String]] = {
get(OSS_MULTI_DIR).map {
ossMultiDir =>
val dirList: Seq[String] = StringUtils.split(ossMultiDir, ",").toSeq
val validated = dirList.forall(ossDir =>
if (!Utils.isS3Path(s3Dir)) {
log.error(s"${OSS_MULTI_DIR.key} configuration is wrong $ossDir. Disable S3 support.")
false
} else {
true
})
(validated, dirList)
} match {
case Some((true, dirList)) => Some(dirList)
case _ => None
}
}

def ossEndpoint: String = get(OSS_ENDPOINT).getOrElse("")
def ossAccessKey: String = get(OSS_ACCESS_KEY).getOrElse("")
def ossSecretKey: String = get(OSS_SECRET_KEY).getOrElse("")
def ossIgnoreCredentials: Boolean = get(OSS_IGNORE_CREDENTIALS)

def hdfsDir: String = {
get(HDFS_DIR).map {
hdfsDir =>
if (!Utils.isHdfsPath(hdfsDir)) {
log.error(s"${HDFS_DIR.key} configuration is wrong $hdfsDir. Disable HDFS support.")
""
} else {
hdfsDir
}
}.getOrElse("")
def hdfsDirList: Option[Seq[String]] = {
get(HDFS_MULTI_DIR).map {
hdfsMultiDir =>
StringUtils.split(hdfsMultiDir, ",")
val dirList: Seq[String] = StringUtils.split(hdfsMultiDir, ",").toSeq
val validated = dirList.forall(hdfsDir =>
if (!Utils.isHdfsPath(hdfsDir)) {
log.error(
s"${HDFS_MULTI_DIR.key} configuration is wrong $hdfsDir. Disable HDFS support.")
false
} else {
true
})
(validated, dirList)
} match {
case Some((true, dirList)) => Some(dirList)
case _ => None
}
}

def remoteStorageDirs: Option[Set[(StorageInfo.Type, String)]] = {
def remoteStorageDirs: Option[Set[(StorageInfo.Type, Set[String])]] = {
val supported = Seq(
(StorageInfo.Type.HDFS, HDFS_DIR, Utils.isHdfsPath _),
(StorageInfo.Type.S3, S3_DIR, Utils.isS3Path _),
(StorageInfo.Type.OSS, OSS_DIR, Utils.isOssPath _))
(StorageInfo.Type.HDFS, HDFS_MULTI_DIR, Utils.isHdfsPath _),
(StorageInfo.Type.S3, S3_MULTI_DIR, Utils.isS3Path _),
(StorageInfo.Type.OSS, OSS_MULTI_DIR, Utils.isOssPath _))

val activeStorageTypes =
get(ACTIVE_STORAGE_TYPES).split(",").map(StorageInfo.Type.valueOf).toList
Expand All @@ -1297,8 +1344,10 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
if (!activeStorageTypes.contains(ty)) None
else {
get(e).flatMap { dir =>
if (checker(dir)) Some((ty, dir))
else {
val dirs = StringUtils.split(dir, ",").toSeq
if (dirs.nonEmpty) {
Some(ty, dirs.filter(checker(_)).toSet)
} else {
log.error(s"${e.key} configuration is invalid: $dir. Disabling $ty support.")
None
}
Expand Down Expand Up @@ -3277,6 +3326,15 @@ object CelebornConf extends Logging {
.stringConf
.createOptional

val HDFS_MULTI_DIR: OptionalConfigEntry[String] =
buildConf("celeborn.storage.hdfs.multi.dir")
.withAlternative("celeborn.storage.hdfs.multi.dir")
.categories("worker", "master", "client")
.version("0.6.0")
.doc("HDFS base directory for Celeborn to store shuffle data, supporting multi dir with comma delimiter")
.stringConf
.createOptional

val S3_DIR: OptionalConfigEntry[String] =
buildConf("celeborn.storage.s3.dir")
.categories("worker", "master", "client")
Expand All @@ -3285,6 +3343,15 @@ object CelebornConf extends Logging {
.stringConf
.createOptional

val S3_MULTI_DIR: OptionalConfigEntry[String] =
buildConf("celeborn.storage.s3.multi.dir")
.withAlternative("celeborn.storage.hdfs.multi.dir")
.categories("worker", "master", "client")
.version("0.6.0")
.doc("S3 base directory for Celeborn to store shuffle data, supporting multi dir with comma delimiter")
.stringConf
.createOptional

val S3_ENDPOINT_REGION: OptionalConfigEntry[String] =
buildConf("celeborn.storage.s3.endpoint.region")
.categories("worker", "master", "client")
Expand Down Expand Up @@ -3333,6 +3400,15 @@ object CelebornConf extends Logging {
.stringConf
.createOptional

val OSS_MULTI_DIR: OptionalConfigEntry[String] =
buildConf("celeborn.storage.oss.multi.dir")
.withAlternative("celeborn.storage.hdfs.multi.dir")
.categories("worker", "master", "client")
.version("0.6.0")
.doc("OSS base directory for Celeborn to store shuffle data, supporting multi dir with comma delimiter")
.stringConf
.createOptional

val OSS_SECRET_KEY: OptionalConfigEntry[String] =
buildConf("celeborn.storage.oss.secret.key")
.categories("worker", "master", "client")
Expand Down
42 changes: 33 additions & 9 deletions common/src/main/scala/org/apache/celeborn/common/util/CelebornHadoopUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,8 +37,8 @@ object CelebornHadoopUtils extends Logging {
private[celeborn] def newConfiguration(conf: CelebornConf): Configuration = {
val hadoopConf = new Configuration()
hadoopConf.set("fs.automatic.close", "false")
if (conf.hdfsDir.nonEmpty) {
val path = new Path(conf.hdfsDir)
if (conf.hdfsDirList.nonEmpty) {
val path = new Path(conf.hdfsDirList.get.head)
val scheme = path.toUri.getScheme
val disableCacheName = String.format("fs.%s.impl.disable.cache", scheme)
hadoopConf.set("dfs.replication", "2")
Expand All @@ -52,7 +52,7 @@ object CelebornHadoopUtils extends Logging {
}
}

if (conf.s3Dir.nonEmpty) {
if (conf.s3DirList.nonEmpty) {
if (conf.s3EndpointRegion.isEmpty) {
throw new CelebornException("S3 storage is enabled but s3EndpointRegion is not set")
}
Expand All @@ -66,7 +66,7 @@ object CelebornHadoopUtils extends Logging {
"org.apache.hadoop.fs.s3a.auth.IAMInstanceCredentialsProvider")

hadoopConf.set("fs.s3a.endpoint.region", conf.s3EndpointRegion)
} else if (conf.ossDir.nonEmpty) {
} else if (conf.ossDirList.nonEmpty) {
if (conf.ossAccessKey.isEmpty || conf.ossSecretKey.isEmpty || conf.ossEndpoint.isEmpty) {
throw new CelebornException(
"OSS storage is enabled but ossAccessKey, ossSecretKey, or ossEndpoint is not set")
Expand All @@ -90,16 +90,40 @@ object CelebornHadoopUtils extends Logging {
}
}

def getHadoopFS(conf: CelebornConf): java.util.Map[StorageInfo.Type, FileSystem] = {
def getHadoopConf(conf: CelebornConf): Configuration = {
newConfiguration(conf)
}

def getHadoopFs(path: String, hadoopConf: Configuration): FileSystem = {
new Path(path).getFileSystem(hadoopConf)
}

def getHadoopFs(path: Path, hadoopConf: Configuration): FileSystem = {
path.getFileSystem(hadoopConf)
}

def getHadoopFS(conf: CelebornConf)
: java.util.Map[StorageInfo.Type, util.List[Tuple2[String, FileSystem]]] = {
val hadoopConf = newConfiguration(conf)
initKerberos(conf, hadoopConf)
val hadoopFs = new java.util.HashMap[StorageInfo.Type, FileSystem]()
val hadoopFs = new java.util.HashMap[StorageInfo.Type, util.List[Tuple2[String, FileSystem]]]()

conf.remoteStorageDirs.foreach(dirs =>
dirs.foreach {
case (storageType, dir) => {
val path = new Path(dir)
hadoopFs.put(storageType, path.getFileSystem(hadoopConf))
case (storageType, dirs) => {
dirs.foreach { dir =>
val list = hadoopFs.get(storageType)
if (list != null && !list.stream().anyMatch((tuple: Tuple2[String, FileSystem]) =>
tuple._1.equals(dir))) {
val path = new Path(dir)
list.add((dir, path.getFileSystem(hadoopConf)))
} else {
val path = new Path(dir)
val list = new util.ArrayList[(String, FileSystem)]
list.add((dir, path.getFileSystem(hadoopConf)))
hadoopFs.put(storageType, list)
}
}
}
})

Expand Down
57 changes: 33 additions & 24 deletions master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -237,7 +237,7 @@ private[celeborn] class Master(
private val slotsAssignInterruptionAwareThreshold =
conf.masterSlotsAssignInterruptionAwareThreshold

private var hadoopFs: util.Map[StorageInfo.Type, FileSystem] = _
private var hadoopFs: util.Map[StorageInfo.Type, util.List[Tuple2[String, FileSystem]]] = _
masterSource.addGauge(MasterSource.REGISTERED_SHUFFLE_COUNT) { () =>
statusSystem.registeredShuffleCount
}
Expand Down Expand Up @@ -391,13 +391,18 @@ private[celeborn] class Master(
metricsSystem.stop()
if (hadoopFs != null) {
hadoopFs.asScala.foreach {
case (storageType, fs) =>
if (fs != null) {
try {
fs.close()
} catch {
case t: Throwable =>
logError(s"Close $storageType FileSystem ${fs.getUri} failed.", t)
case (storageType, list) =>
if (list != null) {
list.asScala.foreach {
case (dir, fs) =>
try {
fs.close()
} catch {
case t: Throwable =>
logError(
s"Close $storageType DfsMountPoint $dir FileSystem ${fs.getUri} failed.",
t)
}
}
}
}
Expand Down Expand Up @@ -1164,28 +1169,32 @@ private[celeborn] class Master(
}
}
remoteStorageDirs.foreach(_.foreach {
case (_, dir) => processDir(dir, expiredDir)
case (_, dirs) => processDir(dirs, expiredDir)
})
}

private def processDir(dfsDir: String, expiredDir: String): Unit = {
val dfsWorkPath = new Path(dfsDir, conf.workerWorkingDir)
hadoopFs.asScala.map(_._2).filter(_.exists(dfsWorkPath)).foreach { fs =>
if (expiredDir.nonEmpty) {
val dirToDelete = new Path(dfsWorkPath, expiredDir)
// delete specific app dir on application lost
CelebornHadoopUtils.deleteDFSPathOrLogError(fs, dirToDelete, true)
} else {
val iter = fs.listStatusIterator(dfsWorkPath)
while (iter.hasNext && isMasterActive == 1) {
val fileStatus = iter.next()
if (!statusSystem.appHeartbeatTime.containsKey(fileStatus.getPath.getName)) {
CelebornHadoopUtils.deleteDFSPathOrLogError(fs, fileStatus.getPath, true)
private def processDir(dfsDirs: Set[String], expiredDir: String): Unit = {
dfsDirs.foreach(dfsDir => {
val dfsWorkPath = new Path(dfsDir, conf.workerWorkingDir)
hadoopFs.values().asScala.flatMap(_.asScala).foreach { tuple =>
val fs = tuple._2
if (expiredDir.nonEmpty) {
val dirToDelete = new Path(dfsWorkPath, expiredDir)
// delete specific app dir on application lost
CelebornHadoopUtils.deleteDFSPathOrLogError(fs, dirToDelete, true)
} else {
val iter = fs.listStatusIterator(dfsWorkPath)
while (iter.hasNext && isMasterActive == 1) {
val fileStatus = iter.next()
if (!statusSystem.appHeartbeatTime.containsKey(fileStatus.getPath.getName)) {
CelebornHadoopUtils.deleteDFSPathOrLogError(fs, fileStatus.getPath, true)
}
}
}

}
}
}
})

}

private def gaugeShuffleFallbackCounts(): Unit = {
Expand Down
Loading
Loading