[SPARK-9850] Adaptive execution in Spark

• Type: Epic
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Component/s: Spark Core, SQL
• Assignee: Yin Huai
• Reporter: Matei Alexandru Zaharia
• Created: 11/Aug/15 23:39
• Updated: 17/Dec/20 18:23

Description

Query planning is one of the main factors in high performance, but the current Spark engine requires the execution DAG for a job to be set in advance. Even with cost­-based optimization, it is hard to know the behavior of data and user-defined functions well enough to always get great execution plans. This JIRA proposes to add adaptive query execution, so that the engine can change the plan for each query as it sees what data earlier stages produced.

We propose adding this to Spark SQL / DataFrames first, using a new API in the Spark engine that lets libraries run DAGs adaptively. In future JIRAs, the functionality could be extended to other libraries or the RDD API, but that is more difficult than adding it in SQL.

I've attached a design doc by Yin Huai and myself explaining how it would work in more detail.

Issue Links

• is a parent of: SPARK-28231 Do not modify the number of partitions set by the user when using repartition
• is duplicated by: SPARK-4630 Dynamically determine optimal number of partitions
• is related to: SPARK-23128 The basic framework for the new Adaptive Query Execution
• is related to: SPARK-31412 New Adaptive Query Execution in Spark SQL

Sub-Tasks

• Support submitting map stages individually in DAGScheduler: Resolved
• Let reduce tasks fetch multiple map output partitions: Resolved
• Introduce an ExchangeCoordinator to estimate the number of post-shuffle partitions.: Resolved
• Aggregation: Determine the number of reducers at runtime: Resolved
• Join: Determine the join strategy (broadcast join or shuffle join) at runtime: Resolved
• Join: Determine the number of reducers used by a shuffle join operator at runtime: Resolved
• Join: Handling data skew: Resolved
• Improve the way that we plan shuffled join: Resolved
• Allow shuffle readers to request data from just one mapper: Resolved