Partitioning: Beginner

When you call spark.read on a directory, Spark lists the files and cuts them into partitions of one split size. spark.sql.files.maxPartitionBytes caps the split at 128MB by default. The split shrinks when a read is small so that every core still gets work: Spark divides the total bytes, plus a 4MB charge per file, by the core count and takes that when it's smaller, with 4MB as the floor. Any read above 128MB times the core count splits at 128MB: a 10GB file on 16 cores becomes 80 partitions, read by 80 tasks in parallel. Files larger than the split are cut at its boundaries, and smaller files are packed together up to the split.

The program above reads a 12-row orders table as 4 partitions and filters it down to the 2 refunded rows. Follow the task count: 4 tasks launch, each scans its 3 rows, and 4 result sets travel back to the driver even though at least 2 of them carry nothing. The filter removed 83% of the rows and left the partition count alone.

A job drowns in small partitions when the count stays high while the bytes per partition collapse, and you have already met the usual sources. A selective filter keeps the partition count: 80 partitions of 128MB that keep 0.1% of their rows become 80 partitions of around 128KB, and every narrow step after it runs as 80 tasks over 10MB of data. A shuffle writes spark.sql.shuffle.partitions partitions, 200 by default, whatever the data: a groupBy over 50MB writes 200 shuffle partitions of roughly 250KB each. Adaptive query execution, on by default since Spark 3.2 and taught later, merges small shuffle partitions at run time, so all 200 run only when it is off. And a directory of thousands of tiny files produces thousands of tiny partitions, which the small-files section prices in detail.

The runnable example below groups the practice orders table by status and counts each group. Run it: the shuffle writes 200 partitions for a handful of statuses, and adaptive execution merges the near-empty ones, so on a cluster the reduce stage runs as around 1 task. With it off, 200 tasks would produce those few rows. The isNotNull filter keeps a null status from forming its own group.

status_counts = (
    orders.filter(F.col("status").isNotNull())
    .groupBy("status")
    .count()
)
status_counts.show()

Unsplittable files are the classic source. A CSV compressed with gzip cannot be cut at an arbitrary byte, because decompression has to start from the beginning of the stream, so a single 10GB .csv.gz arrives as 1 partition and 1 task however many cores you have. A read from a relational database through JDBC is 1 partition unless you tell Spark how to split the query, so a 200 million row table arrives through a single connection into a single task. And a shuffle partition setting that someone lowered to 4 for a small table keeps applying to every groupBy and join in that session, including the one over the big table.

The open cost also sets the partition count. Spark packs small files into partitions, but it charges each file a fixed spark.sql.files.openCostInBytes of 4MB on top of its real size when it fills a 128MB partition, precisely because opening is expensive. A 1MB file counts as 5MB, so a partition holds about 25 of them and about 25MB of real data. The 100,000 files become roughly 4,000 partitions and as many tasks, each reading about 25MB through around 25 opens, where the 1,000 large files become 1,000 tasks reading 100MB through 1 open each.

With inferSchema set, Spark parses every row of every file and, for each column, tracks the narrowest type that still fits every value seen so far: integer until a decimal point appears, double until a letter appears, string after that. That's a complete scan of the input, and it runs when .csv(path) is called, before you have asked for anything. You'll see a job appear in the UI with no action in your code. When you do call an action, the real job reads the same bytes a second time. On a 500GB directory you have read 1TB to get an answer that needed 500GB.

The schema string names each column and its type in the same form you would use in a CREATE TABLE. With it in place, the read opens nothing until an action runs, and then the parser converts each value straight into the declared type. The same schema can be built as a StructType of StructField objects, which you reach for when a field is nested. Either way the cost of learning the types drops to 0 bytes, and the types are the ones you chose rather than the ones a scan happened to find.

The extra scan would be tolerable if it always produced the right schema. It produces the schema that fits the data on that day. A postal code column holding 02134 infers as an integer and comes back as 2134, with the leading zero gone for good. An amount column that is numeric in a billion rows and holds the text N/A in 1 row infers as string, and the arithmetic you write downstream runs on a string column: depending on your settings it either fails at the action, after the job has run for an hour, or quietly turns the bad values into null. A column that is empty in every row infers as string, and the day it fills with numbers the schema changes under every job that reads it. A declared schema keeps the types you chose: a value that doesn't parse becomes null by default, and .option("mode", "FAILFAST") makes the read fail at the first malformed row.

JSON reads pay the same pass. spark.read.json(path) infers by default, so a JSON read without .schema(...) always pays the full pass. Both readers accept a samplingRatio option that infers from a fraction of the rows, which shortens the scan and makes the guess less reliable at the same time; a declared schema avoids both problems.