Wide Transformations
A wide transformation produces output rows that depend on rows from more than one input partition. Summing profit per region has to add up every US order, and US orders sit wherever the file reader placed them, spread across many partitions. The task holding one of those partitions can add up the US rows it sees, but it cannot know whether its subtotal is the whole answer, because another partition may hold US rows it has never seen.
7 orders in 2 partitions
Spark reads an orders table of 7 rows as 2 partitions. The first holds orders 1 to 4, where 1 and 3 are US orders and 2 is EU, with 4 from APAC. The second holds orders 5 to 7. Order 5 is from EU and order 6 from LATAM; order 7 is another US order. Ask for the list of order ids per region and the difficulty shows itself: US appears in both partitions, and so does EU. No task can write the US row of the result from the rows it holds, and the same is true for EU.
So Spark moves rows. Each task reads the region of every row it holds and sends the row to the task that will own that region, so that all 3 US orders arrive at one task and both EU orders arrive at another. Only then can a task write the US row, with order 7 next to orders 1 and 3, and be certain the list is complete. That movement of rows between tasks is a shuffle, and the next section takes it apart.
(orders
.groupBy("region")
.agg(F.collect_list("order_id").alias("order_ids"))
.collect())| order_id | region |
|---|---|
| 1 | US |
| 2 | EU |
| 3 | US |
| 4 | APAC |
| 5 | EU |
| 6 | LATAM |
| 7 | US |
In the diagram, 2 tasks read the input and 2 receive the moved rows. Follow order 7: it starts in the second input partition and ends up next to orders 1 and 3, which started in the first. Follow order 6: LATAM appears once, yet its data still travels, because the task that read it could not know it was alone. A wide transformation moves data for every key, even one with no partner, though for an aggregation like this each task first folds a key's rows into a single partial list and sends that.
The wide calls you will meet first
Each of these needs rows gathered by a key
- groupBy(...).agg(...): every row carrying the same key has to reach a single task before the aggregate is final
- distinct(): a row is a duplicate only if its twin, anywhere in the table, has been brought next to it
- dropDuplicates(["user_id"]): the same gathering, keyed on the columns you name
- orderBy / sort: the first row of the output could be sitting in any input partition
- join between 2 large tables: matching keys from both sides have to meet in one task (the Joins lessons cover the case where one side is small)
Read the argument of the call to find the key. groupBy("region") gathers by region, dropDuplicates(["user_id"]) gathers by user, distinct() gathers by the whole row, and orderBy("profit") gathers by ranges of profit so that one task holds the smallest values and another the largest. Each of these has the same shape underneath, an answer that depends on every row sharing a key, and that shared need is what makes a call wide.
- Narrow
- Output partition i comes from input partition i
- A task finishes when its own rows are done
- Partition count unchanged, or summed by union
- No disk write, no network traffic
- Wide
- Output partition i gathers rows from every input partition
- A task cannot start until all input tasks finish
- Partition count set by a shuffle setting
- Rows written to disk and sent across the network
Which of these programs moves rows between executors when its action runs?
- orders.filter(F.col("profit") > 0).select("order_id", "region").collect()
- orders.withColumn("big", F.col("profit") > 100).filter(F.col("big")).collect()
- orders.select("status").distinct().collect()
- orders.union(returns).select("order_id").collect()
Run the aggregation below. Spark has to bring every row for a region into one task before it can print a line for that region, so the rows move between the filter and the count. The filter itself runs where the rows were read, in the same task, before anything travels.
About this lesson
This section is part of the Shuffles: Beginner lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.