Databricks Query Optimization Patterns: File Sizing, AQE, Join Strategy, and Spill Mitigation

Databricks query performance is driven mainly by four levers: right-sizing files, leveraging AQE (Adaptive Query Execution), choosing the right join strategy, and avoiding spill (disk overflow). This article walks through the design patterns and parameter settings for each lever, framed for both the Data Engineer Professional exam and real-world production use.

File Size Design

Delta Lake and Parquet file sizes directly drive read performance. Files that are too small inflate metadata overhead and task-startup cost (the Small File Problem), while files that are too large reduce intra-partition parallelism.

-- Run OPTIMIZE (can be combined with ZORDER)
OPTIMIZE gold.orders ZORDER BY (customer_id, order_date);

-- Set the target file size via table properties
ALTER TABLE gold.orders
SET TBLPROPERTIES (
  'delta.targetFileSize' = '134217728'  -- 128MB
);

-- Enable autoOptimize
ALTER TABLE silver.events
SET TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact' = 'true'
);

Set ZORDER BY on the columns that show up most often in filter predicates. ZORDER spatially clusters data by column value to maximize Data Skipping. It is less effective on very high-cardinality columns (like UUIDs), so keep it limited to 3-4 columns.

AQE (Adaptive Query Execution) Configuration Parameters

AQE dynamically re-optimizes query plans using runtime statistics. It is enabled by default on Databricks and automatically performs the following three optimizations:

• Coalescing Post-Shuffle Partitions: Coalesces shuffle output partitions that turn out to be too small
• Converting Sort-Merge Join to Broadcast Hash Join: Switches to Broadcast Hash Join when a side proves small enough at runtime
• Optimizing Skew Joins: Splits partitions skewed toward specific keys to parallelize them

Join Optimization: How to Choose a Strategy

Spark has three main join strategies, chosen based on table size and data characteristics. With AQE enabled they can switch at runtime, but understanding them at design time is still essential.

-- Explicitly broadcast a small table (hint syntax)
SELECT /*+ BROADCAST(dim_products) */
  f.order_id,
  f.amount,
  d.product_name
FROM fact_orders f
JOIN dim_products d ON f.product_id = d.product_id;

-- Tune autoBroadcastJoinThreshold
SET spark.sql.autoBroadcastJoinThreshold = 50m;  -- broadcast tables up to 50MB

The BROADCAST hint overrides the optimizer's decision. For joins between a dimension table (tens of thousands to hundreds of thousands of rows) and a fact table (hundreds of millions of rows), the standard play is to put a BROADCAST hint on the dimension side.

Designing Shuffle Partition Counts

spark.sql.shuffle.partitions controls the number of partitions during shuffles. The default is 200, but you should tune it based on data volume.

• Small data (a few GB or less): 200 is overkill. Reduce the partition count to cut task overhead.
• Large data (hundreds of GB or more): with 200 partitions, each one is too big and spill kicks in. Increase the partition count.
• With AQE enabled: coalescePartitions automatically merges small partitions, so it is safe to set this on the high side.

-- Tune the shuffle partition count
SET spark.sql.shuffle.partitions = 400;

-- Advisory partition size when AQE is enabled
SET spark.sql.adaptive.advisoryPartitionSizeInBytes = 128m;

Spill (Disk Overflow) Mitigation

Spill is what happens when shuffle, sort, or aggregation runs out of memory and offloads data to disk. It tanks performance, so designing to minimize spill matters.

-- Metrics to check in the Spark UI
-- Stages > Task Details:
--   Shuffle Spill (Memory): bytes spilled from memory to disk
--   Shuffle Spill (Disk):   bytes written to disk

-- Example mitigations when spill is happening
SET spark.sql.shuffle.partitions = 800;    -- finer-grained partitions
SET spark.sql.adaptive.advisoryPartitionSizeInBytes = 64m; -- smaller AQE advisory size

Leveraging the Photon Engine

Photon is Databricks' native vectorized execution engine that accelerates the Scan, Filter, Aggregation, and Join stages. Just use a Photon-enabled cluster and it kicks in automatically — no SQL rewrites required.

• Scan: Accelerates columnar Parquet/Delta reads with SIMD instructions
• Filter/Aggregation: Executes in native C++ code, eliminating JVM GC overhead
• Join: Hash-table construction is accelerated in native code
• Caveat: UDFs and Python UDFs are not Photon-supported. Stick to built-in SQL/DataFrame functions to get the most out of Photon.

Optimization Checklist

• Run OPTIMIZE + ZORDER on tables regularly to keep file sizes around 128MB
• Enable autoOptimize on streaming tables
• Run AQE on default-enabled and tune advisoryPartitionSizeInBytes to your data volume
• Consider a BROADCAST hint when joining against dimension tables
• Monitor Shuffle Spill (Disk) on the Spark UI Stages tab and redesign any query that spills consistently
• Use Photon-enabled clusters and write logic with built-in SQL functions whenever possible

Check Your Understanding

Frequently Asked Questions