What is the difference between narrow and wide transformations?

Summary

• A narrow transformation computes each output partition from a small number of parent partitions, so Spark can run it in a pipeline inside one stage. A wide transformation needs data from many parent partitions, which requires a shuffle.
• The RDD programming guide defines the shuffle as "Spark's mechanism for re-distributing data so that it's grouped differently across partitions" and calls it "a complex and costly operation". Stages are separated by shuffle operations.
• map, filter, flatMap and mapPartitions are narrow. groupByKey, reduceByKey, cogroup and repartition can cause a shuffle. join is wide unless both inputs share a partitioner, and coalesce reduces partitions without a shuffle by default.
• In DataFrame plans, an Exchange node marks a data shuffle. Joins and aggregations shuffle data, and spark.sql.shuffle.partitions (default 200) sets how many partitions they use.

What is the difference between narrow and wide transformations?

The terms come from the original RDD research paper. It defines narrow dependencies as ones "where each partition of the parent RDD is used by at most one partition of the child RDD", and wide dependencies as ones where multiple child partitions may depend on a parent partition. Spark's API docs describe a narrow dependency as one where "each partition of the child RDD depends on a small number of partitions of the parent RDD", and add that "Narrow dependencies allow for pipelined execution." In practice, a narrow transformation needs no shuffle and a wide transformation does. The paper gives two reasons the split matters: narrow steps pipeline on one node, and after a node failure only the lost parent partitions need to be recomputed.

Which transformations are narrow and which are wide?

Transformation	Dependency	Basis
map, filter, flatMap, mapPartitions	Narrow	Each works on one partition at a time; the paper lists map and filter as narrow
union	Narrow	Each child partition is computed through a narrow dependency on the corresponding parent
coalesce to fewer partitions	Narrow by default	The DataFrame docs say it "results in a narrow dependency"; RDD.coalesce has a shuffle parameter that defaults to False
groupByKey, reduceByKey, aggregateByKey, cogroup	Wide	The RDD guide lists 'ByKey operations (except for counting) and join operations like cogroup and join among operations that can cause a shuffle
join	Wide, or narrow when both inputs use the same partitioner	The paper: a join may give "two narrow dependencies (if they are both hash/range partitioned with the same partitioner)"
repartition	Wide	The docs: "Internally, this uses a shuffle to redistribute data"

What happens during a shuffle?

Take reduceByKey. Values for one key can sit on different partitions or machines, so "to organize all the data for a single reduceByKey reduce task to execute, Spark needs to perform an all-to-all operation." Spark generates map tasks to organize the data and reduce tasks to aggregate it. Map task results stay in memory until they can't fit, are sorted by target partition and written to a single file, and reduce tasks then read the relevant sorted blocks. The guide calls the shuffle expensive "since it involves disk I/O, data serialization, and network I/O", and notes that Spark spills to disk when the data does not fit in memory.

How do you keep shuffles small?

rdd = sc.parallelize([1, 2, 3, 4])
narrow = rdd.map(lambda x: x * 2).filter(lambda x: x > 3)  # no shuffle

pairs = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])
counts = pairs.reduceByKey(lambda a, b: a + b)  # shuffles by key

• Aggregate with reduceByKey or aggregateByKey, not groupByKey. reduceByKey "will also perform the merging locally on each mapper before sending results to a reducer", and the groupByKey docs say the other two "will provide much better performance" for aggregations.
• Reduce partitions with coalesce, not repartition. The repartition docs suggest coalesce when decreasing partitions, because it can avoid a shuffle. A drastic coalesce, such as to one partition, may leave the computation on fewer nodes than you want.
• Read the plan. df.groupBy("customer_id").count().explain() prints the physical plan by default. In the SQL tuning guide's plans, Exchange lines are labeled // DATA SHUFFLE.
• Let adaptive query execution help. AQE is on by default and can coalesce shuffle partitions and convert a sort-merge join into a broadcast hash join at runtime.

How does Databricks approach shuffles?

The Databricks docs describe AQE as enabled by default, with the ability to change a sort merge join into a broadcast hash join and to coalesce partitions after a shuffle exchange. Setting spark.sql.shuffle.partitions to auto enables auto-optimized shuffle, which "automatically determines this number based on the query plan and the query input data size", and the Spark UI guide recommends it when a stage shows high shuffle. The RDD examples above need a SparkContext: serverless compute supports only Spark Connect APIs and does not support Spark RDD APIs, so use DataFrames there.

FAQs

Do DataFrames have narrow and wide transformations?

The terms describe RDD dependencies, but the split shows up in DataFrame plans. The Spark SQL docs describe spark.sql.shuffle.partitions as the number of partitions "when shuffling data for joins or aggregations", and Databricks notes that exchanges appear "usually when there's a join, aggregate, or window".

Does a join always shuffle?

No. For RDDs, a join of two inputs with the same partitioner can use narrow dependencies. In Spark SQL, Storage Partition Join uses the existing storage layout to avoid the shuffle phase for compatible V2 data sources, and a BROADCAST hint asks Spark to prioritize a broadcast join.

Is coalesce better than repartition?

They do different jobs. coalesce reduces partitions without a shuffle, and if you request more partitions it stays at the current number. repartition can increase or decrease parallelism and uses a shuffle to redistribute data.