Fabric Spark Configuration and Performance Tuning: shuffle.partitions, autoBroadcastJoinThreshold, maxPartitionBytes, AQE, Autotune, Native Execution Engine, and Every Setting Data Engineers Must Know

Your Fabric notebook runs. It produces the right output. But it takes 45 minutes. The same logic on 10% of the data took 30 seconds. What happened? You did not change the code — the data just grew. The problem is not your transformations, your joins, or your filters. The problem is the default Spark configuration. The defaults are designed to “just work” for any workload — but “just work” is not the same as “work well.” A 200-partition shuffle on a 10,000-row table creates 200 tiny tasks that spend more time scheduling than processing. A 10 MB broadcast threshold means your 50 MB dimension table triggers a full shuffle join instead of broadcasting to every executor. A 128 MB partition size means your 10 GB file creates 80 partitions when 20 would be faster.

This post covers every Spark configuration that matters for Fabric data engineers — what each setting does, why the default exists, when to change it, and what value to use. These are the three settings that Fabric’s Autotune feature optimizes automatically, plus the broader configuration landscape you need to understand for production pipelines and the DP-700 exam.

Real-life analogy: Spark configuration is like adjusting a car’s settings for different roads. The factory defaults (Spark defaults) work on any road — highway, city, gravel. But a race car driver tunes the suspension for the specific track. spark.sql.shuffle.partitions is like choosing how many lanes the highway has — too few lanes and traffic jams (data skew), too many lanes and each lane has one car (overhead). spark.sql.autoBroadcastJoinThreshold is like deciding whether to bring a map to every passenger (broadcast) or make everyone stop at one gas station for directions (shuffle). spark.sql.files.maxPartitionBytes is like choosing how much cargo each truck carries — too small and you need too many trucks, too large and each truck is overloaded. Tuning these settings is the difference between a 45-minute commute and a 5-minute one — same route, different configuration.

Table of Contents

• Where to Set Spark Configurations in Fabric
• Spark Environments (Persistent)
• %%configure Magic Command (Session-Level)
• spark.conf.set() (Runtime)
• Mutable vs Immutable Properties
• The Big Three — Settings That Matter Most
• spark.sql.shuffle.partitions
• spark.sql.autoBroadcastJoinThreshold
• spark.sql.files.maxPartitionBytes
• Adaptive Query Execution (AQE)
• What AQE Does
• AQE Settings
• Coalesce Partitions
• Auto Broadcast Join Conversion
• Skew Join Optimization
• AQE vs Manual Tuning
• Autotune — Fabric’s ML-Based Optimizer
• What Autotune Does
• How to Enable Autotune
• When Autotune Works Best
• Checking Autotune Recommendations
• Native Execution Engine (NEE)
• What NEE Is
• How to Enable NEE
• NEE Limitations
• Memory and Resource Configuration
• Driver and Executor Memory
• Node Sizes in Fabric
• Starter Pools vs Custom Pools
• High Concurrency Mode
• Delta Lake Optimization Settings
• Optimize Write
• Auto Compaction
• V-Order
• Target File Size
• Join Strategy Deep Dive
• Broadcast Hash Join
• Sort-Merge Join
• Shuffle Hash Join
• Choosing the Right Strategy
• Production Configuration Templates
• Small Data (Under 1 GB)
• Medium Data (1–50 GB)
• Large Data (50–500 GB)
• Massive Data (500 GB+)
• How to Diagnose Performance Issues
• Reading the Spark UI
• Identifying Shuffle Bottlenecks
• Identifying Data Skew
• Common Mistakes
• Interview Questions
• Wrapping Up

Where to Set Spark Configurations in Fabric

Spark configurations can be set at three levels in Fabric, each with different scope and persistence. Understanding where to set them is as important as knowing what to set.

Spark Environments (Persistent)

A Spark Environment is a reusable configuration object in Fabric that defines the Spark runtime version, pre-installed libraries, and Spark properties. Settings defined in an Environment apply to every notebook and Spark job that uses that environment — they are persistent and do not need to be set in each notebook.

# To configure an Environment:
# 1. Go to your Workspace → New → Environment
# 2. Click on the Environment → Spark Properties tab
# 3. Add key-value pairs:
#    spark.sql.shuffle.partitions = 50
#    spark.sql.autoBroadcastJoinThreshold = 104857600
#    spark.sql.adaptive.enabled = true
# 4. Save and Publish the Environment
# 5. Attach the Environment to your notebook (Settings → Environment)

# Environment settings are the RECOMMENDED approach for production
# because they are consistent across all notebooks and do not require
# each developer to remember to set them

%%configure Magic Command (Session-Level)

The %%configure magic command sets Spark properties before the session starts. It must be the first cell in your notebook (or the first line of a Spark Job Definition). It can set both mutable and immutable properties, including driver/executor memory and cores.

%%configure -f
{
    "driverMemory": "56g",
    "driverCores": 32,
    "executorMemory": "28g",
    "executorCores": 4,
    "conf": {
        "spark.sql.shuffle.partitions": "50",
        "spark.sql.autoBroadcastJoinThreshold": "104857600",
        "spark.sql.files.maxPartitionBytes": "268435456",
        "spark.sql.adaptive.enabled": "true",
        "spark.native.enabled": "true",
        "spark.shuffle.manager": "org.apache.spark.shuffle.sort.ColumnarShuffleManager"
    }
}

# IMPORTANT: %%configure must be the FIRST cell in the notebook
# IMPORTANT: Driver/executor settings go at the ROOT level, not inside "conf"
# IMPORTANT: The -f flag forces a session restart if one is already active

# ❌ WRONG — driver/executor memory inside "conf"
# {"conf": {"driverMemory": "56g"}}

# ✅ CORRECT — driver/executor memory at root level
# {"driverMemory": "56g", "conf": {"spark.sql.shuffle.partitions": "50"}}

spark.conf.set() (Runtime)

Use spark.conf.set() to change mutable properties during a running session. This is the most flexible approach — you can change settings between cells, adjust for different stages of your pipeline, or tune based on data characteristics discovered at runtime.

# Set configuration at runtime — applies to all subsequent operations
spark.conf.set("spark.sql.shuffle.partitions", "50")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "104857600")  # 100 MB
spark.conf.set("spark.sql.files.maxPartitionBytes", "268435456")     # 256 MB

# Read current configuration value
current = spark.conf.get("spark.sql.shuffle.partitions")
print(f"Shuffle partitions: {current}")

# Dynamic tuning based on data size
row_count = df.count()
if row_count < 100_000:
    spark.conf.set("spark.sql.shuffle.partitions", "10")
elif row_count < 10_000_000:
    spark.conf.set("spark.sql.shuffle.partitions", "50")
else:
    spark.conf.set("spark.sql.shuffle.partitions", "200")

# Only MUTABLE properties can be changed with spark.conf.set()
# Immutable properties (driver memory, executor cores, shuffle manager)
# can ONLY be set via %%configure or Environments

Mutable vs Immutable Properties

Property Type	Can Change at Runtime?	Where to Set	Examples
Mutable	Yes — spark.conf.set()	%%configure, Environment, or spark.conf.set()	spark.sql.shuffle.partitions, spark.sql.autoBroadcastJoinThreshold, spark.sql.adaptive.enabled
Immutable	No — must be set before session starts	%%configure or Environment only	spark.driver.memory, spark.executor.memory, spark.executor.cores, spark.shuffle.manager

Key point: If you try to change an immutable property with spark.conf.set(), it will silently accept the value but not apply it. The session was already started with the original value. This is a common source of confusion — your code runs without error, but the setting has no effect. Always use %%configure or an Environment for immutable properties.

The Big Three — Settings That Matter Most

These three settings have the largest impact on Spark performance, and they are the three settings that Fabric’s Autotune feature optimizes automatically. Understanding them deeply is essential for the DP-700 exam and for building production pipelines.

spark.sql.shuffle.partitions

What it does: Controls the number of partitions created after a shuffle operation — any operation that requires redistributing data across executors: groupBy(), join(), orderBy(), distinct(), repartition().

Default: 200

Why it matters: The default of 200 is a compromise — reasonable for medium data, but terrible for small and large data. For a 10,000-row DataFrame, 200 shuffle partitions means 200 tasks with ~50 rows each. The overhead of scheduling and managing 200 tasks far exceeds the time to process 50 rows. For a 100 GB DataFrame, 200 partitions means each partition holds 500 MB — too large for efficient processing, causing memory pressure and spills to disk.

# Default: 200 shuffle partitions
spark.conf.set("spark.sql.shuffle.partitions", "200")  # default

# What happens during a shuffle:
# df.groupBy("department").agg(sum("salary"))
#
# BEFORE shuffle: data is spread across READ partitions (from the source file)
# DURING shuffle: Spark redistributes data so all rows with the same "department"
#                 are on the same partition → this creates SHUFFLE PARTITIONS
# AFTER shuffle: the aggregation runs on each shuffle partition independently
#
# If you have 200 shuffle partitions but only 5 departments,
# 195 partitions are EMPTY — pure overhead

# Guidelines for setting shuffle partitions:
# Small data   (< 1 GB):     10 - 20 partitions
# Medium data  (1 - 10 GB):  20 - 100 partitions
# Large data   (10 - 100 GB): 100 - 500 partitions
# Massive data (100 GB+):     500 - 2000 partitions

# Rule of thumb: aim for 100-200 MB per partition after shuffle
# 10 GB data ÷ 100 MB per partition = 100 partitions
# 50 GB data ÷ 200 MB per partition = 250 partitions

# Set it:
spark.conf.set("spark.sql.shuffle.partitions", "50")  # For medium datasets

Real-life analogy: Shuffle partitions are like checkout lanes at a grocery store. If you open 200 checkout lanes for 10 customers, 190 cashiers sit idle — wasteful. If you open 5 checkout lanes for 10,000 customers, each line is enormous — slow. The right number of lanes matches the number of customers: enough to keep everyone moving, few enough that no lane is empty.

# DEMONSTRATION: Impact of shuffle partitions on a small dataset

# ❌ Default 200 partitions on a small table
spark.conf.set("spark.sql.shuffle.partitions", "200")
df_small = spark.range(10000).groupBy((col("id") % 5).alias("group")).count()
print(df_small.rdd.getNumPartitions())   # 200 — 195 are empty!

# ✅ Right-sized: 10 partitions
spark.conf.set("spark.sql.shuffle.partitions", "10")
df_small = spark.range(10000).groupBy((col("id") % 5).alias("group")).count()
print(df_small.rdd.getNumPartitions())   # 10 — much better

# With AQE enabled (recommended), Spark can AUTOMATICALLY coalesce
# the 200 empty partitions down to ~5. But setting the right value
# upfront avoids the coalescing overhead entirely.

spark.sql.autoBroadcastJoinThreshold

What it does: Sets the maximum table size (in bytes) that Spark will automatically broadcast to every executor during a join. If a table is smaller than this threshold, Spark copies the entire table to every executor — eliminating the need to shuffle the larger table.

Default: 10 MB (10485760 bytes). Note: some Fabric documentation states the default is 25.6 MB in Fabric Starter Pools, but the standard Spark default is 10 MB.

Why it matters: A shuffle join (sort-merge join) is the most expensive operation in Spark — both sides of the join must be shuffled so that matching keys end up on the same executor. This means all data moves across the network. A broadcast join avoids this entirely — the small table is copied to every executor, and each executor joins its local partition of the large table with the complete small table. No shuffle, no network transfer for the large table.

# Default: 10 MB — tables smaller than this are automatically broadcast
spark.conf.get("spark.sql.autoBroadcastJoinThreshold")   # "10485760" (10 MB)

# Increase to 100 MB — broadcast dimension tables up to 100 MB
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "104857600")

# Increase to 256 MB — aggressive broadcasting for large dimension tables
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "268435456")

# Disable automatic broadcasting entirely (force sort-merge joins)
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")

# WHY increase from the default?
# Most dimension tables (products, customers, regions, dates) are 10-200 MB
# At the default 10 MB, a 50 MB dimension table triggers a sort-merge join
# with a 100 GB fact table — shuffling ALL 100 GB across the network
# Increasing to 100 MB means the 50 MB dimension is broadcast instead,
# eliminating the shuffle of the 100 GB fact table entirely

Real-life analogy: Imagine joining a million-row fact_sales table with a 10,000-row dim_product table. A shuffle join is like making every salesperson in 100 offices call headquarters to ask for each product’s name — 100 million phone calls. A broadcast join is like printing the product catalog and mailing one copy to every office — 100 copies of a small catalog, and now every office can look up products locally. The catalog is small (dimension table), the offices are many (executors), and local lookups (broadcast join) are infinitely faster than phone calls (shuffle).

# Manual broadcast hint — works even if the table exceeds the threshold
from pyspark.sql.functions import broadcast

# Broadcast the small dimension table explicitly
result = fact_sales.join(
    broadcast(dim_product),   # Force broadcast regardless of size
    "product_id"
)

# When to use manual broadcast vs auto threshold:
# - Auto threshold: set once, applies to ALL joins in the session
# - Manual hint: apply per join, when you KNOW a table is small
#   even if Spark's statistics say otherwise

# ⚠ WARNING: Broadcasting a table that is too large causes OutOfMemoryError
# Rule: Never broadcast a table larger than 20% of executor memory
# If executors have 8 GB, max broadcast size ≈ 1.5 GB
# If executors have 64 GB, max broadcast size ≈ 12 GB

# Check if a join used broadcast (look at the physical plan):
result.explain()
# Look for "BroadcastHashJoin" (good) vs "SortMergeJoin" (shuffle)

spark.sql.files.maxPartitionBytes

What it does: Controls the maximum number of bytes packed into a single partition when reading files (Parquet, JSON, ORC, CSV). This determines how Spark splits input files into partitions for parallel processing.

Default: 128 MB (134217728 bytes)

Why it matters: If your data files are large (multi-GB), Spark splits them into partitions based on this setting. 128 MB means a 10 GB file creates ~80 partitions. If your executors have plenty of memory, increasing to 256 MB or 512 MB means fewer, larger partitions — fewer tasks to schedule, less overhead, and potentially faster reads. If your files are small (many 1 MB files), this setting is less relevant — each small file typically becomes its own partition regardless.

# Default: 128 MB per partition when reading files
spark.conf.get("spark.sql.files.maxPartitionBytes")   # "134217728" (128 MB)

# Increase to 256 MB — fewer, larger partitions (less scheduling overhead)
spark.conf.set("spark.sql.files.maxPartitionBytes", "268435456")   # 256 MB

# Increase to 512 MB — for large datasets with ample executor memory
spark.conf.set("spark.sql.files.maxPartitionBytes", "536870912")   # 512 MB

# When to change:
# INCREASE (256-512 MB) when:
#   - Reading large Parquet/Delta files (10+ GB)
#   - Executors have 64+ GB memory
#   - You see too many small tasks in the Spark UI
#   - Simple transformations (filter, select) that don't need much memory per row
#
# DECREASE (64 MB) when:
#   - Complex transformations that expand row size (explode, UDFs)
#   - Executors have limited memory (starter pools)
#   - You see OutOfMemoryError during reads
#
# LEAVE at default (128 MB) when:
#   - Mixed workloads, starter pools, or uncertain about data characteristics

Real-life analogy: maxPartitionBytes is like choosing how much cargo each delivery truck carries. Small trucks (64 MB) make many trips but never get overloaded. Large trucks (512 MB) make fewer trips but each one is heavier. If you have a highway with no weight limits (plenty of executor memory), use big trucks. If the roads are narrow (limited memory), use smaller trucks. The default 128 MB truck works for most roads but is not optimal for any specific one.

Adaptive Query Execution (AQE)

What AQE Does

Adaptive Query Execution (AQE) is Spark’s ability to re-optimize the query plan at runtime based on actual data statistics observed during execution. Instead of committing to a plan before seeing any data, AQE adjusts the plan between stages — after it knows the actual data sizes, partition counts, and distribution.

Real-life analogy: Without AQE, Spark is like a GPS that plans your route before you leave home — it commits to highways and turns without knowing about traffic, road closures, or construction. With AQE, Spark is like a GPS with live traffic updates — it re-routes between turns based on actual conditions. If it discovers a small table during execution, it switches from a shuffle join to a broadcast join. If it finds empty partitions, it merges them. If it detects data skew, it splits the skewed partition.

AQE Settings

# AQE is ENABLED by default in Fabric (Spark 3.x+)
spark.conf.get("spark.sql.adaptive.enabled")   # "true" in Fabric

# Key AQE sub-settings:

# 1. Coalesce Partitions — merge small post-shuffle partitions
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")    # default: true
spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionSize", "1m")  # minimum 1 MB
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128m")  # target size after coalescing

# 2. Auto Broadcast Join Conversion — switch to broadcast at runtime
spark.conf.set("spark.sql.adaptive.autoBroadcastJoinThreshold", "10485760")  # same as static threshold

# 3. Skew Join Optimization — split skewed partitions
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")               # default: true
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")     # 5x median = skewed
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256m")  # min 256 MB

Coalesce Partitions

After a shuffle, some partitions may be very small or empty. Coalescing merges adjacent small partitions into larger ones, reducing the number of tasks in the next stage. This is particularly valuable when spark.sql.shuffle.partitions is set high (e.g., 200) but the actual data produces many empty partitions (e.g., groupBy on 5 departments creates only 5 non-empty partitions out of 200).

# Without AQE coalescing:
# shuffle.partitions = 200, actual groups = 5
# → 200 tasks scheduled, 195 are empty, 5 do all the work
# → Next stage has 200 input partitions (195 empty reads)

# With AQE coalescing:
# shuffle.partitions = 200, actual groups = 5
# → 200 tasks run, AQE observes 195 empty partitions
# → AQE coalesces to ~5-10 partitions for the next stage
# → Next stage processes only 5-10 tasks — much faster

# This is why AQE makes shuffle.partitions LESS critical —
# set it high (200+) and let AQE reduce it automatically.
# But setting it right upfront is STILL better because it
# avoids the overhead of creating and coalescing empty partitions.

Auto Broadcast Join Conversion

AQE can convert a planned sort-merge join to a broadcast join at runtime — after the shuffle, it discovers that one side is smaller than the broadcast threshold. This is not as efficient as planning a broadcast join from the start (the shuffle already happened), but it avoids the expensive sort-merge comparison by reading shuffle files locally.

Skew Join Optimization

Data skew occurs when one partition has significantly more data than others — for example, joining on customer_id where one customer has 10 million orders and most have 100. Without AQE, the skewed partition takes 100x longer than the others, making the entire stage wait. AQE detects the skew and splits the oversized partition into smaller sub-partitions, processing them in parallel.

# AQE detects skew when a partition is:
# - More than 5x the median partition size (skewedPartitionFactor)
# - AND larger than 256 MB (skewedPartitionThresholdInBytes)

# Example: joining orders (skewed on customer_id = "WALMART")
# Without AQE: WALMART partition has 10M rows, takes 30 minutes
# With AQE: WALMART partition split into 10 sub-partitions, each 1M rows
#           processed in parallel — takes 3 minutes

# AQE handles the split automatically — no code changes needed
# Just ensure these settings are enabled:
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

AQE vs Manual Tuning

AQE is excellent but not a replacement for understanding your data. AQE reacts to observed data — it cannot predict. Manual tuning prevents problems; AQE fixes them after they occur. Best practice: set reasonable defaults manually (shuffle partitions matching your data size, broadcast threshold matching your dimension table sizes), then let AQE fine-tune at runtime.

Autotune — Fabric’s ML-Based Optimizer

What Autotune Does

Autotune is a Fabric-specific feature (currently in preview) that uses machine learning to automatically tune the Big Three settings. It examines each query, builds a separate ML model for that query shape, and learns the best configuration over repeated runs. Unlike AQE (which reacts within a single run), Autotune learns across runs — the more you run a query, the better it tunes.

Autotune optimizes exactly three settings: spark.sql.shuffle.partitions, spark.sql.autoBroadcastJoinThreshold, and spark.sql.files.maxPartitionBytes.

How to Enable Autotune

# Enable Autotune for a single session (notebook or Spark Job Definition)
spark.conf.set("spark.ms.autotune.enabled", "true")

# Or enable via Environment:
# Go to Environment → Spark Properties → Add:
#   spark.ms.autotune.enabled = true

# Or enable via %%configure:
# %%configure -f
# {"conf": {"spark.ms.autotune.enabled": "true"}}

# Autotune is OFF by default — you must explicitly enable it
# It works with notebooks, Spark Job Definitions, and pipelines

When Autotune Works Best

Autotune shows the largest gains for exploratory data analysis patterns — reads, joins, aggregations, and sorts. It learns from your specific query shape, so repeated runs of the same pipeline (daily ETL, scheduled notebooks) benefit the most. The first run uses defaults; subsequent runs apply learned optimizations.

Checking Autotune Recommendations

# After a run with Autotune enabled, check the driver logs
# in the Monitoring Hub for entries starting with [Autotune]:
#
# [Autotune] Recommended spark.sql.shuffle.partitions = 48
# [Autotune] Recommended spark.sql.autoBroadcastJoinThreshold = 52428800
# [Autotune] Recommended spark.sql.files.maxPartitionBytes = 268435456

# These recommendations show what Autotune calculated for your specific query
# If you prefer manual control, use these values as your baseline and tune from there

Native Execution Engine (NEE)

What NEE Is

The Native Execution Engine (NEE) is a Fabric-specific optimization that replaces Spark’s JVM-based execution with a C++ compiled engine based on Velox (a database acceleration library from Meta) and Gluten (a middle layer that offloads Spark SQL to native engines). NEE can be 2-4x faster for read-heavy workloads (Parquet and Delta reads, joins, aggregations) because it avoids JVM overhead, uses vectorized execution, and leverages CPU SIMD instructions.

How to Enable NEE

# Enable NEE via %%configure (must be FIRST cell)
# %%configure -f
# {
#     "conf": {
#         "spark.native.enabled": "true",
#         "spark.shuffle.manager": "org.apache.spark.shuffle.sort.ColumnarShuffleManager"
#     }
# }

# Or via Environment → Spark Properties:
#   spark.native.enabled = true
#   spark.shuffle.manager = org.apache.spark.shuffle.sort.ColumnarShuffleManager

# Or at runtime (only spark.native.enabled — shuffle.manager is IMMUTABLE):
spark.conf.set("spark.native.enabled", "true")
# ⚠ spark.shuffle.manager is IMMUTABLE — can only be set via %%configure or Environment

# IMPORTANT: spark.shuffle.manager is immutable
# Setting it via %%configure causes a full cluster restart (~4 minutes)
# Setting it via Environment avoids this penalty

NEE Limitations

• No UDF support — NEE does not support Python UDFs, Pandas UDFs, or Java UDFs. If your query contains a UDF, Spark automatically falls back to the JVM engine for that operation (no error, just slower).
• Custom pools required — NEE works best on custom pools, not starter pools.
• Not all operations are accelerated — complex operations, custom data sources, and some data types may fall back to the JVM engine.
• Check compatibility — examine the Spark UI to see which operations used NEE (native) vs JVM (fallback).

Memory and Resource Configuration

Driver and Executor Memory

# Driver memory — the "coordinator" node
# - Collects results from .collect(), .toPandas(), .show()
# - Stores broadcast variables before sending to executors
# - Manages the DAG, task scheduling, and metadata
# Increase if: collect() or toPandas() on large data, or broadcasting large tables

# Executor memory — the "worker" nodes
# - Processes data partitions in parallel
# - Stores cached DataFrames
# - Performs shuffles, joins, aggregations
# Increase if: OutOfMemoryError during joins, aggregations, or caching

# Set via %%configure (must be FIRST cell):
# %%configure -f
# {
#     "driverMemory": "28g",
#     "driverCores": 8,
#     "executorMemory": "28g",
#     "executorCores": 4
# }

Node Sizes in Fabric

Node Size	vCores	Memory	Best For
Small	4	32 GB	Development, small data exploration
Medium	8	64 GB	Standard development, starter pools
Large	16	128 GB	Production ETL, medium datasets
X-Large	32	256 GB	Large datasets, complex joins, ML workloads

Starter Pools vs Custom Pools

Feature	Starter Pool	Custom Pool
Startup time	~5-15 seconds (pre-provisioned)	~3-5 minutes (cold start)
Node size	Medium (8 vCores, 64 GB)	Configurable (Small to X-Large)
Sharing	Shared with other tenants	Dedicated to your capacity
NEE support	Limited	Full support
High Concurrency	Not available	Available
Best for	Development, quick exploration	Production workloads

High Concurrency Mode

High Concurrency mode allows multiple notebooks to share a single Spark session. Instead of each notebook starting its own cluster (3-5 minute cold start), the first notebook starts the cluster and subsequent notebooks join the existing session. This can reduce startup time by up to 36x for the second and subsequent notebooks. Requirements: notebooks must share the same Lakehouse, Environment, and Spark configuration.

Delta Lake Optimization Settings

Optimize Write

# Optimize Write coalesces small partitions into larger files at write time
# Prevents the "small files problem" — thousands of tiny Parquet files
# that slow down reads and increase metadata overhead

spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")

# Or set as a table property (persistent, per-table):
# ALTER TABLE silver.customers SET TBLPROPERTIES ('delta.autoOptimize.optimizeWrite' = 'true')

# Without Optimize Write: 200 shuffle partitions → 200 files (many tiny)
# With Optimize Write: 200 shuffle partitions → ~10-20 optimally-sized files

Auto Compaction

# Auto Compaction runs a mini OPTIMIZE after every write operation
# Compacts small files automatically — no manual OPTIMIZE needed for most tables

spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

# Auto Compaction is less thorough than manual OPTIMIZE
# It targets the most fragmented files, not the entire table
# For large production tables, still schedule periodic manual OPTIMIZE

V-Order

V-Order is a Fabric-specific write-time optimization that sorts and organizes data within Parquet files for maximum read performance. It is enabled by default when writing Delta tables in Fabric notebooks. V-Order optimized files are 15-50% faster to read for analytical queries and work seamlessly with Power BI’s Direct Lake mode.

# V-Order is ON by default in Fabric — usually no action needed
spark.conf.get("spark.sql.parquet.vorder.enabled")   # "true" in Fabric

# Disable only if you need compatibility with non-Fabric Spark readers:
spark.conf.set("spark.sql.parquet.vorder.enabled", "false")

Target File Size

# Controls the target size of Delta files written by OPTIMIZE
# Default varies by context:
# - 128-256 MB for tables read by Direct Lake (Power BI)
# - 256 MB - 1 GB for large tables read by Spark

# Set as a table property:
# ALTER TABLE gold.fact_sales SET TBLPROPERTIES ('delta.targetFileSize' = '256mb')

# Smaller files (128 MB): better for Direct Lake, frequent small queries
# Larger files (1 GB): better for full-table scans in Spark

Join Strategy Deep Dive

Broadcast Hash Join

How it works: The small table is collected by the driver, then broadcast to every executor. Each executor builds a hash table from the small table and probes it with its local partition of the large table. No shuffle of the large table.

When Spark uses it: Automatically when one side is smaller than autoBroadcastJoinThreshold. Or when you use the broadcast() hint.

Best for: Joining a large fact table with a small dimension table (star schema pattern). The most common join in data warehousing.

Sort-Merge Join

How it works: Both tables are shuffled by the join key (all rows with the same key end up on the same partition), then each partition is sorted by the join key, then the sorted partitions are merged. This is the most expensive join but works for any data size.

When Spark uses it: Default for joins where both sides exceed the broadcast threshold.

Best for: Large-to-large table joins (two fact tables, both 100+ GB).

Shuffle Hash Join

How it works: Both tables are shuffled by the join key (like sort-merge), but instead of sorting, Spark builds a hash table from the smaller side and probes it with the larger side. Faster than sort-merge when the smaller side fits in memory, because it avoids the sort step.

When Spark uses it: AQE may convert a sort-merge join to a shuffle hash join when it discovers one side is small enough.

Choosing the Right Strategy

Scenario	Best Strategy	How to Ensure
Large fact + small dimension (< 200 MB)	Broadcast Hash Join	Increase autoBroadcastJoinThreshold or use broadcast()
Large fact + medium dimension (200 MB – 2 GB)	Broadcast if memory allows	Use broadcast() hint with caution
Large + large (both > 2 GB)	Sort-Merge Join	Default. Optimize shuffle partitions for parallelism
Skewed join keys	Sort-Merge with AQE skew handling	Enable AQE skew join. Or salt the join key manually

Production Configuration Templates

Small Data (Under 1 GB)

# Small data — development, testing, small reference tables
spark.conf.set("spark.sql.shuffle.partitions", "10")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "104857600")  # 100 MB
spark.conf.set("spark.sql.files.maxPartitionBytes", "134217728")     # 128 MB (default)
spark.conf.set("spark.sql.adaptive.enabled", "true")

# Consider: for data under 1 GB, pandas may be faster than Spark
# Spark has fixed overhead (session startup, task scheduling)
# that dominates processing time for small datasets

Medium Data (1–50 GB)

# Medium data — daily ETL, standard production workloads
spark.conf.set("spark.sql.shuffle.partitions", "50")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "104857600")  # 100 MB
spark.conf.set("spark.sql.files.maxPartitionBytes", "268435456")     # 256 MB
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

Large Data (50–500 GB)

# Large data — production ETL on major fact tables
spark.conf.set("spark.sql.shuffle.partitions", "200")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "268435456")  # 256 MB
spark.conf.set("spark.sql.files.maxPartitionBytes", "268435456")     # 256 MB
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")
spark.conf.set("spark.native.enabled", "true")

Massive Data (500 GB+)

# Massive data — migrations, large historical loads, data science on full datasets
spark.conf.set("spark.sql.shuffle.partitions", "1000")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "536870912")  # 512 MB
spark.conf.set("spark.sql.files.maxPartitionBytes", "536870912")     # 512 MB
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")
spark.conf.set("spark.native.enabled", "true")
# Use Custom Pools with Large or X-Large nodes
# Consider partitioning source tables by date for partition pruning

How to Diagnose Performance Issues

Reading the Spark UI

The Spark UI (accessible from the Monitoring Hub → click on a notebook run → Spark UI) shows the execution details for every job, stage, and task. The key tabs to check:

• Jobs tab: shows each action (write, collect, show) and its stages. Look for stages that took significantly longer than others.
• Stages tab: shows input/output size, shuffle read/write, and task distribution. Look for stages with high shuffle write.
• SQL/DataFrame tab: shows the physical plan and which join strategy was used (BroadcastHashJoin vs SortMergeJoin). Check if your intended broadcast is actually happening.
• Executors tab: shows memory usage per executor. Look for executors near their memory limit.

Identifying Shuffle Bottlenecks

# Signs of shuffle bottlenecks:
# 1. High "Shuffle Write" in the Stages tab (GBs of shuffle data)
# 2. One stage takes much longer than others
# 3. "Spill (Memory)" or "Spill (Disk)" in task metrics

# Fixes:
# - Increase autoBroadcastJoinThreshold to broadcast small tables
# - Use broadcast() hint for medium-sized dimension tables
# - Filter data BEFORE joining (reduce shuffle volume)
# - Use partition pruning (filter on partition columns before join)
# - Increase executor memory to reduce spills

Identifying Data Skew

# Signs of data skew:
# 1. In the Stages tab, one task's duration is 10-100x others
# 2. The Summary Metrics table shows huge gap between Median and Max
#    Median: 2s, Max: 300s ← this is skew
# 3. One executor's Shuffle Read is much larger than others

# Fixes:
# - Enable AQE skew join (automatic partition splitting)
# - Salt the join key: add a random prefix, join on salted key, aggregate
# - If skew is on groupBy, use a two-stage aggregation:
#   Stage 1: groupBy(salted_key) to spread data
#   Stage 2: groupBy(original_key) to combine partial results
# - Filter out the skewed key, process it separately, UNION

Common Mistakes

1. Never changing shuffle partitions from the default 200 — 200 partitions for a 10,000-row table creates 200 nearly-empty tasks with more scheduling overhead than processing time. For small data, set to 10-20. For large data, set to 500-2000. Or enable AQE coalescing.
2. Leaving autoBroadcastJoinThreshold at 10 MB — most dimension tables are 10-200 MB. At the default 10 MB, a 50 MB dimension triggers a full shuffle of your 100 GB fact table. Increase to 100-256 MB for star schema workloads.
3. Trying to set immutable properties with spark.conf.set() — spark.driver.memory, spark.executor.memory, and spark.shuffle.manager are immutable. Setting them at runtime silently does nothing. Use %%configure (first cell) or an Environment.
4. Putting driver/executor memory inside the “conf” block of %%configure — driverMemory and executorMemory go at the root level of the JSON, not inside "conf": {}. Inside “conf,” they are silently ignored.
5. Broadcasting tables that are too large — broadcasting a 5 GB table to executors with 8 GB memory causes OutOfMemoryError. Rule: never broadcast more than 20% of executor memory.
6. Not enabling AQE — AQE is enabled by default in Fabric, but some environments or configurations may disable it. Always verify: spark.conf.get("spark.sql.adaptive.enabled") should return “true.”
7. Ignoring the Spark UI — the Spark UI shows exactly what happened: which join strategy was used, how much data was shuffled, whether partitions were skewed, and where time was spent. Not checking it means guessing instead of diagnosing.
8. Using Spark for small data — for datasets under 1 GB, pandas is often 2-5x faster because Spark has fixed overhead (session startup, task scheduling, serialization). Use Spark when you actually need distributed processing.

Interview Questions

Q: What are the three Spark configurations that Fabric Autotune optimizes? A: spark.sql.shuffle.partitions (number of partitions after shuffle operations — default 200), spark.sql.autoBroadcastJoinThreshold (maximum table size for automatic broadcast joins — default 10 MB), and spark.sql.files.maxPartitionBytes (maximum bytes per partition when reading files — default 128 MB). Autotune uses machine learning to find optimal values for each query shape based on historical execution data.

Q: What is the difference between a broadcast join and a sort-merge join? A: A broadcast join copies the small table to every executor and performs a local hash join — no shuffle of the large table. A sort-merge join shuffles both tables by the join key, sorts each partition, then merges them. Broadcast is much faster when one side is small (under the broadcast threshold) because it avoids shuffling the large table. Sort-merge is the fallback for large-to-large joins.

Q: What is Adaptive Query Execution (AQE) and what does it optimize? A: AQE re-optimizes the query plan at runtime based on actual data statistics. It performs three key optimizations: coalescing small post-shuffle partitions (reduces empty partitions), converting sort-merge joins to broadcast joins when a table is smaller than expected, and splitting skewed partitions to prevent straggler tasks. AQE is enabled by default in Fabric.

Q: What is the difference between mutable and immutable Spark properties? A: Mutable properties (like shuffle.partitions, autoBroadcastJoinThreshold) can be changed at runtime with spark.conf.set(). Immutable properties (like driver.memory, executor.memory, shuffle.manager) must be set before the session starts — via %%configure or an Environment. Trying to change an immutable property at runtime silently does nothing.

Q: What is the Native Execution Engine in Fabric? A: NEE replaces Spark’s JVM execution with a C++ compiled engine based on Velox and Gluten. It can be 2-4x faster for Parquet/Delta reads, joins, and aggregations. Enable with spark.native.enabled = true and spark.shuffle.manager = org.apache.spark.shuffle.sort.ColumnarShuffleManager. Limitations: no UDF support (falls back to JVM), works best on custom pools.

Q: Why should you increase autoBroadcastJoinThreshold from the default 10 MB? A: Most dimension tables in a star schema are 10-200 MB. At the default 10 MB, a 50 MB dimension table triggers a sort-merge join that shuffles the entire large fact table across the network. Increasing to 100-256 MB allows Spark to broadcast the dimension table to every executor, eliminating the shuffle entirely. This is often the single largest performance improvement in data warehouse workloads.

Q: How do you handle data skew in Spark joins? A: Enable AQE skew join optimization (spark.sql.adaptive.skewJoin.enabled = true), which automatically detects and splits oversized partitions. For manual handling: salt the join key (add a random prefix, join on salted key, then aggregate), use a two-stage aggregation for skewed groupBy, or filter the skewed key and process it separately with a UNION.

Q: When should you use pandas instead of Spark in a Fabric notebook? A: For datasets under 1 GB that fit in memory on a single machine. Spark has fixed overhead (session startup, task scheduling, serialization) that makes it slower than pandas for small data. Use pandas.read_parquet() for small files and df.toPandas() when you need pandas-specific operations. Switch to Spark when data exceeds memory or needs distributed processing.

Wrapping Up

Spark performance tuning in Fabric comes down to three layers: the Big Three settings (shuffle.partitions, autoBroadcastJoinThreshold, maxPartitionBytes), Adaptive Query Execution for runtime optimization, and the infrastructure layer (node sizes, custom pools, Native Execution Engine). The defaults are safe but slow. The right configuration for your data size and workload pattern can reduce a 45-minute notebook to 5 minutes without changing a single line of transformation logic.

Start with AQE enabled (default) and Autotune for automatic tuning. Then set the Big Three based on your data size: 10-20 shuffle partitions for small data, 50-200 for medium, 500+ for large. Increase the broadcast threshold to 100-256 MB to broadcast dimension tables instead of shuffling them. And use the Spark UI to verify — check that your joins are using BroadcastHashJoin, your partitions are not skewed, and your shuffle writes are not excessive. Measure, tune, verify. That is the cycle.

Previous in this series: Apache Spark Configuration in Fabric

Next in this series: Lazy Evaluation in PySpark

---

Naveen Vuppula is a Senior Data Engineering Consultant and app developer based in Ontario, Canada. He writes about Python, SQL, AWS, Azure, and everything data engineering at DriveDataScience.com.