Partitions and Parallelism

Concepts covered: Distributed Primitives

A table never arrives at an executor whole. When Spark reads a dataset it cuts the rows into partitions, each a slice that 1 task reads and processes, and the default target for a partition read from files is 128 MB, set by spark.sql.files.maxPartitionBytes. One task processes exactly 1 partition, never half of one and never 2, so the partition count of a dataset is also the task count of the work that reads it.

• 8,192

The animation below runs the smallest version of this. The order_items table holds 8 rows and is read as 4 partitions, so each partition carries 2 rows. The program adds a line_total column, quantity times unit price, and collects the result.

(order_items
   .withColumn("line_total", F.col("quantity") * F.col("unit_price"))
   .collect())

Four tasks ran, 1 per partition, and all 4 ran at the same moment. Each task computed its 2 line totals from the rows it already held: the first partition produced 30 and 40 for order 1, the second produced 24 for order 2 and 120 for order 3, the third 20 and 50 for orders 4 and 5, and the fourth 60 and 15 for orders 6 and 7. No row left its partition, because a line total needs nothing from any other row, and a transformation with that property is called narrow. At the end the driver collected 8 rows, the 1 moment data crossed from executors to the driver. Multiply the table by 1 million and the job keeps its shape: 8 million rows in 4 partitions still run as 4 tasks, each a million times longer.

The ceiling on parallelism

Since each task is 1 partition, the partition count is a hard ceiling on how many tasks can run at once. A dataset in 4 partitions runs as 4 tasks, and a cluster with 200 cores assigns 4 of them and leaves 196 idle for as long as that work lasts. The opposite case also fails. A dataset in 500,000 partitions of 2 KB each produces 500,000 tasks, and every task costs the driver a few milliseconds to schedule and record, so the driver spends more time on bookkeeping than the executors spend on rows. A SQL engine makes this decision internally. Spark exposes it, which is why partition count is the first thing to check when a job is slow and the cluster is quiet.

Where the count comes from

The count on read comes from the files. A folder of 80 Parquet files at 128 MB each reads as 80 partitions, and a 1 GB Parquet file reads as 8, because the reader splits a file at 128 MB boundaries. A single 10 GB gzip file reads as 1 partition, because gzip cannot be split mid-stream, and that 1 task runs on 1 core for the several minutes it takes to decompress 10 GB, while every other core in the cluster waits. Snappy-compressed Parquet splits cleanly, which is one reason it is Spark's default output format. The count can be changed after the read with repartition and coalesce, which a later lesson covers, and a grouping changes it again, as the last section shows.

The cell below runs a narrow chain on the practice seed's full order_items table: it computes each line's total, keeps the lines with a quantity of 2 or more, and selects the columns worth reading. Every step runs inside the task that holds the row, as the animation showed, with no task waiting on another.

Every row in that result was produced by the task holding its partition, and no task coordinated with another. What the cell hides is when those 3 steps ran, and the next section shows that none of them ran at the line where it was written.

About this lesson

This section is part of the Execution Model: 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.