03.7 Shuffle

Architect rule: Shuffle is a data redistribution boundary. It is a primary place to investigate slow, expensive, memory-heavy, or skewed Spark workloads.

1. What is shuffle?

Shuffle redistributes records across partitions so that data required by a downstream operation is colocated. For example, a groupBy on customer_id may require all records for the same customer to reach the same downstream partition.

Stage 1 → map work → Shuffle / Exchange → Stage 2 → reduce work

Apache Spark describes shuffle as redistribution across partitions and notes its network, serialization, disk, and memory costs. RDD Programming Guide

2. Why shuffle creates a stage boundary

A downstream stage generally requires a different data distribution from the upstream stage. The physical plan often exposes this as an Exchange operator.

HashAggregate
  Exchange hashpartitioning(customer_id, 200)
    HashAggregate
      Scan

The pattern is commonly: scan → partial aggregation → redistribution → final aggregation.

3. Narrow vs wide dependency

A narrow dependency can often process each child partition from a small, predictable set of parent partitions. filter, select, and withColumn commonly behave this way.

A wide dependency requires records from multiple upstream partitions. groupBy, many joins, distinct, global ordering, and repartition commonly introduce redistribution.

P0 ─┐
P1 ─┼─→ Shuffle → New partitions
P2 ─┤
P3 ─┘

Always verify the physical plan rather than relying only on an operation name.

4. Shuffle anatomy

Map side

Upstream tasks process input partitions, determine target shuffle partitions, serialize or buffer records, and produce shuffle blocks.

Reduce side

Downstream tasks fetch the blocks they need, deserialize and merge them, and execute the next operator.

Map tasks → Shuffle write → Shuffle storage/network → Shuffle read → Reduce tasks

5. Why shuffle is expensive

• Network: records may move between executors and machines.
• CPU: hashing, serialization, sorting, merging, and aggregation consume CPU.
• Disk: shuffle data can use executor-local disk and spill paths.
• Memory: aggregation and shuffle structures can create large per-task working sets.
• Scheduling: many shuffle partitions create many downstream tasks.

The fastest shuffle is often the shuffle you avoid.

6. Shuffle read, write, and spill

Shuffle write is produced by upstream tasks; shuffle read is consumed by downstream tasks. If intermediate structures do not fit comfortably in memory, Spark can spill to disk.

Input → memory structures → insufficient memory → spill → disk merge/read

Use Spark UI metrics for shuffle read/write, spill, task duration, and partition distribution.

7. Shuffle partition count

spark.sql.shuffle.partitions controls the default number of Spark SQL shuffle partitions for joins and aggregations when another mechanism does not determine the count. Apache Spark 4.3.0 documents a default of 200. Do not treat 200 as a universal production value.

spark.conf.set('spark.sql.shuffle.partitions', 400)

Too few partitions can create oversized tasks and memory pressure. Too many can create tiny tasks and scheduling overhead.

8. AQE and shuffle

Adaptive Query Execution uses runtime statistics after exchanges to improve execution. Current Databricks guidance describes AQE as enabled by default and highlights dynamic partition coalescing, skew handling, runtime join changes, and empty-relation propagation.

Initial plan → Shuffle → Runtime statistics → AQE re-optimization
                                      ├→ coalesce small partitions
                                      ├→ split eligible skewed partitions
                                      └→ improve join strategy

Do not fight AQE with unnecessary hard-coded settings. Measure first.

• Databricks AQE
• Spark SQL Performance Tuning

9. Shuffle skew

Skew occurs when one or a few shuffle partitions contain dramatically more data than others.

Partition 0: 2 GB
Partition 1: 2 GB
Partition 2: 1.5 GB
Partition 3: 80 GB  ← straggler

Common causes include hot keys, low-cardinality join keys, concentrated customers, NULL-heavy keys, and uneven source data.

Databricks recommends comparing maximum task duration with percentile durations in Spark UI. AQE can split eligible skewed shuffle partitions for supported shuffle joins. Databricks skew guidance

10. Shuffle in joins

For orders joined to customers on customer_id, Spark may redistribute one or both sides when compatible partitioning is unavailable and a broadcast strategy is not selected.

Orders ──→ hash(customer_id) ─┐
                              ├─→ Shuffle → join partitions → Join
Customers → hash(customer_id) ─┘

A broadcast join can avoid shuffling a large side when one side is genuinely small enough. AQE can also make runtime join decisions from observed statistics.

11. Reduce data before shuffle

Bad conceptual pattern:

5 TB scan → Shuffle 5 TB → Filter

Better when semantics permit:

5 TB scan → Filter / Project / Partial Aggregate → smaller Shuffle

Use predicate pushdown, column pruning, early aggregation, selective joins, and unnecessary-column removal to reduce bytes crossing the boundary.

12. Repartition is an explicit shuffle request

df2 = df.repartition(400, 'customer_id')

Repartition deliberately redistributes data. It can correct imbalance, increase parallelism, prepare downstream processing, or control output layout. It can also make a job slower if redistribution was unnecessary.

13. Windows and global ordering

A window such as ROW_NUMBER partitioned by customer_id and ordered by order_date can require redistribution and sorting. Large or highly skewed window partitions can therefore become expensive.

Similarly, orderBy requests global ordering and can require substantial redistribution and sorting. If global ordering is not required, a local sort has different semantics and potentially lower cost.

14. Push-based shuffle

Apache Spark also provides push-based shuffle for suitable workloads. Shuffle blocks can be pushed to an external shuffle service and merged by shuffle partition, improving some large disk-I/O-heavy workloads. Spark Configuration

Treat advanced shuffle features as workload-specific optimizations.

15. Production troubleshooting workflow

1. Find the slow stage in Spark UI.
2. Check shuffle read, shuffle write, task count, and input size.
3. Compare median, 75th percentile, and maximum task duration.
4. Check memory and disk spill.
5. Read the physical plan for Exchange, joins, sorts, and aggregations.
6. Classify the issue: unnecessary shuffle, partition sizing, skew, join strategy, excessive pre-shuffle data, window cost, or global ordering.
7. Apply one targeted change.
8. Re-run and compare runtime, shuffle bytes, spill, task distribution, and compute cost.

16. Hands-on lab — identify shuffle

from pyspark.sql import functions as F

orders = (
    spark.range(0, 5_000_000)
    .withColumn('customer_id', (F.col('id') % 100_000).cast('long'))
    .withColumn('amount', (F.col('id') % 1000).cast('double'))
)

result = (
    orders.groupBy('customer_id')
          .agg(F.sum('amount').alias('revenue'))
)

result.explain('formatted')
result.count()

Find Exchange in the physical plan. Then use Spark UI to inspect stage boundaries, shuffle read/write, task count, and task-duration distribution.

17. Hands-on lab — observe skew

skewed = (
    spark.range(0, 10_000_000)
    .withColumn(
        'customer_id',
        F.when(F.col('id') < 9_000_000, F.lit('HOT'))
         .otherwise(F.col('id').cast('string'))
    )
)

result = skewed.groupBy('customer_id').count()
result.count()

Inspect the stage for a long-tail task distribution and connect it to the HOT key.

18. Common anti-patterns

• Shuffle is always bad — false; many distributed operations require it.
• Repartition before every join — can add unnecessary redistribution.
• Increase shuffle partitions blindly — can create excessive task overhead.
• Ignore skew — one straggler can determine stage runtime.
• Increase cluster size to hide a shuffle problem — more compute does not remove unnecessary data movement.
• coalesce(1) for large production output — can serialize a large final write.

19. Architect decision framework

Do I need this redistribution?
  No → remove/rewrite
  Yes
   ↓
Can I reduce data before shuffle?
  Yes → filter/project/aggregate early
  No
   ↓
Can a better join strategy avoid large redistribution?
  Yes → broadcast/AQE/etc.
  No
   ↓
Is the distribution balanced?
  No → address skew / partition sizing
  Yes → validate shuffle partitions and compute size

20. Shuffle cost model

Shuffle cost ≈ network + serialization + sorting/hashing
             + disk I/O + memory pressure + scheduling + skew

Measure before and after: shuffle bytes, stage duration, task-duration spread, spill, CPU, disk utilization, and compute cost.

21. Interview questions

1. What is a Spark shuffle?
2. Why does groupBy usually require a shuffle?
3. What is an Exchange operator?
4. What is the difference between narrow and wide dependency?
5. What happens during shuffle write and shuffle read?
6. Why can shuffle cause spill?
7. What is spark.sql.shuffle.partitions?
8. How does AQE optimize shuffle partitions?
9. How does AQE handle skew?
10. How can you reduce shuffle volume without changing business semantics?
11. A 10 TB join takes 45 minutes. How do you investigate it?
12. One task runs 30× longer than the others. What do you check?

22. Cheat sheet

Concept	Mental model
Shuffle	Redistribute data across partitions
Exchange	Physical-plan signal for redistribution
Narrow dependency	Little/no redistribution
Wide dependency	Downstream partition depends on many upstream partitions
Shuffle write	Upstream tasks produce shuffle blocks
Shuffle read	Downstream tasks fetch shuffle blocks
Spill	Intermediate data exceeds practical in-memory capacity
Skew	Uneven partition/work distribution
AQE	Runtime re-optimization using actual statistics
Best optimization	Avoid unnecessary data movement

23. The architect mental model

Operation → Required data distribution → Exchange / Shuffle
        → Network + CPU + disk + memory → New partitions
        → Downstream tasks → Stage runtime

When a Spark workload is slow, ask: Where is data moving, why is it moving, how much is moving, and is the resulting distribution balanced?

Next

Continue to 03.8 Spark UI — where we will prove these conclusions using real execution metrics, stages, tasks, shuffle, spill, skew, and physical-plan evidence.