Configuring disaggregated shuffle for runtimes

Applies to :

Spark engine

Apache Gluten accelerated Spark engine

Introduction

Disaggregated shuffle service allows the bulk of the shuffle data to reside in a separate storage location (separate shared volume mount or object store) from the compute nodes thereby enabling efficient resource utilization. The Disaggregated shuffle plugin makes use of the Spark shuffle manager API that replaces the existing Shuffle manager. It registers a new ShuffleManager and ShuffleDataIO component which enables the plugin to override the Spark shuffle I/O methods.

Required permissions
You must have the User role and access to storage volume.

Procedure

The Disaggregated shuffle plugin is used by the Spark runtimes that run out of local temporary storage or experience pod failures. It is also used when the local storage resources are insufficient, or as an alternative to emptyDir to boost performance. The Disaggregated shuffle plugin supports the following storage:

  • Software Hub shared volumes
  • Any Spark-compatible storage such as cos://bucket, s3a://bucket, hdfs://, or file:///path
  1. Save the following Python sample file.
    
from pyspark.sql import SparkSession
import time

def init_spark():
    spark = SparkSession.builder.appName("auto-scale-test").getOrCreate()
    sc = spark.sparkContext
    return spark, sc

def main():
    spark, sc = init_spark()
    partitions = [10, 5, 8, 4, 9, 4, 7]
    parallelism = sc.defaultParallelism
    for num_partitions in partitions:
        print(f"Running experiment with {num_partitions} partitions leveraging {parallelism} cores.")
        data = range(1, 20000000)
        v0 = sc.parallelize(data, parallelism)
        gr = v0.groupBy(lambda x: x % num_partitions, numPartitions=num_partitions).map(lambda x: (x[0], len(x[1])))
        dd = gr.collect()
        print(f"Buckets {dd}")
        time.sleep(5)

if __name__ == "__main__":
    main()

  2. Upload the Python file to the Cloud Object Storage bucket.

  3. To enable shuffle on a shared volume within Analytics Engine, you must mount both the shared volume and specify the path within Disaggregated Shuffle. Run the following curl command to submit the Spark application.
    curl -k -X POST https://<cpd_host_name>/lakehouse/api/<api_version>/spark_engines/<spark_engine_id>/applications \
    -H 'LhInstanceId: <instance_id>' \
    -H "Authorization: ZenApiKey <api-key>" \
    -d '{  "application_details": {\
    "conf": {\
           "spark.hadoop.fs.s3a.endpoint": "s3.us-south.cloud-object-storage.appdomain.cloud",\
           "spark.hadoop.fs.s3a.access.key": "<access-key>",\
           "spark.hadoop.fs.s3a.secret.key": "<secret-key>",\
           "spark.hadoop.fs.s3a.path.style.access": "true",\
           "spark.hadoop.fs.s3a.fast.upload": "true","spark.hadoop.fs.s3a.impl": "org.apache.hadoop.fs.s3a.S3AFileSystem",\
           "spark.shuffle.manager": "org.apache.spark.shuffle.sort.DisaggregatedShuffleManager",\
           "spark.shuffle.sort.io.plugin.class": "org.apache.spark.shuffle.DisaggregatedShuffleDataIO",\
           "spark.shuffle.disaggregated.rootDir": "file:///<path-name>/",\
           "spark.shuffle.disaggregated.folderPrefixes": "1"    },\
           "application": "s3a://<cos-bucket-name>/<app-name>.py"  },
            "volumes": [{    "name": "cpd-instance::<volume-name>",
            "mount_path": "file:///<path-name>"  }]}'
    The plugin cleans up the created shuffle files automatically. However the shuffle files created when a Spark driver is killed or crashes are not removed.It is recommend to regularly clean up the shuffle directory. If the shuffle files have an object store backup, lifecycle policy can be implemented on the bucket.
    When using the v2 API, set the <api_version> parameter to v2; for the v3 API, set it to v3.

Performance Optimizations

  • If the shuffle files are stored on a distributed filesystems, it is recommend to set spark.shuffle.disaggregated.folderPrefixes to 1.

  • For object stores like IBM COS or AWS S3, it is recommend to set spark.shuffle.disaggregated.folderPrefixes to 10. For better performance on object stores, the shuffle plugin needs to be configured with the storage bucket and without any extra path added. Example, if the bucket is called “spark-shuffle-data”, then spark.shuffle.disaggregated.rootDir should be set to "s3a://spark-shuffle-data/" because the folder prefix option allows the Disaggregated Shuffle plugin to distribute reads and writes over multiple prefixes.

  • The disaggreagated shuffle plugin uses buffers of the size 8388608 bytes (configured in spark.shuffle.disaggregated.bufferSize) for each task. Buffers are written in chunks of 2097152 bytes to the underlying filesystem (configured in spark.shuffle.disaggregated.bufferChunkSize, which by default is 1/4th of the buffer size). If your filesystem requires bigger chunks, you can increase both the buffer size and the chunk size.

  • Buffer sizes to read data can be configured separately. By default, Disaggregated Shuffle uses 134217728 bytes (128 MB) per task. This value can be changed by configuring spark.shuffle.disaggregated.maxBufferSizeTask.

  • Each shuffle task pre-fetches the shuffle files in the background using a thread pool. The concurrency of the prefetcher is estimated based on the measured I/O latency. The maximum number of threads can be configured by setting spark.shuffle.disaggregated.maxConcurrencyTask (default: 5).

Limitation

Shuffle data is not cleaned up when a Spark application is forcefully stopped
If your Spark application stops unexpectedly, the shuffle data stored in your remote storage location is not automatically removed. This happens because an unexpected stop prevents the normal cleanup process from running.
You may notice unused shuffle files accumulating in your configured storage path (spark.shuffle.disaggregated.rootDir). Over time, this can lead to unnecessary storage consumption.

Workaround:

  • Periodically check your remote storage and delete any leftover shuffle directories. Shuffle data for each application is stored under a subfolder named after the application ID: <rootDir>/<prefixFolders>/<appId>.

  • If your shuffle data is stored in an object store such as IBM COS, Amazon S3, or Azure Data Lake Storage (ADLS), set up a bucket lifecycle policy to automatically delete objects under the shuffle prefix after a suitable number of days. This prevents storage from growing over time without manual intervention.

Note: Shuffle data is cleaned up automatically when your application finishes normally or is stopped gracefully. This issue affects only applications that are forcefully terminated.