BeginnerSpark · 13 min · 3 challenges

Shuffles: Beginner

Cutting a Spark job from 2 hours to 20 minutes almost never comes from cleverer logic. It comes from moving less data across the network. Every Spark operation falls into one of two categories, and the performance gap between them is large: one category stays cheap no matter how much data you have, and the other can run your whole runtime. Those two categories are narrow and wide, and learning to tell them apart is where this lesson starts.

Narrow Transformations

Daily Life

Interviews

A narrow transformation is one where each output partition is built from exactly one input partition. Think of filter: to decide which rows of a partition to keep, an executor only needs the rows already sitting in front of it. It never has to look at any other partition, on any other machine. The same is true of select, which picks columns, and withColumn, which computes a new one, and map, which transforms each row. Every one of these works on a partition in place, using only what is already there.

Because no information has to cross between partitions, narrow operations need no coordination and no network. Each executor runs the operation on its own partitions independently, in parallel, and nothing waits on anything else. This is the cheap, embarrassingly-parallel half of Spark, and it is where you want as much of your work as possible to live. A chain of narrow operations is the closest Spark gets to free.

Another property worth naming is that narrow operations pipeline. When you chain filter, then withColumn, then select on a partition, Spark does not materialise the intermediate results and hand them off; it fuses the chain into one pass over the rows of that partition, applying all 3 transformations to each row before moving to the next. The whole narrow chain becomes a single tight loop per partition with no buffering between steps. This is why adding another narrow operation to a chain is so nearly free: it is one more cheap function in a loop that was already running, not a new pass over the data and certainly not a new round of communication.

filterselect / withColumnmap

filter

Keep some rows

Each executor decides which of its own rows to keep. No other partition is consulted.

select / withColumn

Reshape each row

Pick or compute columns using only the row in hand. Nothing crosses partitions.

map

Transform each row

One row in, one row out, on the partition where it already lives. Pure local work.

With narrow operations, the work travels to the data. The data is already spread across the cluster, and a narrow operation applies a small function to it wherever it sits. Nothing moves. This is the ideal distributed computing aims for, and a job built mostly of narrow operations scales well: double the data and the executors, and the job takes about the same time, because each executor is still chewing through its own local share.

Make it concrete with the orders table. Suppose it is spread across 8 partitions on 4 machines, two partitions per machine, and you run orders.filter(col("status") == "Completed"). Each of the 8 partitions is scanned where it already sits; an executor reads its rows, keeps the completed ones, drops the rest, and emits a smaller partition in the same place. No executor ever asks another what its status counts look like, because the keep-or-drop decision for a row depends only on that row. If you then chain withColumn to add a margin column computed from price and cost, that work also lands on the same partition, on the same machine, with no further movement. 8 independent streams of work, zero conversation between them.

If you know SQL, the analogy is exact and worth holding onto. A WHERE clause and a scalar SELECT expression are narrow: a database can evaluate them row by row without consulting any other row in the table. price > 50 and price * quantity AS revenue both look at a single row and produce an answer. The operations that force a database to build a hash table or sort, GROUP BY and JOIN and DISTINCT, are the ones that cannot be answered one row at a time, and those are the wide operations we turn to next. Narrow Spark transformations are the distributed cousins of the row-at-a-time parts of a query plan.

A purely narrow chain

The demo below is all narrow work: a filter and a column selection, with no operation that needs to look across partitions. Run it and recognise that every executor could do this on its own slice with zero communication. This is the shape of a job that costs almost nothing beyond reading the data.

(products
   .filter(F.col("price") > 50)
   .select("product_name", "category", "price")
   .orderBy("product_name"))

One honest caveat keeps you from oversimplifying: the orderBy at the end of that demo is actually a wide operation, because a global sort has to compare rows across partitions. We left it in because it is how you would really write the query, but the filter and the select are the narrow core, and those are the parts that cost nothing. The lesson for now is to see the narrow operations clearly, and the next sections will teach you to spot the orderBy as the one line in that chain that does move data. A chain is rarely all narrow or all wide; read it operation by operation.

The filter and select in that chain are narrow to the core: each executor handled its own partitions with no awareness of the others. Recognise that category on sight, because it does not cost. Now we turn to the other category, the one that does.

Wide Transformations

Daily Life

Interviews

A wide transformation is one where an output partition needs data from many input partitions. The classic example is groupBy. To sum profit per region, every row for a given region has to end up in the same place so it can be added together, but those rows are scattered across every partition on every machine. Spark has to gather all the rows for each region together first, and that gathering pulls data from everywhere.

Join is the same story. To match orders to products on a product id, every order and every product sharing that id must meet on the same machine, and since they started life spread across the cluster, they have to be brought together by key. distinct must compare rows from all partitions to find duplicates. Each of these operations has the same defining feature: the output cannot be computed partition by partition in isolation, because the answer for one output piece depends on input from many places.

Take the groupBy by region in physical terms. The orders table is scattered across the cluster with no regard for region; partition 3 on machine A holds a mix of EU, APAC, and US rows, and so does every other partition. To compute one number per region, Spark cannot let each partition report a local total and stop there, because no single partition holds all of any region's rows. It has to physically relocate rows so that every EU row, wherever it started, ends up on the one machine that will sum EU. That relocation is the expense. The aggregation arithmetic itself, adding up profits, is trivial; the cost is the moving.

It helps to count the dependency. A narrow operation's output partition has exactly one parent: the input partition it was built from. A wide operation's output partition has many parents, potentially every input partition in the job, because the rows it needs could have started anywhere. That fan-in is the formal definition, and it is what forces the data movement. When you read that groupBy depends on all input partitions, you are reading the reason it cannot be free: there is no way to assemble the EU partition without reaching into every input partition to collect the EU rows hiding in each one.

•Narrow (output from ONE input)

filter, select, withColumn, map
Each executor works alone
No network, no coordination
Scales almost for free

⚠Wide (output from MANY inputs)

groupBy, join, distinct, orderBy
Rows must be regrouped by key
Data crosses the network
Where the cost concentrates

Distinct is the wide operation people most often forget is wide, so it deserves a closer look. To return the distinct set of customer regions from orders, Spark cannot just dedupe within each partition and stop, because the same region almost certainly appears in many partitions on many machines. It has to bring all the candidate values for a given region together to confirm there is only one, and that gathering is a shuffle. The same is true of dropDuplicates on a key and of a count of distinct values. Any time the question is is this value unique across the whole dataset, the answer requires comparing across partitions, and comparing across partitions means moving data.

Wide matters because this gathering is not a metaphor. It is a physical movement of data across the network between machines, and the rest of this lesson is about how that movement works and what it costs. For now, learn to recognise it: when you see groupBy, join, distinct, or a global orderBy, hear an alarm that says data is about to move. Those are the operations that turn a cheap job into an expensive one.

Watch a wide op regroup the data

The demo below has a groupBy, the canonical wide operation. To produce the count per region, Spark must bring every row for each region together, which means moving data across the cluster. Run it, and as you do, contrast it with the purely narrow chain from the last section: that one moved nothing; this one cannot avoid it.

(orders
   .groupBy("region")
   .agg(F.count(F.lit(1)).alias("order_count"))
   .orderBy(F.col("order_count").desc()))

Everything before that groupBy can be cheap and local; the groupBy itself forces the expensive regrouping. Spotting wide operations on sight is the most useful skill for reasoning about Spark performance, because they are where almost all of a job's time and money go.

A word on orderBy, because it surprises people. A global sort is wide for the same reason a groupBy is: to produce a single ordered result, the rows have to be compared across every partition, which means redistributing them by their sort key so that, say, all the smallest values land on the first reduce task and the largest on the last. orderBy(col("order_count").desc()) in the demo above is doing that under the hood. So the demo actually contains two wide operations, the groupBy and the final sort, though on a handful of region rows the sort is negligible. Wide is about whether output depends on many inputs, not about how the operation is spelled, and a sort qualifies.

What a Shuffle Actually Is

Daily Life

Interviews

The movement that a wide operation forces has a name, and you will hear it constantly: the shuffle. A shuffle is the physical redistribution of data across the cluster so that rows which need to be together end up together. It is the most expensive thing Spark does, and it follows directly from a wide transformation. Every wide operation triggers a shuffle; every shuffle is triggered by a wide operation. The two are the same event seen from two angles, the logical operation and its physical cost.

Here is what has to happen for a groupBy by region. Each executor looks at its local rows and sorts them into buckets, one bucket per region. Then every executor sends its region buckets to whichever executor is responsible for collecting that region. Rows fly across the network in every direction at once, each heading to the machine that will aggregate its key. Once the dust settles, every region's rows sit together on one machine, and the aggregation can finally run. That cross-cluster all-to-all movement is the shuffle.

all-to-all

the shape of a shuffle

There is a useful piece of vocabulary that comes from the older MapReduce world and still describes a shuffle exactly: the map side and the reduce side. The map side is the set of executors holding the input, the ones that bucket their local rows by key. The reduce side is the set of executors that collect a given key and finish the work, one per output group. A shuffle is the data crossing from the map side to the reduce side. You will see those words throughout the intermediate tier and in the Spark UI, and they always mean the same two ends of the same movement: where the data starts and where it has to end up.

The word shuffle is apt: the data is genuinely shuffled, reorganised from an arrangement based on how it happened to be read into an arrangement based on the key you are grouping or joining by. Nothing about the data changes; only where each row physically lives changes. But that change of location is enormously expensive, because it means writing data out, sending it over the network, and reading it back in, for potentially the entire dataset.

TIP

When an interviewer asks what a shuffle is, the crisp answer names the cause and the cost in one breath: "It is the data movement a wide operation forces, so rows with the same key land together, and it is expensive because of the disk write and network transfer." Then stop. That is a complete answer that proves you understand the mechanism.

Be concrete about how much can move. If the orders table is a terabyte and you group it by region, the shuffle does not move a summarised terabyte; before the aggregation runs, every one of those rows has to travel to the machine that owns its region, so close to a full terabyte crosses the network and gets written to disk on the way. The aggregation shrinks the data, but the shrinking happens after the shuffle, not before it. This is why the size of the input to a wide operation, not the size of its output, is what sets the bill. A groupBy that returns 4 region rows can still be one of the most expensive things in your job if a terabyte had to move to produce those 4 rows.

Hold onto this equivalence as you read the rest of the lesson: wide operation, shuffle, network movement, expense. They are all the same thing. When you optimise a Spark job, you are almost always trying to do one of two things: avoid a shuffle entirely, or make the shuffle you cannot avoid move less data.

Why Wide Costs and Narrow Does Not

Daily Life

Interviews

Being precise about why the two categories differ so dramatically in cost is what lets you predict performance instead of memorising rules. A narrow operation reads a partition that is already in memory on an executor, applies a function, and produces output in memory on the same executor. The data never leaves the machine. The cost is just the CPU work of the function, which is usually trivial compared to anything involving disk or network.

A wide operation, by contrast, pays 3 expensive costs stacked on top of each other. First, the sending side writes its sorted buckets to local disk, because the data has to be staged before it can be sent. Disk is far slower than memory. Second, the data travels across the network to the receiving executors, and network bandwidth is a shared, limited resource. Third, both ends pay to serialise and deserialise the data into and out of its wire format. Disk, plus network, plus serialisation, all for potentially the whole dataset, is why one wide operation can cost more than everything else in a job combined.

	Narrow operation	Wide operation (shuffle)
Data movement	None; stays on the executor	Across the network, all-to-all
Disk	None	Write buckets out, read them back
Coordination	None; fully independent	Every executor waits for the exchange
Cost	CPU only, near free	Disk + network + serialisation, dominant

To feel the gap in real numbers, hold a rough hierarchy of speeds in your head. Reading data already in an executor's memory runs at many gigabytes per second. Writing to a local disk is perhaps an order of magnitude slower, and a network link between machines is often slower still and shared among every task using it at that moment. A narrow operation pays only the first, fastest cost. A wide operation pays the slow disk write, then the shared network, then the disk read on the far side, then the CPU to serialise and deserialise every row in between. That is not a small constant factor; for the same volume of data it separates work measured in seconds from work measured in tens of minutes.

There is a subtler cost to a shuffle too: it is a synchronisation barrier. The stage after the shuffle cannot start until the shuffle is complete, because it needs all the regrouped data to be in place. So even the fastest executors sit idle waiting for the slowest one to finish sending and receiving. Beyond the data movement, a shuffle costs you the parallelism, because it forces everyone to wait at the boundary. This is why one slow task in a shuffle, a single oversized partition, can stall the entire job.

This barrier is also why a shuffle splits a Spark job into stages. A stage is a run of narrow operations that can be pipelined together on each partition without stopping; the moment a wide operation appears, Spark has to end the current stage, perform the shuffle, and begin a new stage on the regrouped data. So when you see a job described as having 3 stages, you are really seeing that it has two shuffles, one at each stage boundary. Counting the wide operations in your code tells you how many stages and how many shuffles you will get before you run anything, and that count predicts how expensive the job will be better than anything else.

Put it all together and the rule for optimising is plain. Narrow work is the cheap part you can pile on freely. Wide work is the expensive part you spend deliberately. A well-tuned Spark job is one that does as much narrow work as possible and as little wide work as it can get away with, and where the wide work it does do is shaped to move the least data. Everything in the intermediate and advanced tiers is a refinement of that one idea.

Spotting Shuffles in Your Code

Daily Life

Interviews

The practical skill this lesson builds is reading your own code and seeing, before you run it, where the shuffles are. It rests on a small vocabulary of operations that signal a wide transformation. When you see any of these in a chain, a shuffle is coming, and that is where the cost will be. Everything else is narrow and cheap.

groupBy / agg by key

Rows for each key must gather together. A shuffle.

join

Matching keys must meet on one machine. A shuffle, unless one side is broadcast.

distinct / dropDuplicates

Rows from all partitions must be compared. A shuffle.

orderBy / sort (global)

A total ordering needs to compare across partitions. A shuffle.

repartition

You explicitly asked Spark to reshuffle into new partitions. A shuffle by definition.

Run your eye down a chain and tag each line. filter, select, withColumn: narrow, free. Then a groupBy: there is your shuffle, there is the cost. The exercise sounds trivial, and it is what separates someone who can reason about Spark performance from someone who cannot. Most slow jobs are slow at one or two specific wide operations, and finding them starts with reading the code and naming them.

Take a realistic chain over the seed tables and tag it line by line. You read orders, filter to one country, join to a small customers table on customer_id, group by region, and sort the result. Reading down: the read is neither here nor there, the filter is narrow, the join is wide because matching customer_ids must meet, the groupBy is wide because regions must gather, and the final orderBy is wide because the result must be totally ordered. That is 3 shuffles in one short chain. Now you know, before running it, that the join, the grouping, and the sort are where the time will go, and that the filter is your one free lever to make all 3 cheaper by shrinking the data first.

Be careful about the one operation that looks wide but is not: a non-global withColumn or select that references a window without an order, or a simple limit on an already-sorted stream, can be narrow even though it feels like it should reach across data. The reliable test is always the same question, not the operation's name: to produce one output partition, does Spark need rows from more than one input partition? If yes, it is wide and it shuffles. If each output partition can be built from a single input partition, it is narrow and free. When in doubt, ask whether the answer for one row depends on other rows that might live elsewhere.

Tag the wide operation

The fill-in below builds a job with one wide operation hidden among narrow ones. Complete the operations, and as you do, notice which one is the shuffle: the grouping that brings every category's rows together. The filter before it is narrow and free; the grouping is where the data moves.

> From products, for in-stock items only (in_stock = 1), return the number of products in each category, highest count first. The filter is narrow; the grouping is the one wide operation that shuffles.

(products
   .filter(F.col("in_stock") == 1)
   .___("category")
   .agg(F.___(F.lit(1)).alias("n"))
   .orderBy(F.col("n").desc(), F.col("category").asc()))

groupBy

count

filter

select

That groupBy is the shuffle, and now you can find it on sight. Spark performance tuning starts here: read the code, name the wide operations, and you have found where the time goes before you have run a thing. The intermediate tier takes you inside that shuffle to see what it does, and the advanced tier shows you how to make shuffles cheaper or remove them entirely.

✓Do

Tag each operation narrow or wide as you read a chain; the wide ones are the cost.
Pile on narrow work freely; it is per-partition and nearly free.
Treat every groupBy, join, distinct, and global sort as a deliberate shuffle.
Filter early so the wide operation that follows shuffles less data.

✗Don't

Don't assume all operations cost the same; narrow is free, wide is dominant.
Don't add a wide operation casually; each one writes to disk, hits the network, and serialises.
Don't forget a global orderBy is a shuffle; it needs a total order across partitions.
Don't ignore that a shuffle is a barrier; the next stage waits for the slowest task.

❯❯❯PUTTING IT ALL TOGETHER

> You are handed a Spark job that reads a large events table, filters it to one country, joins it to a small lookup of content titles, and counts views per title. It is slow, and you have not run it yet.

You read the chain and tag each operation: the filter and the column work are narrow and free.

The join and the groupBy are the two wide operations, so those are the two shuffles where the cost lives.

You note the filter to one country should run before the join, so the wide operations move less data.

Without running anything, you can already say the job's time will go to the join and the count, and that is where you would optimise.

KEY TAKEAWAYS

A narrow transformation builds each output partition from one input partition and moves no data.

A wide transformation (groupBy, join, distinct, global sort) needs many input partitions and shuffles.

A shuffle is the all-to-all network movement a wide operation forces so keys land together.

A shuffle costs disk, network, and serialisation stacked, plus it is a barrier that costs parallelism.

Optimising Spark is mostly avoiding shuffles or making the unavoidable ones move less data.

One category is free. The other can run your whole bill.

Category: Spark
Difficulty: beginner
Duration: 13 minutes
Challenges: 3 hands-on challenges

Topics covered: Narrow Transformations, Wide Transformations, What a Shuffle Actually Is, Why Wide Costs and Narrow Does Not, Spotting Shuffles in Your Code

Lesson Sections

Narrow Transformations (concepts: paSparkExecutionModel)
A narrow transformation is one where each output partition is built from exactly one input partition. Think of filter: to decide which rows of a partition to keep, an executor only needs the rows already sitting in front of it. It never has to look at any other partition, on any other machine. The same is true of select, which picks columns, and withColumn, which computes a new one, and map, which transforms each row. Every one of these works on a partition in place, using only what is already t
Wide Transformations (concepts: paSparkExecutionModel)
A wide transformation is one where an output partition needs data from many input partitions. The classic example is groupBy. To sum profit per region, every row for a given region has to end up in the same place so it can be added together, but those rows are scattered across every partition on every machine. Spark has to gather all the rows for each region together first, and that gathering pulls data from everywhere. Join is the same story. To match orders to products on a product id, every o
What a Shuffle Actually Is (concepts: paShuffleOptimization)
The movement that a wide operation forces has a name, and you will hear it constantly: the shuffle. A shuffle is the physical redistribution of data across the cluster so that rows which need to be together end up together. It is the most expensive thing Spark does, and it follows directly from a wide transformation. Every wide operation triggers a shuffle; every shuffle is triggered by a wide operation. The two are the same event seen from two angles, the logical operation and its physical cost
Why Wide Costs and Narrow Does Not (concepts: paShuffleOptimization)
Being precise about why the two categories differ so dramatically in cost is what lets you predict performance instead of memorising rules. A narrow operation reads a partition that is already in memory on an executor, applies a function, and produces output in memory on the same executor. The data never leaves the machine. The cost is just the CPU work of the function, which is usually trivial compared to anything involving disk or network. A wide operation, by contrast, pays 3 expensive costs
Spotting Shuffles in Your Code (concepts: paShuffleOptimization)
The practical skill this lesson builds is reading your own code and seeing, before you run it, where the shuffles are. It rests on a small vocabulary of operations that signal a wide transformation. When you see any of these in a chain, a shuffle is coming, and that is where the cost will be. Everything else is narrow and cheap. Run your eye down a chain and tag each line. filter, select, withColumn: narrow, free. Then a groupBy: there is your shuffle, there is the cost. The exercise sounds triv