What you will be able to do
- Decide whether a slow stage is I/O bound using the 3 MB per core per second rule of thumb
- Recognise the common causes of a long stage that runs as a single task
- Use SQL DAG operator metrics to spot small files, cartesian joins and exploding joins
- Explain how adaptive query execution detects and splits skewed partitions during joins
1.Is the stage I/O bound?
A slow stage with no spill and no skew can still be slow because it moves a lot of data. To judge whether the I/O is high, take the largest value among the stage's Input, Output, Shuffle Read and Shuffle Write columns. Divide it by the total number of worker cores in the cluster, then divide that by the stage's duration in seconds. The rule of thumb is that each core can read and write about 3 MB per second. If your result is close to that, the stage is probably I/O bound.
Checkpoint 1 of 3· Check yourself
A stage's largest I/O column is 36 GB (about 36,000 MB). The cluster has 100 worker cores and the stage ran for 120 seconds. What does the guide's rule of thumb suggest?
36,000 MB ÷ 100 cores ÷ 120 s = 3 MB per core per second. That matches the rule of thumb for how much each core can read and write, so the stage is probably I/O bound.
“If the result is around 3 MB, then you're probably I/O bound.”Source: docs.databricks.com
Which column is high tells you where the time goes. Heavy Input means the stage spends a lot of time reading data. The guide's fixes include using Delta, using liquid clustering for better data skipping, trying Photon, and making the query more selective. Heavy Output means the stage spends a lot of time writing, often because it rewrites data, for example in a merge. Photon or a larger cluster can help with that. This objective is about identifying the bottleneck, so it's enough to know which column points to which kind of problem.
The read-side advice. A lot of Input means the stage spends most of its time reading. The first step is to identify which data it reads, then speed up the read: Delta, liquid clustering for data skipping, Photon, a more selective query, or the Delta cache if the same data is read repeatedly.
Sources1
2.One task, or slow with little I/O
A long stage with just one task is likely a problem in itself. While that task runs, only one CPU is busy and the rest of the cluster may be idle. The usual causes are an expensive UDF on small data, a window function without PARTITION BY, reading an unsplittable file type such as gzip, setting the multiLine option when reading JSON or CSV, schema inference on a large file, or calling repartition(1) or coalesce(1).
The Shuffle Read and Shuffle Write columns you compared earlier measure a shuffle, which is Spark's mechanism for re-distributing data so that it is grouped differently across partitions. Joins, aggregations and repartition operations trigger it, because all the rows for a key must end up together. A shuffle typically copies data across executors and machines, so it is costly: it involves disk I/O, data serialization and network I/O. It is also where spill most often happens. Spill means Spark ran low on execution memory and started moving data from memory to disk, and that is most common during data shuffling. That is why a large Shuffle Read or Shuffle Write on a long stage is worth a closer look.
A stage with many tasks and little I/O can also be slow. Common causes are reading or writing many small files, slow UDFs, cartesian joins and exploding joins. Most of these show up in the SQL DAG, which you open from Associated SQL Query at the top of the job page. Opening an operator shows metrics such as number of output rows, peak memory and spill size. Some nodes also show time spent. That time is cumulative across executors, not elapsed clock time, but it still correlates with clock time and cost.
| What you see in the DAG | Likely bottleneck |
|---|---|
| A scan operator reading tens of thousands of files or more | Small file problem (files should be no less than 8MB) |
| A cartesian join or nested loop join node | A very expensive join; check that it's intended |
| A few rows going into a node and magnitudes more coming out | An exploding join or explode() |
Small files are usually caused by partitioning on too many columns or on a high-cardinality column. Sometimes running OPTIMIZE is enough. Databricks also recommends enabling predictive optimization and reconsidering the file layout.
Checkpoint 2 of 3· Match them up
Match each observation to the bottleneck it most likely indicates.
Tap a term, then the definition that fits it.
Gzip can't be read in parts, so the read runs as one big task. A file count in the tens of thousands points to small files. A large jump in row count points to an explode, and cartesian or nested loop joins are expensive by nature.
“If you're reading tens of thousands of files or more, you may have a small file problem.”Source: docs.databricks.com
3.How AQE detects and splits skewed partitions
Some skew is handled before you ever see it in the Spark UI. Adaptive query execution (AQE) re-optimizes a query while it runs. It is enabled by default. One of its features is skew handling for sort merge joins and shuffle hash joins: AQE splits skewed tasks, replicating them if needed, into tasks of roughly even size. AQE applies only to non-streaming queries that contain at least one exchange (usually a join, aggregate or window), at least one subquery, or both. So a skewed streaming query, or a skewed stage outside a join, won't get this automatic split.
AQE ties back to the shuffle metrics you read in the Spark UI because the number of shuffle partitions is hard to tune. If there are too few partitions, each one is very large and the tasks that process them may need to spill data to disk, which slows the query. If there are too many, each partition is tiny and the stage pays for many small network fetches and extra scheduler work. AQE uses the statistics from the finished shuffle to coalesce small partitions and to split skewed ones, which is why a stage that still shows spill or one straggler task after AQE is worth investigating.
| Property | Meaning | Default |
|---|---|---|
| spark.sql.adaptive.skewJoin.enabled | Whether skew join handling is enabled | true |
| spark.sql.adaptive.skewJoin.skewedPartitionFactor | Multiplied by the median partition size to help decide whether a partition is skewed | 5 |
| spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes | A size threshold that also helps decide whether a partition is skewed | 256MB |
| spark.sql.adaptive.advisoryPartitionSizeInBytes | Target size of partitions after coalescing | 64MB |
AQE flags a partition as skewed only when both conditions hold. The partition must be larger than the factor times the median partition size, and it must also be larger than the byte threshold. With the defaults, a 300MB partition counts as skewed only if the median is under 60MB. AQE also coalesces very small post-shuffle partitions, because tiny tasks have poor I/O throughput and pay more scheduling overhead.
Checkpoint 3 of 3· Check yourself
With the default AQE settings, the median partition in a sort merge join is 10MB and one partition is 200MB. Will AQE treat that partition as skewed?
200MB is more than 5 × 10MB, but it isn't larger than skewedPartitionThresholdInBytes (256MB). AQE needs both conditions to be true, so it doesn't split this partition.
“A partition is considered skewed when both”Source: docs.databricks.com
Exam traps
Each one states something that sounds right. Open it to see what is actually true.
1.The time shown on a SQL DAG node is the wall-clock time that operator took.Why is that wrong?
DAG node times are cumulative. They add up executor time across all tasks, so they correlate with clock time and cost but are not elapsed time.
Covered in One task, or slow with little I/O
2.A long stage with a single task just means the stage has little work to do, so it isn't worth investigating.Why is that wrong?
One long task means only one CPU is working while the rest of the cluster may be idle. Common causes are gzip input, multiLine reads, a window function without PARTITION BY, or coalesce(1).
Covered in One task, or slow with little I/O
Sources
Every claim above is drawn from one of these pages, quoted as it was written on the date shown.
- 1.
“Generally each core can read and write about 3 MBs per second.”
↩︎ Is the stage I/O bound?“Divide your biggest I/O column by the number of cluster worker cores, then divide that by duration seconds.”
↩︎ Is the stage I/O bound?“If you see a lot of input into your stage, that means you're spending a lot of time reading data.”
↩︎ Is the stage I/O bound?“Use liquid clustering for better data skipping.”
↩︎ Is the stage I/O bound?“If the result is around 3 MB, then you're probably I/O bound.”
↩︎ Checkpoint - 2.
“Gzip is an example of an unsplittable file type.”
↩︎ One task, or slow with little I/O“While this one task is running only one CPU is utilized and the rest of the cluster may be idle.”
↩︎ Exam trap 2 - 3.
“It is most common during data shuffling.”
↩︎ One task, or slow with little I/O - 4.https://spark.apache.org/docs/latest/rdd-programming-guide.htmlSecondary source
“The shuffle is Spark’s mechanism for re-distributing data so that it’s grouped differently across partitions.”
↩︎ One task, or slow with little I/O“The Shuffle is an expensive operation since it involves disk I/O, data serialization, and network I/O.”
↩︎ One task, or slow with little I/O - 5.https://docs.databricks.com/aws/en/optimizations/spark-ui-guide/slow-spark-stage-low-ioOfficial docs
“When you open an operator, it shows metrics such as number of output rows, peak memory, and spill size.”
↩︎ One task, or slow with little I/O“Your files should be no less than 8MB.”
↩︎ One task, or slow with little I/O“If you see a cartesian join or nested loop join in your DAG, you should know that these joins are very expensive.”
↩︎ One task, or slow with little I/O“you may be suffering from an exploding join or explode()”
↩︎ One task, or slow with little I/O“These times are cumulative, so they represent the sum of the time spent by executors on each task, not the clock (elapsed) time.”
↩︎ Exam trap 1“If you're reading tens of thousands of files or more, you may have a small file problem.”
↩︎ Checkpoint - 6.https://docs.databricks.com/aws/en/optimizations/aqeOfficial docs
“Dynamically handles skew in sort merge join and shuffle hash join by splitting (and replicating if needed) skewed tasks into roughly evenly sized tasks.”
↩︎ How AQE detects and splits skewed partitions“AQE is enabled by default.”
↩︎ How AQE detects and splits skewed partitions“A factor that when multiplied by the median partition size contributes to determining whether a partition is skewed”
↩︎ How AQE detects and splits skewed partitions“A partition is considered skewed when both”
↩︎ Checkpoint - 7.https://www.databricks.com/blog/2020/05/29/adaptive-query-execution-speeding-up-spark-sql-at-runtime.htmlSecondary source
“the tasks to process these large partitions may need to spill data to disk”
↩︎ How AQE detects and splits skewed partitions