[CELEBORN-1530] support MPU for S3

### What changes were proposed in this pull request? as title ### Why are the changes needed? AWS S3 doesn't support append, so Celeborn had to copy the historical data from s3 to worker and write to s3 again, which heavily scales out the write. This PR implements a better solution via MPU to avoid copy-and-write. ### Does this PR introduce _any_ user-facing change? ### How was this patch tested? ![WechatIMG257](https://github.com/user-attachments/assets/968d9162-e690-4767-8bed-e490e3055753) I conducted an experiment with a 1GB input dataset to compare the performance of Celeborn using only S3 storage versus using SSD storage. The results showed that Celeborn with SSD storage was approximately three times faster than with only S3 storage. <img width="1728" alt="Screenshot 2024-11-16 at 13 02 10" src="https://github.com/user-attachments/assets/8f879c47-c01a-4004-9eae-1c266c1f3ef2"> The above screenshot is the second test with 5000 mapper and reducer that I did. Closes #2830 from zhaohehuhu/dev-1021. Lead-authored-by: zhaohehuhu <[email protected]> Co-authored-by: He Zhao <[email protected]> Signed-off-by: mingji <[email protected]>
apache · Nov 22, 2024 · a2d3972 · a2d3972
1 parent 317cb97
commit a2d3972
Show file tree

Hide file tree

Showing 17 changed files with 417 additions and 99 deletions.
diff --git a/build/make-distribution.sh b/build/make-distribution.sh
@@ -264,8 +264,8 @@ function sbt_build_service {
   echo "Celeborn $VERSION$GITREVSTRING" > "$DIST_DIR/RELEASE"
   echo "Build flags: $@" >> "$DIST_DIR/RELEASE"
 
-  if [[ $@ == *"hadoop-aws"* ]]; then
-     SBT_MAVEN_PROFILES="hadoop-aws"
+  if [[ $@ == *"aws"* ]]; then
+     export SBT_MAVEN_PROFILES="aws"
   fi
   BUILD_COMMAND=("$SBT" clean package)
 

diff --git a/common/pom.xml b/common/pom.xml
@@ -209,25 +209,5 @@
         </dependency>
       </dependencies>
     </profile>
-    <profile>
-      <id>hadoop-aws</id>
-      <activation>
-        <property>
-          <name>hadoop-aws-deps</name>
-        </property>
-      </activation>
-      <dependencies>
-        <dependency>
-          <groupId>org.apache.hadoop</groupId>
-          <artifactId>hadoop-aws</artifactId>
-          <version>${hadoop.version}</version>
-        </dependency>
-        <dependency>
-          <groupId>com.amazonaws</groupId>
-          <artifactId>aws-java-sdk-bundle</artifactId>
-          <version>${aws.version}</version>
-        </dependency>
-      </dependencies>
-    </profile>
   </profiles>
 </project>
diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -1147,7 +1147,9 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
 
   def s3SecretKey: String = get(S3_SECRET_KEY).getOrElse("")
 
-  def s3Endpoint: String = get(S3_ENDPOINT).getOrElse("")
+  def s3EndpointRegion: String = get(S3_ENDPOINT_REGION).getOrElse("")
+
+  def s3MultiplePartUploadMaxRetries: Int = get(S3_MPU_MAX_RETRIES)
 
   def s3Dir: String = {
     get(S3_DIR).map {
@@ -3062,14 +3064,22 @@ object CelebornConf extends Logging {
       .stringConf
       .createOptional
 
-  val S3_ENDPOINT: OptionalConfigEntry[String] =
-    buildConf("celeborn.storage.s3.endpoint")
+  val S3_ENDPOINT_REGION: OptionalConfigEntry[String] =
+    buildConf("celeborn.storage.s3.endpoint.region")
       .categories("worker", "master", "client")
       .version("0.6.0")
       .doc("S3 endpoint for Celeborn to store shuffle data.")
       .stringConf
       .createOptional
 
+  val S3_MPU_MAX_RETRIES: ConfigEntry[Int] =
+    buildConf("celeborn.storage.s3.mpu.maxRetries")
+      .categories("worker")
+      .version("0.6.0")
+      .doc("S3 MPU upload max retries.")
+      .intConf
+      .createWithDefault(5)
+
   val WORKER_DISK_RESERVE_SIZE: ConfigEntry[Long] =
     buildConf("celeborn.worker.storage.disk.reserve.size")
       .withAlternative("celeborn.worker.disk.reserve.size")
@@ -3552,7 +3562,7 @@ object CelebornConf extends Logging {
       .version("0.6.0")
       .doc("Size of buffer used by a S3 flusher.")
       .bytesConf(ByteUnit.BYTE)
-      .createWithDefaultString("4m")
+      .createWithDefaultString("6m")
 
   val WORKER_WRITER_CLOSE_TIMEOUT: ConfigEntry[Long] =
     buildConf("celeborn.worker.writer.close.timeout")

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
@@ -49,17 +49,17 @@ object CelebornHadoopUtils extends Logging {
     }
 
     if (conf.s3Dir.nonEmpty) {
-      if (conf.s3AccessKey.isEmpty || conf.s3SecretKey.isEmpty || conf.s3Endpoint.isEmpty) {
+      if (conf.s3AccessKey.isEmpty || conf.s3SecretKey.isEmpty || conf.s3EndpointRegion.isEmpty) {
         throw new CelebornException(
-          "S3 storage is enabled but s3AccessKey, s3SecretKey, or s3Endpoint is not set")
+          "S3 storage is enabled but s3AccessKey, s3SecretKey, or s3EndpointRegion is not set")
       }
       hadoopConf.set("fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
       hadoopConf.set(
         "fs.s3a.aws.credentials.provider",
         "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider")
       hadoopConf.set("fs.s3a.access.key", conf.s3AccessKey)
       hadoopConf.set("fs.s3a.secret.key", conf.s3SecretKey)
-      hadoopConf.set("fs.s3a.endpoint", conf.s3Endpoint)
+      hadoopConf.set("fs.s3a.endpoint.region", conf.s3EndpointRegion)
     }
     appendSparkHadoopConfigs(conf, hadoopConf)
     hadoopConf

diff --git a/dev/deps/dependencies-server b/dev/deps/dependencies-server
@@ -19,6 +19,7 @@ HikariCP/4.0.3//HikariCP-4.0.3.jar
 RoaringBitmap/1.0.6//RoaringBitmap-1.0.6.jar
 aopalliance-repackaged/2.6.1//aopalliance-repackaged-2.6.1.jar
 ap-loader-all/3.0-8//ap-loader-all-3.0-8.jar
+aws-java-sdk-bundle/1.12.367//aws-java-sdk-bundle-1.12.367.jar
 classgraph/4.8.138//classgraph-4.8.138.jar
 commons-cli/1.5.0//commons-cli-1.5.0.jar
 commons-crypto/1.0.0//commons-crypto-1.0.0.jar
@@ -27,6 +28,7 @@ commons-lang3/3.17.0//commons-lang3-3.17.0.jar
 commons-logging/1.1.3//commons-logging-1.1.3.jar
 failureaccess/1.0.2//failureaccess-1.0.2.jar
 guava/33.1.0-jre//guava-33.1.0-jre.jar
+hadoop-aws/3.3.6//hadoop-aws-3.3.6.jar
 hadoop-client-api/3.3.6//hadoop-client-api-3.3.6.jar
 hadoop-client-runtime/3.3.6//hadoop-client-runtime-3.3.6.jar
 hk2-api/2.6.1//hk2-api-2.6.1.jar
@@ -145,4 +147,5 @@ swagger-integration/2.2.1//swagger-integration-2.2.1.jar
 swagger-jaxrs2/2.2.1//swagger-jaxrs2-2.2.1.jar
 swagger-models/2.2.1//swagger-models-2.2.1.jar
 swagger-ui/4.9.1//swagger-ui-4.9.1.jar
+wildfly-openssl/1.1.3.Final//wildfly-openssl-1.1.3.Final.jar
 zstd-jni/1.5.2-1//zstd-jni-1.5.2-1.jar
diff --git a/docs/configuration/client.md b/docs/configuration/client.md
@@ -132,6 +132,6 @@ license: |
 | celeborn.storage.hdfs.dir | &lt;undefined&gt; | false | HDFS base directory for Celeborn to store shuffle data. | 0.2.0 |  | 
 | celeborn.storage.s3.access.key | &lt;undefined&gt; | false | S3 access key for Celeborn to store shuffle data. | 0.6.0 |  | 
 | celeborn.storage.s3.dir | &lt;undefined&gt; | false | S3 base directory for Celeborn to store shuffle data. | 0.6.0 |  | 
-| celeborn.storage.s3.endpoint | &lt;undefined&gt; | false | S3 endpoint for Celeborn to store shuffle data. | 0.6.0 |  | 
+| celeborn.storage.s3.endpoint.region | &lt;undefined&gt; | false | S3 endpoint for Celeborn to store shuffle data. | 0.6.0 |  | 
 | celeborn.storage.s3.secret.key | &lt;undefined&gt; | false | S3 secret key for Celeborn to store shuffle data. | 0.6.0 |  | 
 <!--end-include-->
diff --git a/docs/configuration/master.md b/docs/configuration/master.md
@@ -89,6 +89,6 @@ license: |
 | celeborn.storage.hdfs.kerberos.principal | &lt;undefined&gt; | false | Kerberos principal for HDFS storage connection. | 0.3.2 |  | 
 | celeborn.storage.s3.access.key | &lt;undefined&gt; | false | S3 access key for Celeborn to store shuffle data. | 0.6.0 |  | 
 | celeborn.storage.s3.dir | &lt;undefined&gt; | false | S3 base directory for Celeborn to store shuffle data. | 0.6.0 |  | 
-| celeborn.storage.s3.endpoint | &lt;undefined&gt; | false | S3 endpoint for Celeborn to store shuffle data. | 0.6.0 |  | 
+| celeborn.storage.s3.endpoint.region | &lt;undefined&gt; | false | S3 endpoint for Celeborn to store shuffle data. | 0.6.0 |  | 
 | celeborn.storage.s3.secret.key | &lt;undefined&gt; | false | S3 secret key for Celeborn to store shuffle data. | 0.6.0 |  | 
 <!--end-include-->
diff --git a/docs/configuration/worker.md b/docs/configuration/worker.md
@@ -48,7 +48,8 @@ license: |
 | celeborn.storage.hdfs.kerberos.principal | &lt;undefined&gt; | false | Kerberos principal for HDFS storage connection. | 0.3.2 |  | 
 | celeborn.storage.s3.access.key | &lt;undefined&gt; | false | S3 access key for Celeborn to store shuffle data. | 0.6.0 |  | 
 | celeborn.storage.s3.dir | &lt;undefined&gt; | false | S3 base directory for Celeborn to store shuffle data. | 0.6.0 |  | 
-| celeborn.storage.s3.endpoint | &lt;undefined&gt; | false | S3 endpoint for Celeborn to store shuffle data. | 0.6.0 |  | 
+| celeborn.storage.s3.endpoint.region | &lt;undefined&gt; | false | S3 endpoint for Celeborn to store shuffle data. | 0.6.0 |  | 
+| celeborn.storage.s3.mpu.maxRetries | 5 | false | S3 MPU upload max retries. | 0.6.0 |  | 
 | celeborn.storage.s3.secret.key | &lt;undefined&gt; | false | S3 secret key for Celeborn to store shuffle data. | 0.6.0 |  | 
 | celeborn.worker.activeConnection.max | &lt;undefined&gt; | false | If the number of active connections on a worker exceeds this configuration value, the worker will be marked as high-load in the heartbeat report, and the master will not include that node in the response of RequestSlots. | 0.3.1 |  | 
 | celeborn.worker.applicationRegistry.cache.size | 10000 | false | Cache size of the application registry on Workers. | 0.5.0 |  | 
@@ -84,7 +85,7 @@ license: |
 | celeborn.worker.flusher.hdd.threads | 1 | false | Flusher's thread count per disk used for write data to HDD disks. | 0.2.0 |  | 
 | celeborn.worker.flusher.hdfs.buffer.size | 4m | false | Size of buffer used by a HDFS flusher. | 0.3.0 |  | 
 | celeborn.worker.flusher.hdfs.threads | 8 | false | Flusher's thread count used for write data to HDFS. | 0.2.0 |  | 
-| celeborn.worker.flusher.s3.buffer.size | 4m | false | Size of buffer used by a S3 flusher. | 0.6.0 |  | 
+| celeborn.worker.flusher.s3.buffer.size | 6m | false | Size of buffer used by a S3 flusher. | 0.6.0 |  | 
 | celeborn.worker.flusher.s3.threads | 8 | false | Flusher's thread count used for write data to S3. | 0.6.0 |  | 
 | celeborn.worker.flusher.shutdownTimeout | 3s | false | Timeout for a flusher to shutdown. | 0.2.0 |  | 
 | celeborn.worker.flusher.ssd.threads | 16 | false | Flusher's thread count per disk used for write data to SSD disks. | 0.2.0 |  | 

diff --git a/master/pom.xml b/master/pom.xml
@@ -164,38 +164,33 @@
       </dependencies>
     </profile>
     <profile>
-      <id>hadoop-aws</id>
+      <id>hadoop-2</id>
       <activation>
         <property>
-          <name>hadoop-aws-deps</name>
+          <name>hadoop-2-deps</name>
         </property>
       </activation>
       <dependencies>
         <dependency>
           <groupId>org.apache.hadoop</groupId>
-          <artifactId>hadoop-aws</artifactId>
+          <artifactId>hadoop-client</artifactId>
           <version>${hadoop.version}</version>
         </dependency>
-        <dependency>
-          <groupId>com.amazonaws</groupId>
-          <artifactId>aws-java-sdk-bundle</artifactId>
-          <version>${aws.version}</version>
-        </dependency>
       </dependencies>
     </profile>
     <profile>
-      <id>hadoop-2</id>
-      <activation>
-        <property>
-          <name>hadoop-2-deps</name>
-        </property>
-      </activation>
+      <id>aws</id>
       <dependencies>
         <dependency>
           <groupId>org.apache.hadoop</groupId>
-          <artifactId>hadoop-client</artifactId>
+          <artifactId>hadoop-aws</artifactId>
           <version>${hadoop.version}</version>
         </dependency>
+        <dependency>
+          <groupId>com.amazonaws</groupId>
+          <artifactId>aws-java-sdk-bundle</artifactId>
+          <version>${aws.version}</version>
+        </dependency>
       </dependencies>
     </profile>
   </profiles>

diff --git a/multipart-uploader/pom.xml b/multipart-uploader/pom.xml
@@ -0,0 +1,57 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one or more
+  ~ contributor license agreements.  See the NOTICE file distributed with
+  ~ this work for additional information regarding copyright ownership.
+  ~ The ASF licenses this file to You under the Apache License, Version 2.0
+  ~ (the "License"); you may not use this file except in compliance with
+  ~ the License.  You may obtain a copy of the License at
+  ~
+  ~    http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.apache.celeborn</groupId>
+    <artifactId>celeborn-parent_${scala.binary.version}</artifactId>
+    <version>${project.version}</version>
+    <relativePath>../pom.xml</relativePath>
+  </parent>
+
+  <artifactId>celeborn-multipart-uploader_${scala.binary.version}</artifactId>
+  <packaging>jar</packaging>
+  <name>Celeborn Multipart Uploader</name>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.celeborn</groupId>
+      <artifactId>celeborn-service_${scala.binary.version}</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+    <dependency>
+        <groupId>org.apache.hadoop</groupId>
+        <artifactId>hadoop-aws</artifactId>
+        <version>${hadoop.version}</version>
+    </dependency>
+    <dependency>
+        <groupId>com.amazonaws</groupId>
+        <artifactId>aws-java-sdk-bundle</artifactId>
+        <version>${aws.version}</version>
+    </dependency>
+  </dependencies>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-dependency-plugin</artifactId>
+      </plugin>
+    </plugins>
+  </build>
+</project>