Firestorm is a Remote Shuffle Service, and provides the capability for Apache Spark applications to store shuffle data on remote servers.

Firestorm contains coordinator cluster, shuffle server cluster and remote storage(eg, HDFS) if necessary.

Coordinator will collect status of shuffle server and do the assignment for the job.

Shuffle server will receive the shuffle data, merge them and write to storage.

Depend on different situation, Firestorm supports Memory & Local, Memory & Remote Storage(eg, HDFS), Local only, Remote Storage only.

• Spark driver ask coordinator to get shuffle server for shuffle process

• Spark task write shuffle data to shuffle server with following step:

  1. Send KV data to buffer
  2. Flush buffer to queue when buffer is full or buffer manager is full
  3. Thread pool get data from queue
  4. Request memory from shuffle server first and send the shuffle data
  5. Shuffle server cache data in memory first and flush to queue when buffer manager is full
  6. Thread pool get data from queue
  7. Write data to storage with index file and data file
  8. After write data, task report all blockId to shuffle server, this step is used for data validation later
  9. Store taskAttemptId in MapStatus to support Spark speculation

• Depend on different storage type, spark task read shuffle data from shuffle server or remote storage or both of them.

The shuffle data is stored with index file and data file. Data file has all blocks for specific partition and index file has metadata for every block.

Current support Spark 2.3.x, Spark 2.4.x, Spark3.0.x, Spark 3.1.x, Spark 3.2.x

Note: To support dynamic allocation, the patch(which is included in client-spark/patch folder) should be applied to Spark

Firestorm is built using Apache Maven. To build it, run:

mvn -DskipTests clean package

To package the Firestorm, run:

./build_distribution.sh

rss-xxx.tgz will be generated for deployment

1. unzip package to RSS_HOME
2. update RSS_HOME/bin/rss-env.sh, eg,

     JAVA_HOME=<java_home>
     HADOOP_HOME=<hadoop home>
     XMX_SIZE="16g"
   

3. update RSS_HOME/conf/coordinator.conf, eg,

     rss.rpc.server.port 19999
     rss.jetty.http.port 19998
     rss.coordinator.server.heartbeat.timeout 30000
     rss.coordinator.app.expired 60000
     rss.coordinator.shuffle.nodes.max 5
     rss.coordinator.exclude.nodes.file.path RSS_HOME/conf/exclude_nodes
   

4. start Coordinator

    bash RSS_HOME/bin/start-coordnator.sh
   

1. unzip package to RSS_HOME
2. update RSS_HOME/bin/rss-env.sh, eg,

     JAVA_HOME=<java_home>
     HADOOP_HOME=<hadoop home>
     XMX_SIZE="80g"
   

3. update RSS_HOME/conf/server.conf, the following demo is for memory + local storage only, eg,

     rss.rpc.server.port 19999
     rss.jetty.http.port 19998
     rss.rpc.executor.size 2000
     rss.storage.type MEMORY_LOCALFILE
     rss.coordinator.quorum <coordinatorIp1>:19999,<coordinatorIp2>:19999
     rss.storage.basePath /data1/rssdata,/data2/rssdata....
     rss.server.flush.thread.alive 5
     rss.server.flush.threadPool.size 10
     rss.server.buffer.capacity 40g
     rss.server.read.buffer.capacity 20g
     rss.server.heartbeat.timeout 60000
     rss.server.heartbeat.interval 10000
     rss.rpc.message.max.size 1073741824
     rss.server.preAllocation.expired 120000
     rss.server.commit.timeout 600000
     rss.server.app.expired.withoutHeartbeat 120000
   

4. start Shuffle Server

    bash RSS_HOME/bin/start-shuffle-server.sh
   

1. Add client jar to Spark classpath, eg, SPARK_HOME/jars/

   The jar for Spark2 is located in <RSS_HOME>/jars/client/spark2/rss-client-XXXXX-shaded.jar

   The jar for Spark3 is located in <RSS_HOME>/jars/client/spark3/rss-client-XXXXX-shaded.jar

2. Update Spark conf to enable Firestorm, the following demo is for local storage only, eg,

   spark.shuffle.manager org.apache.spark.shuffle.RssShuffleManager
   spark.rss.coordinator.quorum <coordinatorIp1>:19999,<coordinatorIp2>:19999
   spark.rss.storage.type MEMORY_LOCALFILE
   

To support spark dynamic allocation with Firestorm, spark code should be updated. There are 2 patches for spark-2.4.6 and spark-3.1.2 in spark-patches folder for reference.

After apply the patch and rebuild spark, add following configuration in spark conf to enable dynamic allocation:

spark.shuffle.service.enabled false
spark.dynamicAllocation.enabled true

The important configuration is listed as following.

Firestorm is under the Apache License Version 2.0. See the LICENSE file for details.

For more information about contributing issues or pull requests, see Firestorm Contributing Guide.

We provide free support for users using this project. If you want to join user wechat group for further help and collaboration. You can scan the following QR code or search wechatID xinghuojihua_01, add our assistant on wechat, and remark Firestorm. The assistant will help you join our wechat group.