diff --git a/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java b/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java index d37ec644231..789660d62f6 100644 --- a/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java +++ b/client/src/main/java/org/apache/celeborn/client/ShuffleClient.java @@ -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; @@ -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 hadoopFs; + private static volatile Map>> hadoopFs; private static LongAdder totalReadCounter = new LongAdder(); private static LongAdder localShuffleReadCounter = new LongAdder(); @@ -114,7 +115,8 @@ public static ShuffleClient get( return _instance; } - public static Map getHadoopFs(CelebornConf conf) { + public static Map>> getHadoopFs( + CelebornConf conf) { if (null == hadoopFs) { synchronized (ShuffleClient.class) { if (null == hadoopFs) { diff --git a/client/src/main/java/org/apache/celeborn/client/read/DfsPartitionReader.java b/client/src/main/java/org/apache/celeborn/client/read/DfsPartitionReader.java index 5493e7f2904..2fbeae31f11 100644 --- a/client/src/main/java/org/apache/celeborn/client/read/DfsPartitionReader.java +++ b/client/src/main/java/org/apache/celeborn/client/read/DfsPartitionReader.java @@ -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; @@ -101,15 +103,17 @@ public DfsPartitionReader( this.metricsCallback = metricsCallback; this.location = location; + List> 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 { 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 010cd57a4e0..e177dce6688 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -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} @@ -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 => @@ -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 @@ -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 } @@ -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") @@ -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") @@ -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") diff --git a/common/src/main/scala/org/apache/celeborn/common/util/CelebornHadoopUtils.scala b/common/src/main/scala/org/apache/celeborn/common/util/CelebornHadoopUtils.scala index 2940afcd2f1..8c9b0c4ba76 100644 --- a/common/src/main/scala/org/apache/celeborn/common/util/CelebornHadoopUtils.scala +++ b/common/src/main/scala/org/apache/celeborn/common/util/CelebornHadoopUtils.scala @@ -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") @@ -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") } @@ -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") @@ -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) + } + } } }) diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala index 0799af45cf4..821e6a5013e 100644 --- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala +++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala @@ -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 } @@ -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) + } } } } @@ -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 = { diff --git a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/MemoryManager.java b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/MemoryManager.java index de17c45494a..f0b787ad28e 100644 --- a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/MemoryManager.java +++ b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/memory/MemoryManager.java @@ -100,6 +100,7 @@ public class MemoryManager { private long pinnedMemoryLastCheckTime = 0; private boolean resumingByPinnedMemory = false; private long workerPinnedMemoryResumeKeepTime; + private CelebornConf conf; @VisibleForTesting public static MemoryManager initialize(CelebornConf conf) { @@ -142,6 +143,7 @@ private MemoryManager(CelebornConf conf, StorageManager storageManager, Abstract long readBufferTargetNotifyThreshold = conf.readBufferTargetNotifyThreshold(); boolean aggressiveEvictModeEnabled = conf.workerMemoryFileStorageEictAggressiveModeEnabled(); double evictRatio = conf.workerMemoryFileStorageEvictRatio(); + this.conf = conf; forceAppendPauseSpentTimeThreshold = conf.metricsWorkerForceAppendPauseSpentTimeThreshold(); maxDirectMemory = DynMethods.builder("maxDirectMemory") @@ -596,6 +598,10 @@ public ByteBufAllocator getStorageByteBufAllocator() { return storageManager.storageBufferAllocator(); } + public CelebornConf getConf() { + return conf; + } + @VisibleForTesting public static void reset() { _INSTANCE = null; diff --git a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/DfsPartitionDataReader.java b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/DfsPartitionDataReader.java index 0e426270d98..75c6085356b 100644 --- a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/DfsPartitionDataReader.java +++ b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/DfsPartitionDataReader.java @@ -23,10 +23,8 @@ import io.netty.buffer.ByteBuf; import org.apache.commons.io.IOUtils; import org.apache.hadoop.fs.FSDataInputStream; -import org.apache.hadoop.fs.FileSystem; import org.apache.celeborn.common.meta.DiskFileInfo; -import org.apache.celeborn.common.protocol.StorageInfo; import org.apache.celeborn.common.util.Utils; public class DfsPartitionDataReader extends PartitionDataReader { @@ -39,19 +37,14 @@ public DfsPartitionDataReader( FSDataInputStream dataInputStream, FSDataInputStream indexInputStream, ByteBuffer headerBuffer, - ByteBuffer indexBuffer) - throws IOException { + ByteBuffer indexBuffer, + long dataFileSize, + long indexFileSize) { super(fileInfo, headerBuffer, indexBuffer); - FileSystem fileSystem = - StorageManager.hadoopFs() - .get( - fileInfo.isHdfs() - ? StorageInfo.Type.HDFS - : fileInfo.isS3() ? StorageInfo.Type.S3 : StorageInfo.Type.OSS); this.dataInputStream = dataInputStream; this.indexInputStream = indexInputStream; - this.dataFileSize = fileSystem.getFileStatus(fileInfo.getDfsPath()).getLen(); - this.indexFileSize = fileSystem.getFileStatus(fileInfo.getDfsIndexPath()).getLen(); + this.dataFileSize = dataFileSize; + this.indexFileSize = indexFileSize; } @Override diff --git a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionData.java b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionData.java index fc0522fe599..30335672a64 100644 --- a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionData.java +++ b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionData.java @@ -28,13 +28,17 @@ import io.netty.buffer.ByteBuf; import io.netty.channel.Channel; +import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.celeborn.common.CelebornConf; import org.apache.celeborn.common.meta.DiskFileInfo; import org.apache.celeborn.common.meta.MapFileMeta; +import org.apache.celeborn.common.util.CelebornHadoopUtils; import org.apache.celeborn.common.util.FileChannelUtils; import org.apache.celeborn.common.util.JavaUtils; import org.apache.celeborn.common.util.ThreadUtils; @@ -54,6 +58,8 @@ public class MapPartitionData implements MemoryManager.ReadBufferTargetChangeLis private FileChannel indexChannel; private FSDataInputStream hdfsDataInputStream; private FSDataInputStream hdfsIndexInputStream; + protected long dataFileSize; + protected long indexFileSize; private volatile boolean isReleased = false; private final BufferQueue bufferQueue = new BufferQueue(); private AtomicBoolean bufferQueueInitialized = new AtomicBoolean(false); @@ -63,6 +69,7 @@ public class MapPartitionData implements MemoryManager.ReadBufferTargetChangeLis private int maxReadBuffers; private int minBuffersToTriggerRead; private AtomicBoolean hasReadingTask = new AtomicBoolean(false); + private static volatile Configuration hadoopConf; public MapPartitionData( int minReadBuffers, @@ -99,15 +106,13 @@ public MapPartitionData( false)); if (diskFileInfo.isDFS()) { - this.hdfsDataInputStream = - StorageManager.hadoopFs() - .get(diskFileInfo.getStorageType()) - .open(new Path(diskFileInfo.getFilePath())); - this.hdfsIndexInputStream = - StorageManager.hadoopFs() - .get(diskFileInfo.getStorageType()) - .open(new Path(diskFileInfo.getIndexPath())); - + FileSystem fileSystem = + CelebornHadoopUtils.getHadoopFs( + diskFileInfo.getDfsPath(), getHadoopConf(memoryManager.getConf())); + this.hdfsDataInputStream = fileSystem.open(diskFileInfo.getDfsPath()); + this.hdfsIndexInputStream = fileSystem.open(new Path(diskFileInfo.getIndexPath())); + this.dataFileSize = fileSystem.getFileStatus(diskFileInfo.getDfsPath()).getLen(); + this.indexFileSize = fileSystem.getFileStatus(diskFileInfo.getDfsIndexPath()).getLen(); } else { this.dataFileChanel = FileChannelUtils.openReadableFileChannel(diskFileInfo.getFilePath()); this.indexChannel = FileChannelUtils.openReadableFileChannel(diskFileInfo.getIndexPath()); @@ -187,7 +192,13 @@ public void recycle(ByteBuf buffer) { } protected void openReader(MapPartitionDataReader reader) throws IOException { - reader.open(dataFileChanel, indexChannel, hdfsDataInputStream, hdfsIndexInputStream); + reader.open( + dataFileChanel, + indexChannel, + hdfsDataInputStream, + hdfsIndexInputStream, + dataFileSize, + indexFileSize); } public synchronized void readBuffers() { @@ -289,4 +300,15 @@ public void onChange(long newMemoryTarget) { updateBuffersTarget((int) Math.ceil(newMemoryTarget * 1.0 / mapFileMeta.getBufferSize())); bufferQueue.trim(); } + + public static Configuration getHadoopConf(CelebornConf conf) { + if (null == hadoopConf) { + synchronized (MapPartitionData.class) { + if (null == hadoopConf) { + hadoopConf = CelebornHadoopUtils.getHadoopConf(conf); + } + } + } + return hadoopConf; + } } diff --git a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionDataReader.java b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionDataReader.java index 0f387402755..648533e9a55 100644 --- a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionDataReader.java +++ b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/MapPartitionDataReader.java @@ -132,13 +132,21 @@ public void open( FileChannel dataFileChannel, FileChannel indexFileChannel, FSDataInputStream dataInputStream, - FSDataInputStream indexInputStream) + FSDataInputStream indexInputStream, + long dataFileSize, + long indexFileSize) throws IOException { if (!isOpen) { this.partitionDataReader = fileInfo.isDFS() ? new DfsPartitionDataReader( - fileInfo, dataInputStream, indexInputStream, headerBuffer, indexBuffer) + fileInfo, + dataInputStream, + indexInputStream, + headerBuffer, + indexBuffer, + dataFileSize, + indexFileSize) : new LocalPartitionDataReader( fileInfo, dataFileChannel, indexFileChannel, headerBuffer, indexBuffer); // index is (offset,length) diff --git a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java index cdfb95c1e59..64976f8a585 100644 --- a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java +++ b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java @@ -43,6 +43,7 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.CompositeByteBuf; import org.apache.commons.io.IOUtils; +import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; @@ -99,6 +100,7 @@ public class PartitionFilesSorter extends ShuffleRecoverHelper { private final ExecutorService fileSorterExecutors; private final ExecutorService fileSorterSchedulerThread; + private final Configuration hadoopConf; public PartitionFilesSorter( MemoryManager memoryManager, CelebornConf conf, AbstractSource source) { @@ -134,6 +136,8 @@ public PartitionFilesSorter( this.sortedFilesDb = null; } + hadoopConf = CelebornHadoopUtils.getHadoopConf(conf); + fileSorterExecutors = ThreadUtils.newDaemonCachedThreadPool( "worker-file-sorter-executor", conf.workerPartitionSorterThreads(), 120); @@ -490,7 +494,10 @@ public Set getSortedShuffleFiles(String shuffleKey) { } protected void writeIndex( - Map> indexMap, String indexFilePath, boolean isDfs) + Map> indexMap, + String indexFilePath, + boolean isDfs, + FileSystem hadoopFs) throws IOException { FSDataOutputStream dfsIndexOutput = null; FileChannel indexFileChannel = null; @@ -501,7 +508,6 @@ protected void writeIndex( } else if (Utils.isOssPath(indexFilePath)) { storageType = StorageInfo.Type.OSS; } - FileSystem hadoopFs = StorageManager.hadoopFs().get(storageType); // If the index file exists, it will be overwritten. // So there is no need to check its existence. dfsIndexOutput = hadoopFs.create(new Path(indexFilePath)); @@ -620,13 +626,8 @@ public DiskFileInfo resolve( int indexSize; try { if (isDfs) { - StorageInfo.Type storageType = StorageInfo.Type.HDFS; - if (Utils.isS3Path(indexFilePath)) { - storageType = StorageInfo.Type.S3; - } else if (Utils.isOssPath(indexFilePath)) { - storageType = StorageInfo.Type.OSS; - } - FileSystem hadoopFs = StorageManager.hadoopFs().get(storageType); + FileSystem hadoopFs = + CelebornHadoopUtils.getHadoopFs(indexFilePath, hadoopConf); dfsIndexStream = hadoopFs.open(new Path(indexFilePath)); indexSize = (int) hadoopFs.getFileStatus(new Path(indexFilePath)).getLen(); } else { @@ -710,13 +711,7 @@ class FileSorter { indexFile.delete(); } } else { - StorageInfo.Type storageType = StorageInfo.Type.HDFS; - if (Utils.isS3Path(indexFilePath)) { - storageType = StorageInfo.Type.S3; - } else if (Utils.isOssPath(indexFilePath)) { - storageType = StorageInfo.Type.OSS; - } - this.hadoopFs = StorageManager.hadoopFs().get(storageType); + this.hadoopFs = CelebornHadoopUtils.getHadoopFs(fileInfo.getDfsSortedPath(), hadoopConf); if (hadoopFs.exists(fileInfo.getDfsSortedPath())) { hadoopFs.delete(fileInfo.getDfsSortedPath(), false); } @@ -796,7 +791,7 @@ public void sort() { sortedBlockInfoMap.put(mapId, sortedShuffleBlocks); } - writeIndex(sortedBlockInfoMap, indexFilePath, isDfs); + writeIndex(sortedBlockInfoMap, indexFilePath, isDfs, hadoopFs); updateSortedShuffleFiles(shuffleKey, fileId, originFileLen); originFileInfo.getReduceFileMeta().setSorted(); cleaner.add(this); diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTask.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTask.scala index 5053456d45b..e627459b5a8 100644 --- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTask.scala +++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTask.scala @@ -21,7 +21,7 @@ import java.io.{ByteArrayInputStream, Closeable, IOException} import java.nio.channels.FileChannel import io.netty.buffer.{ByteBufUtil, CompositeByteBuf} -import org.apache.hadoop.fs.Path +import org.apache.hadoop.fs.{FileSystem, Path} import org.apache.celeborn.common.internal.Logging import org.apache.celeborn.common.metrics.source.AbstractSource @@ -101,10 +101,10 @@ private[worker] class HdfsFlushTask( val path: Path, notifier: FlushNotifier, keepBuffer: Boolean, - source: AbstractSource) extends DfsFlushTask(buffer, notifier, keepBuffer, source) { + source: AbstractSource, + hadoopFs: FileSystem) extends DfsFlushTask(buffer, notifier, keepBuffer, source) { override def flush(copyBytes: Array[Byte]): Unit = { val readableBytes = buffer.readableBytes() - val hadoopFs = StorageManager.hadoopFs.get(Type.HDFS) val hdfsStream = hadoopFs.append(path, 256 * 1024) flush(hdfsStream) { hdfsStream.write(convertBufferToBytes(buffer, copyBytes, readableBytes)) diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/PartitionMetaHandler.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/PartitionMetaHandler.scala index 8efa981b8df..6dc1af5ad01 100644 --- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/PartitionMetaHandler.scala +++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/PartitionMetaHandler.scala @@ -93,8 +93,10 @@ trait PartitionMetaHandler { class MapPartitionMetaHandler( diskFileInfo: DiskFileInfo, - notifier: FlushNotifier) extends PartitionMetaHandler { - lazy val hadoopFs: FileSystem = StorageManager.hadoopFs.get(diskFileInfo.getStorageType) + notifier: FlushNotifier, + partitionUniqueId: Option[String] = None) extends PartitionMetaHandler { + lazy val hadoopFs: FileSystem = + StorageManager.dfsTupleForPartition(diskFileInfo.getStorageType, partitionUniqueId.get)._2 val logger: Logger = LoggerFactory.getLogger(classOf[MapPartitionMetaHandler]) val fileMeta: MapFileMeta = diskFileInfo.getFileMeta.asInstanceOf[MapFileMeta] var numSubpartitions = 0 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 a0b46dd6dc3..6b786a807ee 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 @@ -159,13 +159,13 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs (flushers, totalThread) } - val hdfsDir = conf.hdfsDir - val s3Dir = conf.s3Dir - val ossDir = conf.ossDir + val hdfsDirList = conf.hdfsDirList + val s3DirList = conf.s3DirList + val ossDirList = conf.ossDirList val hdfsPermission = new FsPermission("755") val (hdfsFlusher, _totalHdfsFlusherThread) = if (hasHDFSStorage) { - logInfo(s"Initialize HDFS support with path $hdfsDir") + logInfo(s"Initialize HDFS support with path ${hdfsDirList}") try { StorageManager.hadoopFs = CelebornHadoopUtils.getHadoopFS(conf) } catch { @@ -188,7 +188,7 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs val (s3Flusher, _totalS3FlusherThread) = if (hasS3Storage) { - logInfo(s"Initialize S3 support with path $s3Dir") + logInfo(s"Initialize S3 support with path $s3DirList") try { StorageManager.hadoopFs = CelebornHadoopUtils.getHadoopFS(conf) } catch { @@ -211,7 +211,7 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs val (ossFlusher, _totalOssFlusherThread) = if (hasOssStorage) { - logInfo(s"Initialize OSS support with path $ossDir") + logInfo(s"Initialize OSS support with path $ossDirList") try { StorageManager.hadoopFs = CelebornHadoopUtils.getHadoopFS(conf) } catch { @@ -240,7 +240,7 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs lazy val localOrDfsStorageAvailable: Boolean = { StorageInfo.OSSAvailable(activeTypes) || StorageInfo.S3Available(activeTypes) || StorageInfo.HDFSAvailable(activeTypes) || StorageInfo.localDiskAvailable(activeTypes) || - hdfsDir.nonEmpty || !diskInfos.isEmpty || s3Dir.nonEmpty || ossDir.nonEmpty + hdfsDirList.nonEmpty || !diskInfos.isEmpty || s3DirList.nonEmpty || ossDirList.nonEmpty } override def notifyError(mountPoint: String, diskStatus: DiskStatus): Unit = this.synchronized { @@ -655,13 +655,11 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs } expireStorageTypes.foreach(storageType => { try { - val dir = - if (storageType == StorageInfo.Type.HDFS) hdfsDir - else if (storageType == StorageInfo.Type.OSS) ossDir - else s3Dir - StorageManager.hadoopFs.get(storageType).delete( - new Path(new Path(dir, conf.workerWorkingDir), s"$appId/$shuffleId"), - true) + StorageManager.hadoopFs.get(storageType).forEach(tuple => { + val dir = tuple._1 + val fs = tuple._2 + fs.delete(new Path(new Path(dir, conf.workerWorkingDir), s"$appId/$shuffleId"), true) + }) } catch { case e: Exception => logWarning(s"Clean expired $storageType shuffle failed.", e) } @@ -769,21 +767,16 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs } } val dfsCleaned = hadoopFs == null || hadoopFs.asScala.forall { - case (storageType, fs) => - val dfsDir = - if (storageType == StorageInfo.Type.HDFS) - hdfsDir - else if (storageType == StorageInfo.Type.OSS) ossDir - else s3Dir - try { - val dfsWorkPath = new Path(dfsDir, conf.workerWorkingDir) - // DFS path not exist when first time initialize - !fs.exists(dfsWorkPath) || !fs.listFiles(dfsWorkPath, false).hasNext - } catch { - case t: Throwable => - // When DFS is not accessible, assume it's cleaned - logError("checkIfWorkingDirCleaned failed.", t) - true + case (_, list) => + list.asScala.forall { + case (dir, fs) => + val dfsWorkPath = new Path(dir, conf.workerWorkingDir) + // DFS path not exist when first time initialize + if (fs.exists(dfsWorkPath)) { + !fs.listFiles(dfsWorkPath, false).hasNext + } else { + true + } } case _ => true @@ -804,13 +797,15 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs def close(exitKind: Int): Unit = { 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.isEmpty) { + list.asScala.foreach { tuple => + try { + tuple._2.close() + } catch { + case t: Throwable => + logError(s"Close $storageType FileSystem ${tuple._2.getUri} failed.", t) + } } } } @@ -1103,10 +1098,12 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs } if (dirs.isEmpty && location.getStorageInfo.HDFSAvailable()) { + val (hdfsDir, fs) = + StorageManager.dfsTupleForPartition(StorageInfo.Type.HDFS, location.getUniqueId) val shuffleDir = new Path(new Path(hdfsDir, conf.workerWorkingDir), s"$appId/$shuffleId") FileSystem.mkdirs( - StorageManager.hadoopFs.get(StorageInfo.Type.HDFS), + fs, shuffleDir, hdfsPermission) val hdfsFilePath = new Path(shuffleDir, fileName).toString @@ -1121,10 +1118,12 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs hdfsFileInfo) return (hdfsFlusher.get, hdfsFileInfo, null) } else if (dirs.isEmpty && location.getStorageInfo.S3Available()) { + val (s3Dir, fs) = + StorageManager.dfsTupleForPartition(StorageInfo.Type.S3, location.getUniqueId) val shuffleDir = new Path(new Path(s3Dir, conf.workerWorkingDir), s"$appId/$shuffleId") FileSystem.mkdirs( - StorageManager.hadoopFs.get(StorageInfo.Type.S3), + fs, shuffleDir, hdfsPermission) val s3FilePath = new Path(shuffleDir, fileName).toString @@ -1139,10 +1138,12 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs s3FileInfo) return (s3Flusher.get, s3FileInfo, null) } else if (dirs.isEmpty && location.getStorageInfo.OSSAvailable()) { + val (ossDir, fs) = + StorageManager.dfsTupleForPartition(StorageInfo.Type.OSS, location.getUniqueId) val shuffleDir = new Path(new Path(ossDir, conf.workerWorkingDir), s"$appId/$shuffleId") FileSystem.mkdirs( - StorageManager.hadoopFs.get(StorageInfo.Type.OSS), + fs, shuffleDir, hdfsPermission) val ossFilePath = new Path(shuffleDir, fileName).toString @@ -1246,5 +1247,13 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs } object StorageManager { - var hadoopFs: util.Map[StorageInfo.Type, FileSystem] = _ + // StorageInfo -> (mountPoint -> FileSystem) + var hadoopFs: util.Map[StorageInfo.Type, util.List[Tuple2[String, FileSystem]]] = _ + + def dfsTupleForPartition( + storageType: StorageInfo.Type, + partitionUniqueId: String): (String, FileSystem) = { + val hadoopFsSeq = hadoopFs.get(storageType) + hadoopFsSeq.get(Math.floorMod(partitionUniqueId.hashCode, hadoopFsSeq.size())) + } } 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 352abe1e680..0b9d05a2dfe 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 @@ -520,7 +520,9 @@ class DfsTierWriter( storageType == StorageInfo.Type.HDFS || storageType == StorageInfo.Type.OSS || storageType == StorageInfo.Type.S3) flusherBufferSize = conf.workerHdfsFlusherBufferSize private val flushWorkerIndex: Int = flusher.getWorkerIndex - val hadoopFs: FileSystem = StorageManager.hadoopFs.get(storageType) + val hadoopFs: FileSystem = StorageManager.dfsTupleForPartition( + storageType, + partitionDataWriterContext.getPartitionLocation.getUniqueId)._2 var deleted = false private var s3MultipartUploadHandler: MultipartUploadHandler = _ private var ossMultipartUploadHandler: MultipartUploadHandler = _ @@ -595,7 +597,7 @@ class DfsTierWriter( override def genFlushTask(finalFlush: Boolean, keepBuffer: Boolean): FlushTask = { notifier.numPendingFlushes.incrementAndGet() if (dfsFileInfo.isHdfs) { - new HdfsFlushTask(flushBuffer, dfsFileInfo.getDfsPath(), notifier, true, source) + new HdfsFlushTask(flushBuffer, dfsFileInfo.getDfsPath(), notifier, true, source, hadoopFs) } else if (dfsFileInfo.isOSS) { val flushTask = new OssFlushTask( flushBuffer, diff --git a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriterSuiteUtils.java b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriterSuiteUtils.java index 471a4486583..f7a971e3a21 100644 --- a/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriterSuiteUtils.java +++ b/worker/src/test/java/org/apache/celeborn/service/deploy/worker/storage/PartitionDataWriterSuiteUtils.java @@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicInteger; import scala.Function0; +import scala.Option; import scala.Tuple4; import org.mockito.Mockito; @@ -65,7 +66,9 @@ public static StorageManager prepareDiskFileTestEnvironment( PartitionMetaHandler metaHandler = null; if (!reduceMeta) { fileInfo.replaceFileMeta(new MapFileMeta(32 * 1024, 10)); - metaHandler = new MapPartitionMetaHandler(fileInfo, flushNotifier); + String partitionUniqueId = context.getPartitionLocation().getUniqueId(); + metaHandler = + new MapPartitionMetaHandler(fileInfo, flushNotifier, Option.apply(partitionUniqueId)); } else { metaHandler = new ReducePartitionMetaHandler(conf.shuffleRangeReadFilterEnabled(), fileInfo); }