You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2023/03/18 03:47:51 UTC
[incubator-celeborn] branch main updated: [CELEBORN-442]Support hdfs compatible file system (#1360)
This is an automated email from the ASF dual-hosted git repository.
zhouky pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 0b78c6d32 [CELEBORN-442]Support hdfs compatible file system (#1360)
0b78c6d32 is described below
commit 0b78c6d3258e860b46b670753e0ed7b243066c68
Author: 乐活优格 <tc...@gmail.com>
AuthorDate: Sat Mar 18 11:47:46 2023 +0800
[CELEBORN-442]Support hdfs compatible file system (#1360)
---
.../main/scala/org/apache/celeborn/common/util/Utils.scala | 3 ++-
.../service/deploy/worker/storage/StorageManager.scala | 12 +++++++-----
2 files changed, 9 insertions(+), 6 deletions(-)
diff --git a/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala b/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
index e4abc6bd8..bb028d063 100644
--- a/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/util/Utils.scala
@@ -957,9 +957,10 @@ object Utils extends Logging {
val SORTED_SUFFIX = ".sorted"
val INDEX_SUFFIX = ".index"
val SUFFIX_HDFS_WRITE_SUCCESS = ".success"
+ val COMPATIBLE_HDFS_REGEX = "^[a-zA-z0-9]+://((\\w)+/?)+$"
def isHdfsPath(path: String): Boolean = {
- path.startsWith("hdfs://")
+ path.matches(COMPATIBLE_HDFS_REGEX)
}
def getSortedFilePath(path: String): String = {
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
index 386a8ac01..e1ac71922 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
@@ -130,13 +130,15 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
val hdfsWriters = new ConcurrentHashMap[String, FileWriter]()
val (hdfsFlusher, _totalHdfsFlusherThread) =
if (!hdfsDir.isEmpty) {
+ val path = new Path(hdfsDir)
+ val scheme = path.toUri.getScheme
+ val disableCacheName = String.format("fs.%s.impl.disable.cache", scheme);
val hdfsConfiguration = new Configuration
- hdfsConfiguration.set("fs.defaultFS", hdfsDir)
hdfsConfiguration.set("dfs.replication", "2")
- hdfsConfiguration.set("fs.hdfs.impl.disable.cache", "false")
- logInfo("Celeborn will ignore cluster settings" +
- " about fs.hdfs.impl.disable.cache and set it to false")
- StorageManager.hadoopFs = FileSystem.get(hdfsConfiguration)
+ hdfsConfiguration.set(disableCacheName, "false")
+ logInfo("Celeborn will ignore cluster settings " +
+ disableCacheName + " and set it to false")
+ StorageManager.hadoopFs = path.getFileSystem(hdfsConfiguration)
(
Some(new HdfsFlusher(
workerSource,