Narrow Transformations

Concepts covered: Spark Execution Model

A narrow transformation computes every output row from rows that sit in a single input partition. The task holding partition 2 reads its rows and applies the call, then writes partition 2 of the result; no row arrives from any other task and none leaves. This is how filter and select work, and withColumn and drop as well. union is narrow too, along with every column function you wrap inside any of them.

One partition in, one partition out

Take an orders table of 8 rows that Spark reads as 4 partitions of 2 rows each, and a job that keeps rows where status equals Completed. The task on the first partition sees orders 101 and 102 and keeps 101, which finishes its work. The task on the second sees 103 and 104 and keeps 103. Each task decides the fate of its rows by reading a column value already in front of it, so there is nothing to request from another executor and nothing to wait for.

(orders
   .filter(F.col("status") == "Completed")
   .collect())

Each of the 4 tasks reads its own 2 rows and hands the survivors straight to the driver. No arrow crosses from one task to another, and the result still has 4 partitions: 3 of them now hold 1 row each and the fourth holds 2. filter and select can shrink a partition or change its columns, as can withColumn, and none of them moves the partition count or the executor holding each partition. A union places both inputs' partitions side by side, unchanged.

What a narrow task touches

A narrow task reads its partition from storage and applies the call. Then it either passes rows on to the next narrow call or sends its result to the driver. It writes nothing to local disk and sends nothing to another executor. On a cluster with 16 slots and a table of 200 partitions, Spark runs 16 of these tasks at a time until all 200 are done, and a task that finishes early frees its slot for the next partition without waiting on any neighbour.

The partition count after a filter equals the count before it, so keeping 1% of the rows leaves you with the same 200 partitions, most of them nearly empty. The Partitioning lessons deal with what to do about that.

Chaining narrow calls costs a single pass

Because a narrow call needs nothing from outside its partition, Spark runs a chain of them inside the same task. In the program below, each task takes a row, tests status, tests profit, computes profit_x2, drops the columns that select leaves out, and moves to the next row. The row stays in that executor's memory from the first call to the last. The code has 4 transformations, and they still add up to a single pass over the data.

Run the chain and read the 5 rows it prints. Then change the withColumn to compute F.col("profit") / 2 and run again. The values you see change, while the number of rows and the number of partitions stay the same, because every change you made is decided row by row inside the task that read the row.

Spark reads an events table as 200 partitions and runs the program below, then calls collect(). How many partitions does result hold when the action runs?

result = (events
    .filter(F.col("country") == "US")
    .select("user_id", "event_type")
    .withColumn("is_click", F.col("event_type") == "click"))

• 200, the same as the input, because each task filters and rewrites its own partition and no row moves anywhere
• Fewer than 200, because the filter drops rows and Spark merges the partitions that end up empty into their neighbours
• 1, because withColumn has to see every row at once to add the new column consistently across all of them
• 3, because select keeps 2 columns and withColumn adds a third, and Spark holds a partition per column

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.