[SHUFFLE] [WIP] Prototype: store shuffle file on external storage like S3

assembly/pom.xml

@@ -268,5 +268,17 @@
         </dependency>
       </dependencies>
     </profile>
+
+    <!-- Profile to include external shuffle storage jar file in Spark distribution -->
+    <profile>
+      <id>external-shuffle-storage</id>
+      <dependencies>
+        <dependency>
+          <groupId>org.apache.spark</groupId>
+          <artifactId>external-shuffle-storage_${scala.binary.version}</artifactId>
+          <version>${project.version}</version>
+        </dependency>
+      </dependencies>
+    </profile>
   </profiles>
 </project>

external-shuffle-storage/README.md

@@ -0,0 +1,55 @@
+# External Shuffle Storage
+
+This module provides support to store shuffle files on external shuffle storage like S3. It helps Dynamic
+Allocation on Kubernetes. Spark driver could release idle executors without worrying about losing
+shuffle data because the shuffle data is store on external shuffle storage which are different 
+from executors.
+
+This module implements a new Shuffle Manager named as StarShuffleManager, and copies a lot of codes
+from Spark SortShuffleManager. This is for a quick prototype. We want to use this as an example to discuss
+with Spark community and get feedback. We will work with the community to remove code duplication later
+and make StarShuffleManager more integrated with Spark code.
+
+## How to Build Spark Distribution with StarShuffleManager jar File
+
+Follow [Building Spark](https://spark.apache.org/docs/latest/building-spark.html) instructions,
+with extra `-Pexternal-shuffle-storage` to generate the new shuffle implementation jar file.
+
+Following is one command example to use `dev/make-distribution.sh` under Spark repo root directory:
+
+```
+./dev/make-distribution.sh --name spark-with-external-shuffle-storage --pip --tgz -Phive -Phive-thriftserver -Pkubernetes -Phadoop-3.2 -Phadoop-cloud -Dhadoop.version=3.2.0 -Pexternal-shuffle-storage
+```
+
+If you want to build a Spark docker image, you could unzip the Spark distribution tgz file, and run command like following:
+
+```
+./bin/docker-image-tool.sh -t spark-with-external-shuffle-storage build
+```
+
+This command creates `external-shuffle-storage_xxx.jar` file for StarShuffleManager
+under `jars` directory in the generated Spark distribution. Now you could use this Spark 
+distribution to run your Spark application with external shuffle storage.
+
+## How to Run Spark Application With External Shuffle Storage in Kubernetes
+
+### Run Spark Application With S3 as External Shuffle Storage and Dynamic Allocation
+
+Add configure to your Spark application like following (you need to adjust the values based on your environment):
+
+```
+spark.shuffle.manager=org.apache.spark.shuffle.StarShuffleManager
+spark.shuffle.star.rootDir=s3://my_bucket_name/my_shuffle_folder
+spark.dynamicAllocation.enabled=true
+spark.dynamicAllocation.shuffleTracking.enabled=true
+spark.dynamicAllocation.shuffleTracking.timeout=1
+```
+
+### How to specify AWS region for the S3 files
+
+Add Spark config like following:
+
+```
+spark.hadoop.fs.s3a.endpoint.region=us-west-2
+```
+

external-shuffle-storage/pom.xml

@@ -0,0 +1,143 @@
+<?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.spark</groupId>
+        <artifactId>spark-parent_2.12</artifactId>
+        <version>3.3.0-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>external-shuffle-storage_2.12</artifactId>
+    <packaging>jar</packaging>
+    <name>External Shuffle Storage</name>
+    <url>http://spark.apache.org/</url>
+
+    <properties>
+        <sbt.project.name>external-shuffle-storage</sbt.project.name>
+        <build.testJarPhase>none</build.testJarPhase>
+        <build.copyDependenciesPhase>package</build.copyDependenciesPhase>
+        <hadoop.deps.scope>provided</hadoop.deps.scope>
+        <hive.deps.scope>provided</hive.deps.scope>
+        <parquet.deps.scope>provided</parquet.deps.scope>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-core_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-core_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-sql_${scala.binary.version}</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-math3</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.scalacheck</groupId>
+            <artifactId>scalacheck_${scala.binary.version}</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.scala-lang</groupId>
+            <artifactId>scala-library</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-all</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-java-sdk-s3</artifactId>
+            <version>1.11.975</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testng</groupId>
+            <artifactId>testng</artifactId>
+            <version>6.14.3</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory>
+        <testOutputDirectory>target/scala-${scala.binary.version}/test-classes</testOutputDirectory>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-deploy-plugin</artifactId>
+                <configuration>
+                    <skip>true</skip>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-install-plugin</artifactId>
+                <configuration>
+                    <skip>true</skip>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <configuration>
+                    <outputDirectory>${jars.target.dir}</outputDirectory>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>

external-shuffle-storage/src/main/java/org/apache/spark/starshuffle/ByteBufUtils.java

@@ -0,0 +1,46 @@
+/*
+ * This file is copied from Uber Remote Shuffle Service
+ * (https://github.com/uber/RemoteShuffleService) and modified.
+ *
+ * Licensed 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.
+ */
+
+package org.apache.spark.starshuffle;
+
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+
+public class ByteBufUtils {
+  public static final void writeLengthAndString(ByteBuf buf, String str) {
+    if (str == null) {
+      buf.writeInt(-1);
+      return;
+    }
+
+    byte[] bytes = str.getBytes(StandardCharsets.UTF_8);
+    buf.writeInt(bytes.length);
+    buf.writeBytes(bytes);
+  }
+
+  public static final String readLengthAndString(ByteBuf buf) {
+    int length = buf.readInt();
+    if (length == -1) {
+      return null;
+    }
+
+    byte[] bytes = new byte[length];
+    buf.readBytes(bytes);
+    return new String(bytes, StandardCharsets.UTF_8);
+  }
+}