[SPARK-25299] Use remote storage for persisting shuffle data

• Type: New Feature
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 3.1.0
• Component/s: Shuffle, Spark Core
• Labels: SPIP
• Assignee: Unassigned
• Reporter: Matt Cheah
• Created: 01/Sep/18 00:25
• Updated: 10/Dec/21 22:01

Description

In Spark, the shuffle primitive requires Spark executors to persist data to the local disk of the worker nodes. If executors crash, the external shuffle service can continue to serve the shuffle data that was written beyond the lifetime of the executor itself. In YARN, Mesos, and Standalone mode, the external shuffle service is deployed on every worker node. The shuffle service shares local disk with the executors that run on its node.

There are some shortcomings with the way shuffle is fundamentally implemented right now. Particularly:

• If any external shuffle service process or node becomes unavailable, all applications that had an executor that ran on that node must recompute the shuffle blocks that were lost.
• Similarly to the above, the external shuffle service must be kept running at all times, which may waste resources when no applications are using that shuffle service node.
• Mounting local storage can prevent users from taking advantage of desirable isolation benefits from using containerized environments, like Kubernetes. We had an external shuffle service implementation in an early prototype of the Kubernetes backend, but it was rejected due to its strict requirement to be able to mount hostPath volumes or other persistent volume setups.

In the following architecture discussion document (note: not an SPIP), we brainstorm various high level architectures for improving the external shuffle service in a way that addresses the above problems. The purpose of this umbrella JIRA is to promote additional discussion on how we can approach these problems, both at the architecture level and the implementation level. We anticipate filing sub-issues that break down the tasks that must be completed to achieve this goal.

Edit June 28 2019: Our SPIP is here: https://docs.google.com/document/d/1d6egnL6WHOwWZe8MWv3m8n4PToNacdx7n_0iMSWwhCQ/edit

Issue Links

• blocks: SPARK-24432 Add support for dynamic resource allocation
• is blocked by: SPARK-26268 Decouple shuffle data from Spark deployment
• relates to: SPARK-27941 Serverless Spark in the Cloud
• relates to: SPARK-31924 Create remote shuffle service reference implementation
• relates to: SPARK-1529 Support DFS based shuffle in addition to Netty shuffle
• links to: [Github] Pull Request #22777 (ifilonenko)
• links to: Shuffle Metadata Tracking Discussion

Sub-Tasks

• Shuffle Storage API: Writer API and usage in BypassMergeSortShuffleWriter: Resolved
• Shuffle Storage API: Reads: Open
• Shuffle Storage API: Driver Lifecycle: Resolved
• Shuffle Storage API: Shuffle Cleanup: Open
• Make Javadoc in org.apache.spark.shuffle.api visible: Resolved
• Shuffle Storage API: Use writer API in UnsafeShuffleWriter: Resolved
• Don't hold a reference to two partitionLengths arrays: Resolved
• Shuffle Writer API: Indeterminate shuffle support in ShuffleMapOutputWriter: Resolved
• Shuffle storage API: Use API in SortShuffleWriter: Resolved
• Mark new Shuffle apis as @Experimental (instead of @Private): Open
• Register shuffle map output metadata with a shuffle output tracker: In Progress
• Use Spark plugin support to manage shuffle plugin lifecycle: In Progress
• Return map output metadata from shuffle writers: Resolved
• Shuffle Storage API: Dynamic updates of shuffle metadata: Open
• Abstract Location in MapStatus to enable support for custom storage: In Progress
• Define local/hostlocal/remote fetch for custom storage: Open
• Support json serde for the custom location: Open
• Allow ShuffleDriverComponent to declare if shuffle data is reliably stored: Resolved