Shuffle

Overview

The Shuffle is the most expensive operation in Apache Spark. Learn why it occurs, how it moves data over the network, and how to reduce shuffle overhead.

What You Will Learn

In this lesson, you will learn:

Shuffle Mechanism: Writing partitions to local disk and reading them over network.
Shuffle Writes and Reads: The stages of shuffle transport.
Reducing Shuffles: Strategies like broadcast joins to skip shuffles.

Detailed Concept Explanation

A Shuffle occurs when Spark needs to redistribute data across the cluster to group records by key (e.g. for joins, groupBy, or repartition).

text

Worker A (Part 1)  ---\                     /--> Worker A (New Part 1)
                       ====== Network Shuffle ======
Worker B (Part 2)  ---/                     \--> Worker B (New Part 2)

Why Shuffling is Expensive

Disk I/O (Shuffle Write): Executors must serialize and write partition outputs to their local disk first.
Network I/O (Shuffle Read): Executor workers fetch their target key partitions from other nodes over the network.
Serialization Overhead: Converting Java objects to byte arrays and back.
Garbage Collection: Recreating objects on the JVM during shuffle reads can trigger GC pauses.

Code Examples

Python (PySpark) Implementation

python

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ShuffleConcept").getOrCreate()

data = [("A", 10), ("B", 20), ("A", 30)]
df = spark.createDataFrame(data, ["key", "value"])

# Trigger a shuffle by running groupBy sum
grouped = df.groupBy("key").sum("value")
grouped.show()

Expected Output

text

+---+----------+
|key|sum(value)|
+---+----------+
|  A|        40|
|  B|        20|
+---+----------+

Execution Plan Diagram (Python & Scala)

Execution Plan Diagram

SparkSession.builder

createDataFrame

groupBy(key).sum() [Triggers Shuffle Write & Read]

show()

Scala Implementation

scala

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().appName("ShuffleScala").getOrCreate()
import spark.implicits._

val df = Seq(("A", 10), ("B", 20), ("A", 30)).toDF("key", "value")
val grouped = df.groupBy("key").sum("value")
grouped.show()

Best Practices

Map-side combining: When using the RDD API, use reduceByKey instead of groupByKey to combine values locally before shuffling.
Broadcast Joins: When joining a large table with a small table, use a broadcast join to eliminate the shuffle entirely.