How Many Pieces

Four hundred Daily Grind shops, one year of tills, and the question of how many pieces to cut it into. Too few and one executor works while ninety-nine wait; too many and Spark spends its day scheduling thousand-row tasks. The number of data partitions is the number of tasks in a stage (Who Does What on a Cluster), and this lesson is the two methods that change it, and the difference between them.

Watch it happen

The eleven orders again. spark_partition_id() tells each row which data partition it sits in, so counting the distinct values counts the pieces:

def pieces(df):
    return df.select(F.spark_partition_id().alias("partition")).distinct()

pieces(orders).count()                              # 1
pieces(orders.repartition(4)).count()               # 4
pieces(orders.repartition(4).coalesce(2)).count()   # 2

And where the rows went:

result = (orders.repartition(4)
                .withColumn("partition", F.spark_partition_id())
                .groupBy("partition").count())

result = (orders.repartition(4).coalesce(2)
                .withColumn("partition", F.spark_partition_id())
                .groupBy("partition").count())

repartition shuffles; coalesce merges

The plans say which is which:

orders.repartition(4).explain()
orders.repartition(4).coalesce(2).explain()

ExplicitRepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
  DataSourceExec: partitions=1

CoalesceExec: partitions=2
  ExplicitRepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
    DataSourceExec: partitions=1

repartition(n) deals the rows out again, round-robin, into n pieces of about equal size. It is a shuffle: every row moves, and the count can go up or down. repartition("size") hashes a column instead, so every row with the same key lands in the same piece, which is what a groupBy does for you before it aggregates.

coalesce(n) does not deal rows out. Spark's docs: "this operation results in a narrow dependency, e.g. if you go from 1000 partitions to 100 partitions, there will not be a shuffle, instead each of the 100 new partitions will claim 10 of the current partitions." Neighbouring pieces merge, and the count can only go down. The docs carry the trap in the next sentence: "if you're doing a drastic coalesce, e.g. to numPartitions = 1, this may result in your computation taking place on fewer nodes than you like." coalesce(1) to write one file pulls the whole stage above it onto one task as well; repartition(1) costs a shuffle and keeps the work before it parallel.

Where the 200 comes from

After a shuffle, Spark uses spark.sql.shuffle.partitions pieces, default 200. On a laptop-sized table that is 200 tasks of nearly nothing; on a cluster with ten billion rows it may be far too few. Since Spark 3.2, adaptive query execution is on by default and "coalesces the post shuffle partitions based on the map output statistics", which fixes the too-many case at run time. Sail reports the same two defaults if you ask spark.conf.get, and plans Hash([size], 2) on this site's shuffles: the number is Spark's, and the behaviour you can measure here is one process's.

How many is right

The tuning guide's rule of thumb: "In general, we recommend 2-3 tasks per CPU core in your cluster." Pieces of a few hundred megabytes; tasks of seconds, not milliseconds and not hours.

The symptom, and the fix

Two opposite symptoms. The Stages tab shows thousands of tasks that each took a few milliseconds: too many pieces. coalesce down, or let adaptive query execution do it. One stage shows a handful of tasks each taking an hour while the cluster idles: too few. repartition up before the expensive step. Count the tasks, compare with the cores, aim for two or three each.

Self-check

You have 1,000 partitions and want 100. coalesce(100) or repartition(100)?

coalesce: no shuffle, each new piece claims ten old ones. repartition only if the thousand are badly uneven and you want them dealt out again, which is the next lesson's problem.

Nothing to grade here. Next, When One Key Has Everyone: what happens when the pieces are the right number and one of them holds nine-tenths of the rows.