Spark Partitioning

A partition is the unit of work in Spark: one partition is read by one task, on one core, start to finish. So "how many partitions" is really "how many tasks", and "which rows are in this partition" is really "how much work does that one task have". Three different rules decide the answer, and only one of them is the one you called.

The three rules are: a byte-packing rule when Spark reads files, a round-robin counter when you call repartition(n), and a hash of the key when you call repartition(n, "col") — or when Spark inserts an exchange of its own to satisfy a join or a group-by. That last case is the one that matters, because Spark's own rule overwrites yours.

The panel below runs all three. Pick a key distribution, pick what you wrote in your code, and read the log to see which layout actually reaches the next stage. The control that carries the lesson is the join checkbox.

Start with One hot key, df.repartition(n), and the join unchecked. Every partition is the same height: round-robin repartitioning really does balance the data perfectly. Now tick the join. The bars collapse into one 2.7-million-row spike, and no value of n flattens it again. The layout you built was thrown away before the join ever ran.

Rule 1: round-robin ignores your data entirely

df.repartition(n) with no column produces RoundRobinPartitioning(n). Each input task picks a starting partition at random, then deals its rows out one at a time: row 1 to partition p, row 2 to p+1, wrapping at n. Nothing looks at the row's contents.

That is why it balances perfectly and why it is useless for a join. It is the right tool for exactly two jobs: turning 8 fat input partitions into 200 so your cluster's cores are not idle, and controlling how many output files a write produces. It is the wrong tool whenever the next operator needs rows grouped by key, because grouping by key is precisely what it does not do.

Set the distribution to Uniform, choose df.repartition(n), and drag n. Rows sent over the network stays at 6.00M for every value: a round-robin repartition moves every row across the network, whether n is 4 or 400. That cost is the same whether or not it helped.

Rule 2: the hash, which is where a key becomes indivisible

df.repartition(n, "customer_id") produces HashPartitioning(customer_id, n), and Spark defines the partition id for that in one line of sql/catalyst/.../physical/partitioning.scala:

Pmod(new Murmur3Hash(expressions), Literal(numPartitions))

In words: hash the key columns with MurmurHash3 (a non-cryptographic hash function; Spark uses the 32-bit x86 variant with seed 42), then take that modulo the partition count. Pmod is modulo that always returns a non-negative result, because hashes can be negative and partition ids cannot.

Every consequence people find surprising falls out of that one expression. A key maps to exactly one id, so all 2.7 million rows of customer c0001 land in one partition and one task reads them all. Changing n changes which id, never how many — drag n with One hot key selected and watch the spike jump from bar to bar without ever shrinking.

The hash is also worse at balancing than people expect when the number of distinct keys is close to the number of partitions. Select Uniform, where all 1,200 customers have about 5,000 rows each, and tick the join: with 200 partitions there are only six keys per partition on average, and the luck of the hash gives the biggest partition roughly twice the median. That is not skew in your data — it is skew in the assignment. It disappears once you have thousands of keys per partition, and it is the reason a groupBy on a low-cardinality column (say 12 regions) into 200 partitions leaves 188 tasks with nothing to do.

This is also the rule Spark applies to itself. A join, a groupBy, a window function and a distinct all declare that they need rows clustered by their key, and the planner satisfies that with a HashPartitioning exchange. You do not have to write repartition for the hash to happen; you only get to choose whether yours happens too, in addition.

Rule 3: coalesce moves nothing, and that is the catch

coalesce(n) is not a small repartition. It never sends a row over the network. It just tells n tasks to each read several of the existing partitions, so partition sizes add up: merge three 50,000-row partitions and you get one task with 150,000 rows.

Two behaviours catch people. First, coalesce(n) cannot increase the partition count — with 12 partitions, coalesce(40) silently leaves you 12, because adding partitions requires moving rows. Select df.coalesce(n) and drag n above 12 to see the log say so.

Second, and more expensively: coalesce narrows the stage it is in, not just the step you wrote. If you read 2,000 files, run three withColumn expressions, then coalesce(1) to write a single file, all 2,000 files are read and all three expressions are evaluated by one task, on one core. This is the mechanism behind "coalesce(1) ran out of memory" — the one surviving task is holding the whole dataset. repartition(1) does not have this problem, because the shuffle in the middle lets the upstream work stay wide; it costs you a full write-and-read of the data to buy that.

Why repartition before a join usually buys nothing

A join requires both sides to be clustered by the join key, and to be clustered into the same number of partitions — task 7 on the left must meet task 7 on the right. The rule that enforces this is EnsureRequirements, and in Spark 3.x it works like this:

• Any child that does not already satisfy the requirement gets a shuffle exchange to HashPartitioning(key, spark.sql.shuffle.partitions) — 200 partitions by default.
• Then, if both children came out of shuffles, Spark picks one side's partitioning for both. A side is only a candidate if its partition count is at least spark.sql.shuffle.partitions, and among the candidates the largest count wins.

So repartition(96, "key") before a join, with the default 200, is discarded: 96 is below 200, your side is not a candidate, and it gets shuffled a second time. You paid for two full shuffles and got the layout you would have had for free. Set the operator to df.repartition(n, "key"), tick the join, and drag n past 200 — the log flips from "your side is shuffled again" to "your layout is adopted for both sides". The biggest task does not change.

That last sentence is the whole point of the free panel above. The reason repartition cannot fix skew is not that Spark discards it. It is that the layout Spark would have built is the same hash function, and a hash cannot split one key. The only things that help a single hot key are changing the key (salting), removing the shuffle (broadcasting the small side), or letting adaptive query execution (AQE) slice the partition at runtime — which it also cannot do when the partition is one key.

Partition pruning: the one place partition count is decided by bytes

Everything above is about shuffles. On the read side there is no hash at all. Spark lists the files it needs, cuts them into chunks, and packs the chunks into partitions until each is full. Two separate things decide how much data that is.

Partition pruning is directory elimination. A table written with partitionBy("dt") stores rows in paths like /events/dt=2026-08-07/. A filter that mentions only dt is evaluated against the directory names before any file is opened, so unmatched directories are never listed. This is not the same thing as predicate pushdown: pushdown sends a filter into an opened Parquet file to skip row groups, and it can only skip data whose min/max range excludes the value. Pruning avoids the file; pushdown avoids part of a file it has already opened.

Byte packing then decides the task count, using this rule from FilePartition.maxSplitBytes:

totalBytes    = sum(fileLength + openCostInBytes)   // openCostInBytes default 4 MB
bytesPerCore  = totalBytes / minPartitionNum        // minPartitionNum default = default parallelism
maxSplitBytes = min(maxPartitionBytes, max(openCostInBytes, bytesPerCore))

Files are then cut at maxSplitBytes and greedily packed, with each file charged an extra openCostInBytes so that many small files do not become many tiny tasks. Move the controls and watch which of the three terms is the binding one — the answer flips as the filter gets tighter.

Three things to reach for. Drag WHERE dt matches down to 1: bytes read falls, and the task count falls with it — but not below the point where bytesPerCore takes over from maxPartitionBytes, after which Spark deliberately makes tasks smaller than 128 MB so your cores are not idle. Push files per day to 120 and bytes per day down to 0.4 GB: each file is now 3 MB, smaller than the 4 MB open cost charged for opening it, so more than half of what Spark is packing is not data at all. The bars go warm and the task count stops tracking the bytes — you are scheduling file handles, which is the small file problem. And tick country = 'US': bytes read does not move, because country is not in the path.

That last one is the practical failure. Partitioning a table by a column nobody filters on costs you directory sprawl and buys you nothing; filtering on a column the table is not partitioned by reads everything and throws most of it away after decompression. Within a listed file, the only thing that can still save you is Parquet row-group statistics, which need the data to be physically clustered by that column — the job that Z-ordering exists to do.

Checking it in a real cluster

Four specific places to look, in the order you should look at them.

• Is your layout surviving? Run df.explain() and count the Exchange nodes. Two consecutive Exchange hashpartitioning(key, …) nodes with different partition counts means the first one — usually yours — was discarded. One Exchange where you wrote a repartition and expected the join to reuse it means it worked.
• Is one task doing all the work? In the Spark UI, open the stage and read the Shuffle Read Size / Records summary row. Compare the Max column to the Median column. A 40× ratio there is a hot key, and no partition count will change it — that is the shuffle lesson.
• Did pruning happen? In the physical plan, the FileScan node prints PartitionFilters: [isnotnull(dt), (dt = 2026-08-07)] and a number of partitions read metric. If your filter shows up under PushedFilters instead of PartitionFilters, no directory was skipped. The files pruned and size of files read metrics on the scan node give you the bytes.
• Is your task count coming from the byte rule? If a scan produces exactly your core count in tasks rather than totalBytes / 128 MB, bytesPerCore is the binding term, and raising maxPartitionBytes will do nothing at all.

Next: what actually happens to those rows once they are hashed, in the shuffle; how the planner guesses partition sizes before any of this runs, in table statistics; and what the one fat task is doing to the executor it lands on, in executor memory.