Shuffles

Recognize when rows must move and reduce unnecessary data before an exchange.

Step 1 of 6 · Learn

Reduce shuffle input

A grouped sum can combine values locally before sending partial results across an exchange. Filtering individual orders first reduces the input to that work. Shuffles can involve serialization, network traffic, and disk I/O; six teaching rows do not model production timing. Final display sorting can introduce a separate exchange.

Lesson reference: PySpark and Scala

Reduce shuffle input

PySpark example

from pyspark.sql.functions import col, sum, count, lit

result = orders.filter(col("total") >= 100).groupBy("status").agg(
    sum("total").alias("total_value")
).orderBy("status")
result.show(truncate=False)

Scala example

import org.apache.spark.sql.functions._

val result = orders.filter(col("total") >= 100).groupBy("status").agg(
    sum("total").alias("total_value")
).orderBy("status")
result.show(false)

Deduplicate across partitions

distinct must identify duplicates across all input partitions, not only adjacent rows. Spark commonly uses local and final aggregation with an exchange. Selecting only the required key first avoids deduplicating entire order rows. Already-compatible partitioning can change the physical work; do not infer a fixed exchange count from source code alone.

PySpark example

from pyspark.sql.functions import col, sum, count, lit

result = orders.select(col("status").cast("string").alias("key")).distinct().orderBy("key")
result.show(truncate=False)

Scala example

import org.apache.spark.sql.functions._

val result = orders.select(col("status").cast("string").alias("key")).distinct().orderBy("key")
result.show(false)

Sort globally

orderBy establishes ordering across the whole result. Spark generally distributes rows into ordered ranges and sorts within them. sortWithinPartitions only orders each partition and does not guarantee global order. A small limit may let Spark choose a different top-K strategy, so this example returns every row.

PySpark example

from pyspark.sql.functions import col, sum, count, lit

result = orders.select("order_id", "total").orderBy(
    col("total").asc(), col("order_id").asc()
)
result.show(truncate=False)

Scala example

import org.apache.spark.sql.functions._

val result = orders.select("order_id", "total").orderBy(
    col("total").asc(), col("order_id").asc()
)
result.show(false)

example.pyPySpark

from pyspark.sql.functions import col, sum, count, lit result = orders.filter(col("total") >= 100).groupBy("status").agg( sum("total").alias("total_value") ).orderBy("status") result.show(truncate=False)

1. Partial workCombine matching keys inside input partitions where possible.

2. ExchangeRedistribute by key so a group can be completed.

3. Final workCombine partial results, then sort the small output for display.

Source orders6 rows

order_id	customer_id	status	total	item_count
1001	101	Delivered	89.5	2
1002	102	Shipped	149	3
1003	101	Cancelled	35	1
1004	103	Delivered	219.99	4
1005	104	Delivered	49.99	1
1006	105	Shipped	120	2

Reduce shuffle input

Result2 rows

status	total_value
Delivered	219.99
Shipped	269

The example sends only qualifying orders into aggregation: 1002, 1004, and 1006.

The example is loaded in the editor. Run it as written, then try a small change.

Runs on the Spark backend. First startup may take a moment.

Run your code to see the result.