What are the partitioning best practices for performance?

Summary

• In Apache Spark, each partition becomes a task. Shuffles use spark.sql.shuffle.partitions (default 200), and file scans pack up to spark.sql.files.maxPartitionBytes (default 128 MB) into each partition.
• Adaptive Query Execution (AQE), on by default since Spark 3.2.0, coalesces small shuffle partitions at runtime. It ignores the 64 MB advisory size until you set spark.sql.adaptive.coalescePartitions.parallelismFirst to false.
• The COALESCE, REPARTITION, REPARTITION_BY_RANGE and REBALANCE hints control output partitions and file counts.
• On storage, avoid high-cardinality partition columns and partition only when each partition holds at least 1 GB (Delta Lake's rule of thumb). Liquid clustering and Iceberg's hidden partitioning let the layout change without rewriting existing data.

What are the partitioning best practices for performance?

Partitioning means two things: how Apache Spark splits data into partitions that run as parallel tasks, and how a Delta Lake or Apache Iceberg table is laid out on storage. Both affect performance.

How many partitions should Spark use?

Each partition becomes a task. Spark's tuning guide says clusters "will not be fully utilized unless you set the level of parallelism for each operation high enough," and recommends 2-3 tasks per CPU core in your cluster. These settings control partition counts:

Configuration	Default	What it controls
spark.sql.shuffle.partitions	200	Partitions used when shuffling data for joins or aggregations
spark.sql.files.maxPartitionBytes	128 MB	Maximum bytes packed into one partition when reading file-based sources such as Parquet
spark.sql.adaptive.enabled	true	Turns Adaptive Query Execution (AQE) on or off
spark.sql.adaptive.coalescePartitions.enabled	true	Coalesces contiguous shuffle partitions after a shuffle
spark.sql.adaptive.coalescePartitions.parallelismFirst	true	Ignores the advisory size and only respects the 1MB minimum partition size
spark.sql.adaptive.advisoryPartitionSizeInBytes	64 MB	Target size for coalescing small partitions and splitting skewed ones

How does AQE tune shuffle partitions at runtime?

With AQE you don't need to fit the shuffle partition number to each dataset. Spark "can pick the proper shuffle partition number at runtime once you set a large enough initial number of shuffle partitions" through spark.sql.adaptive.coalescePartitions.initialPartitionNum, which falls back to spark.sql.shuffle.partitions.
One default changes what coalescing does. While parallelismFirst is true, Spark ignores the 64 MB advisoryPartitionSizeInBytes target to maximize parallelism and only respects the 1MB spark.sql.adaptive.coalescePartitions.minPartitionSize. The docs recommend setting it to false on a busy cluster:

spark.conf.set("spark.sql.adaptive.coalescePartitions.parallelismFirst", "false")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "64MB")

How do you control output partitions and file counts?

COALESCE, REPARTITION and REPARTITION_BY_RANGE match the coalesce, repartition and repartitionByRange Dataset APIs. REBALANCE exists only as a hint:

SELECT /*+ COALESCE(3) */ * FROM t;
SELECT /*+ REPARTITION(3, c) */ * FROM t;
SELECT /*+ REPARTITION_BY_RANGE(3, c) */ * FROM t;
SELECT /*+ REBALANCE(c) */ * FROM t;

• COALESCE reduces to the given number of partitions. Like DataFrame.coalesce, it creates a narrow dependency, so going from 1000 to 100 partitions doesn't shuffle. A drastic coalesce, such as to one partition, can run the work on fewer nodes than you want, and repartition() avoids that by adding a shuffle.
• REPARTITION takes a partition number, columns, or both. REPARTITION_BY_RANGE needs column names and an optional number.
• REBALANCE makes output partitions "of a reasonable size (not too small and not too big)", splitting skewed ones on a best-effort basis, which helps when writing a table. It's ignored if AQE isn't enabled.

With several hints, the optimizer picks the leftmost.

How should you partition tables on storage?

Delta Lake's best practices give two rules of thumb: don't partition by a column whose cardinality will be very high, such as a user ID, and only partition by a column if each partition will hold at least 1 GB of data.
Delta Lake 3.1.0 and above adds liquid clustering (CLUSTER BY), which "improves the existing partitioning and ZORDER techniques." You can redefine clustering columns without rewriting existing data, and it suits tables filtered by high-cardinality columns. Clustering isn't compatible with partitioning or ZORDER.
Apache Iceberg uses hidden partitioning: it derives partition values through transforms such as bucket(16, id) or days(ts), so queries don't depend on the physical layout. ALTER TABLE ... ADD PARTITION FIELD evolves the spec as a metadata operation. New data uses the new layout and existing data keeps the old one. Inspect partitions with SELECT * FROM prod.db.table.partitions.

How does Storage Partition Join avoid shuffles?

Storage Partition Join (SPJ) uses the existing storage layout to skip the shuffle, for compatible V2 data sources. The Spark docs join two Iceberg tables with identical partitioning on all partition keys:

CREATE TABLE prod.db.target (id INT, salary INT, dep STRING)
USING iceberg
PARTITIONED BY (dep, bucket(8, id));
-- prod.db.source uses the same schema and partitioning

SET 'spark.sql.sources.v2.bucketing.enabled' 'true'
SET 'spark.sql.iceberg.planning.preserve-data-grouping' 'true'
SET 'spark.sql.sources.v2.bucketing.pushPartValues.enabled' 'true'
SET 'spark.sql.requireAllClusterKeysForCoPartition' 'false'
SET 'spark.sql.sources.v2.bucketing.partiallyClusteredDistribution.enabled' 'true'

EXPLAIN SELECT * FROM target t INNER JOIN source s
ON t.dep = s.dep AND t.id = s.id

When SPJ applies, the plan has no Exchange nodes before the join. spark.sql.iceberg.planning.preserve-data-grouping is an Iceberg setting (default false), and spark.sql.requireAllClusterKeysForCoPartition defaults to requiring join keys that match the partition keys in order.

How do you handle skewed partitions?

AQE's skew-join handling is on by default (spark.sql.adaptive.skewJoin.enabled). In a sort-merge join, a partition counts as skewed when it's larger than spark.sql.adaptive.skewJoin.skewedPartitionFactor (default 5.0) times the median partition size and also larger than spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes (default 256MB). Spark then splits, and replicates if needed, the skewed tasks into roughly evenly sized tasks.

How does Databricks handle partitioning?

Databricks recommends liquid clustering for all managed tables. Its docs say most tables with less than 100 TB of data don't need partitioning, because unpartitioned tables are automatically clustered by ingestion time. If you do partition, verify that each partition contains at least 1 GB of data. Setting spark.sql.shuffle.partitions to auto enables auto-optimized shuffle, and auto is the default on serverless compute.

What else do people ask?

Should I always maximize the number of partitions?

No. Databricks' AQE docs note that very small tasks have worse I/O throughput and suffer more from scheduling and task setup overhead. The Spark docs likewise recommend turning off parallelismFirst on a busy cluster to avoid many small tasks.