End-to-End Architecture Map

The complete Spark engine lifecycle on one responsive map — from raw code submission down to OneLake Parquet bytes. Click any node to open its technical inspector, or press Trace a query to walk the full execution sequence.

Spark Engine Internals

An interactive, factual walkthrough of how Apache Spark actually executes a job — from the driver's query plan down to the bytes an executor writes to disk — plus how Microsoft Fabric's Native Execution Engine and Efficient Scaledown change specific parts of that path. Twenty-nine sections covering the engine, Spark SQL and PySpark practice, engine choice, and the Fabric platform around them. Start with the console below: pick a workload, set the engine switches, press Run job, and click the nodes to see what each part actually does.

Cluster Architecture & The 1:1 Rule

Every Spark deployment — local, YARN, Kubernetes, or a Fabric Spark pool — shares the same three roles. What differs across platforms is only who launches the executors and how.

Deployment modes

The distinction that actually matters day to day is where the driver runs, not which cluster manager is in use:

• Client mode — the driver runs on the machine that submitted the job (your laptop, a notebook kernel). Convenient for interactive work; the job dies if that machine disconnects.
• Cluster mode — the driver itself runs inside the cluster, as just another managed process. Standard for production batch jobs; survives the submitting client disconnecting.

Fabric notebooks run the driver inside the Spark pool itself — architecturally closer to cluster mode, with the notebook UI acting as a thin client attached to a live remote session.

Fabric specifically: the 1:1 node→executor rule

This is a real structural difference from classic YARN Spark, and it changes how executor sizing works (see the worked comparison in Sec 15). In Fabric, one Spark pool node hosts exactly one executor — there's no "how many executors should I pack onto this node" decision the way there is on YARN. One node in the pool is always reserved for the driver; the rest become executors 1:1. The only exception is a single-node pool, where the driver and the sole executor share that one node's resources.

Job · Stage · Task Execution Model

Spark is lazy: transformations (select, filter, join) only build up a logical plan. Nothing executes until an action (write, collect, count) forces it. That action produces exactly one job.

Narrow vs. wide transformations

This distinction is the single most load-bearing concept in the entire execution model — it's the difference between "free" and "expensive":

• Narrow — map, filter, select, union. Each output partition depends on exactly one input partition. No data crosses the network; Spark can pipeline these into a single stage.
• Wide — groupBy, join, distinct, repartition. An output partition can depend on every input partition, because rows with the same key can start anywhere. This forces a shuffle (Sec 11) and a new stage.

DAGScheduler vs. TaskScheduler

Two schedulers, two different jobs, both living on the driver:

• DAGScheduler — stage-level. Walks the RDD/plan dependency graph backwards from the action, cuts it into stages at shuffle boundaries, and submits stages in dependency order (a stage can't start until every stage it depends on has finished writing its output).
• TaskScheduler — task-level. Takes one stage's worth of tasks and assigns them to executor cores, handling retries, speculative execution, and locality preference (run a task on the executor that already holds its input data, when possible).

Narrow vs. Wide Transformations: Comprehensive Matrix

Data Locality Levels: How TaskScheduler Places Work

TaskScheduler tries to achieve the tightest data locality possible, waiting a configurable delay before dropping to a looser level:

Catalyst Optimizer & Multi-Engine Execution Architecture

Every SQL query and DataFrame transformation in Apache Spark passes through the Catalyst Optimizer before execution. Catalyst is an extensible relational query compiler written in Scala, structured around tree transformations and rule batches. In Microsoft Fabric, Catalyst compiles queries into physical operator trees that dispatch to either Tungsten Whole-Stage Code Generation (JVM) or the Native Execution Engine (NEE / Meta Velox C++ SIMD), dynamically guided at runtime by Adaptive Query Execution (AQE).

The Four Catalyst Compilation Phases

Physical Join Strategy Selection Hierarchy

Catalyst uses a deterministic priority tree to select the physical join algorithm. Understanding this hierarchy allows you to predict whether Spark will broadcast, shuffle, or sort your data:

Interactive Query Progression Explorer (Simple to Complex)

Explore how Catalyst, Tungsten, and NEE compile and execute queries across increasing levels of complexity. Each level provides side-by-side runnable PySpark DataFrame code and Spark SQL query syntax, alongside the exact plan evolution at each stage:

from pyspark.sql import functions as F

# Simple filter + constant arithmetic + projection
df_simple = (
    spark.read.table("silver.orders")
    .filter(
        (F.col("status") == "COMPLETED") & 
        (F.col("amount") * (1 + 0.20) > 120)
    )
    .select(
        F.col("order_id"),
        F.col("cust_id"),
        (F.col("amount") * 1.20).alias("amount_with_tax")
    )
)
df_simple.explain(mode="formatted")

SELECT
  order_id,
  cust_id,
  amount * 1.20 AS amount_with_tax
FROM silver.orders
WHERE status = 'COMPLETED'
  AND amount * (1 + 0.20) > 120;

'Project [unresolvedalias('order_id), unresolvedalias('cust_id), ('amount * 1.20) AS amount_with_tax#0]
+- 'Filter (('status = COMPLETED) AND (('amount * (1 + 0.20)) > 120))
   +- 'UnresolvedRelation [silver, orders], [], false

Project [order_id#1: bigint, cust_id#2: bigint, (amount#3: double * 1.2) AS amount_with_tax#4: double]
+- Filter ((status#5: string = 'COMPLETED') AND ((amount#3: double * (1.0 + 0.2)) > 120.0))
   +- Relation silver.orders[order_id#1,cust_id#2,amount#3,order_date#4,status#5,notes#6] parquet

Project [order_id#1, cust_id#2, (amount#3 * 1.2) AS amount_with_tax#4]
+- Filter ((isnotnull(amount#3) AND isnotnull(status#5)) AND ((status#5 = 'COMPLETED') AND ((amount#3 * 1.2) > 120.0)))
   +- Relation silver.orders[order_id#1,cust_id#2,amount#3,status#5] parquet  // notes & order_date PRUNED

*(1) Project [order_id#1, cust_id#2, (amount#3 * 1.2) AS amount_with_tax#4]
+- *(1) Filter ((isnotnull(status#5) AND (status#5 = 'COMPLETED')) AND ((amount#3 * 1.2) > 120.0))
   +- *(1) FileScan parquet silver.orders[order_id#1,cust_id#2,amount#3,status#5]
         Batched: true, DataFilters: [isnotnull(status#5), (status#5 = 'COMPLETED')],
         PushedFilters: [IsNotNull(status), EqualTo(status,COMPLETED)]

VeloxColumnarToRowExec
+- ProjectExecTransformer [order_id#1, cust_id#2, (amount#3 * 1.2) AS amount_with_tax#4]
   +- FilterExecTransformer ((status#5 = 'COMPLETED') AND ((amount#3 * 1.2) > 120.0))
      +- NativeFileScanTransformer parquet silver.orders[order_id#1,cust_id#2,amount#3,status#5]

from pyspark.sql import functions as F

df_orders = spark.read.table("silver.orders") \
    .filter(F.col("order_date") >= "2026-01-01")

df_cust = spark.read.table("silver.customers") \
    .filter(F.col("tier") == "PLATINUM")

df_med = (
    df_orders.join(df_cust, "cust_id", "inner")
    .groupBy(
        df_cust.region,
        F.date_trunc("month", df_orders.order_date).alias("order_month")
    )
    .agg(
        F.count("order_id").alias("total_orders"),
        F.sum("amount").alias("total_revenue")
    )
    .filter(F.col("total_revenue") > 10000)
)
df_med.explain(mode="formatted")

SELECT
  c.region,
  date_trunc('month', o.order_date) AS order_month,
  COUNT(o.order_id) AS total_orders,
  SUM(o.amount) AS total_revenue
FROM silver.orders o
JOIN silver.customers c ON o.cust_id = c.cust_id
WHERE o.order_date >= '2026-01-01'
  AND c.tier = 'PLATINUM'
GROUP BY c.region, date_trunc('month', o.order_date)
HAVING SUM(o.amount) > 10000;

Filter (total_revenue#10 > 10000.0)
+- Aggregate [region#21, date_trunc('month', order_date#3)], [region#21, date_trunc('month', order_date#3) AS order_month#8, count(order_id#1) AS total_orders#9L, sum(amount#2) AS total_revenue#10]
   +- Join Inner, (cust_id#4 = cust_id#20)
      :- Filter (order_date#3 >= 2026-01-01)
      :  +- Relation silver.orders[order_id#1,amount#2,order_date#3,cust_id#4] parquet
      +- Filter (tier#22 = 'PLATINUM')
         +- Relation silver.customers[cust_id#20,region#21,tier#22] parquet

Filter (total_revenue#10 > 10000.0)
+- Aggregate [region#21, _date_trunc_month#25], [region#21, _date_trunc_month#25 AS order_month#8, count(order_id#1) AS total_orders#9L, sum(amount#2) AS total_revenue#10]
   +- Project [order_id#1, amount#2, region#21, date_trunc('month', order_date#3) AS _date_trunc_month#25]
      +- Join Inner, (cust_id#4 = cust_id#20)
         :- Relation silver.orders[order_id#1,amount#2,order_date#3,cust_id#4] parquet  // PartitionFilters: [order_date >= 2026-01-01]
         +- Filter (tier#22 = 'PLATINUM')
            +- Relation silver.customers[cust_id#20,region#21,tier#22] parquet   // PushedFilters: [EqualTo(tier,PLATINUM)]

AdaptiveSparkPlan isFinalPlan=false
+- *(3) Filter (total_revenue#10 > 10000.0)
   +- *(3) HashAggregate [region#21, _date_trunc_month#25], [sum(amount#2), count(order_id#1)]  // FINAL AGGREGATE
      +- AQEShuffleRead coalesced
         +- Exchange hashpartitioning(region#21, _date_trunc_month#25, 200)
            +- *(2) HashAggregate [region#21, _date_trunc_month#25], [partial_sum(amount#2), partial_count(order_id#1)] // PARTIAL AGGREGATE
               +- *(2) Project [order_id#1, amount#2, region#21, date_trunc('month', order_date#3) AS _date_trunc_month#25]
                  +- *(2) BroadcastHashJoin [cust_id#4], [cust_id#20], Inner, BuildRight
                     :- *(2) FileScan parquet silver.orders[order_id#1,amount#2,order_date#3,cust_id#4]
                     +- BroadcastExchange IdentityBroadcastMode
                        +- *(1) Filter (tier#22 = 'PLATINUM')
                           +- *(1) FileScan parquet silver.customers[cust_id#20,region#21,tier#22]

from pyspark.sql import Window, functions as F

# 1. Base enriched dataset
df_base = (
    spark.read.table("silver.orders")
    .join(spark.read.table("silver.customers"), "cust_id")
    .filter(F.col("status") == "DELIVERED")
)

# 2. Window ranking per customer segment
w_segment = Window.partitionBy("segment").orderBy(F.col("amount").desc())

df_complex = (
    df_base
    .withColumn("rank_in_segment", F.row_number().over(w_segment))
    .filter(F.col("rank_in_segment") <= 3)
    .select("segment", "rank_in_segment", "order_id", "cust_id", "amount")
)
df_complex.explain(mode="formatted")

WITH DeliveredOrders AS (
  SELECT
    o.order_id,
    o.cust_id,
    c.segment,
    o.amount,
    ROW_NUMBER() OVER (
      PARTITION BY c.segment 
      ORDER BY o.amount DESC
    ) AS rank_in_segment
  FROM silver.orders o
  JOIN silver.customers c ON o.cust_id = c.cust_id
  WHERE o.status = 'DELIVERED'
)
SELECT segment, rank_in_segment, order_id, cust_id, amount
FROM DeliveredOrders
WHERE rank_in_segment <= 3;

AdaptiveSparkPlan isFinalPlan=true
+- *(3) Filter (rank_in_segment#5 <= 3)
   +- *(3) Window [row_number() windowspecdefinition(segment#21, amount#2 DESC NULLS LAST, specifiedwindowframe(RowFrame, unboundedpreceding$, currentrow())) AS rank_in_segment#5], [segment#21], amount#2 DESC NULLS LAST], [segment#21]
      +- *(3) Sort [segment#21 ASC NULLS FIRST, amount#2 DESC NULLS LAST], false, 0
         +- AQEShuffleRead coalesced
            +- Exchange hashpartitioning(segment#21, 200)
               +- *(2) SortMergeJoin [cust_id#4], [cust_id#20], Inner (skew=true)
                  :- *(2) Sort [cust_id#4 ASC NULLS FIRST]
                  :  +- AQEShuffleRead (skewPartitionSplits: [P0_1, P0_2])
                  :     +- Exchange hashpartitioning(cust_id#4, 200)
                  :        +- *(1) Filter (status#3 = 'DELIVERED')
                  :           +- *(1) FileScan parquet silver.orders
                  +- *(2) Sort [cust_id#20 ASC NULLS FIRST]
                     +- Exchange hashpartitioning(cust_id#20, 200)
                        +- *(1) FileScan parquet silver.customers

VeloxColumnarToRowExec
+- FilterExecTransformer (rank_in_segment#5 <= 3)
   +- WindowExecTransformer [row_number() over segment]
      +- SortExecTransformer [segment#21, amount#2 DESC]
         +- ColumnarExchange hashpartitioning(segment#21)
            +- ShuffledHashJoinExecTransformer [cust_id#4], [cust_id#20]
               :- NativeFileScanTransformer parquet silver.orders
               +- NativeFileScanTransformer parquet silver.customers

Tungsten Engine & Whole-Stage Code Generation

Where Catalyst decides what to execute, Project Tungsten optimizes how execution occurs on physical CPU hardware:

1. Binary In-Memory Row Format (UnsafeRow)

Tungsten stores rows as raw, word-aligned byte arrays (UnsafeRow) outside the standard JVM heap. This eliminates Java object header bloat (16 bytes per object), memory pointer chasing, and garbage collection overhead.

2. Cache-Aware Memory Layout & Sorting

Operations like radix sorting and hash aggregation operate directly on 8-byte binary prefixes packed into contiguous cache lines. CPUs prefetch these lines sequentially into L1/L2 hardware caches, achieving near-zero cache miss penalties.

3. Whole-Stage Code Generation (WSCG)

Spark's Janino compiler collapses entire pipelines of operators into a single tight Java while loop. Variables are kept directly in CPU registers rather than written to memory between operator boundaries:

// Virtual function call per row per operator
for row in scan.next():
  if filter.eval(row):      // virtual dispatch
    out = project.eval(row)   // virtual dispatch
    emit(out)

// Single fused Java function per stage
while (scan.hasNext()) {
  UnsafeRow row = scan.next();
  if (row.getDouble(2) > 100.0) {
    emit(row.getString(0), row.getDouble(2) * 1.2);
  }
}

Adaptive Query Execution (AQE)

Catalyst's physical plan (Sec 04) is built from statistics known before the query runs — table metadata, maybe column stats. AQE, on by default since Spark 3.2, corrects that plan during execution, using real statistics from stages that have already run.

The Five Core AQE Features

AdaptiveSparkPlan isFinalPlan=true
+- AQEShuffleRead coalesced (200 -> 14 partitions, target 128MB)
   +- Exchange hashpartitioning(customer_id#4, 200)

// Driver Log: "AdaptiveSparkPlan: Plan changed from SortMergeJoin to BroadcastHashJoin"
+- *(2) BroadcastHashJoin [customer_id#4], [customer_id#20], Inner, BuildRight
   :- *(1) FileScan parquet silver.orders (Fact Table - ZERO SHUFFLE!)
   +- BroadcastExchange HashedRelationBroadcastMode (Materialized 4.2 MB)
      +- *(1) Filter (date#2 = '2026-08-17')
         +- *(1) FileScan parquet silver.dim_customers

+- *(2) SortMergeJoin [cust_id#4], [cust_id#20], Inner (skew=true)
   :- *(2) Sort [cust_id#4 ASC NULLS FIRST]
   :  +- AQEShuffleRead (skewPartitionSplits: [P0_1, P0_2, P0_3])
   :     +- Exchange hashpartitioning(cust_id#4, 200)
   +- *(2) Sort [cust_id#20 ASC NULLS FIRST]
      +- AQEShuffleRead (skewPartitionReplicated: [P0])
         +- Exchange hashpartitioning(cust_id#20, 200)

+- FileScan parquet gold.fact_sales
   PartitionFilters: [dynamicpruningexpression(date_key#1 IN subquery#25)]
   ReadSchema: struct<order_id:bigint, amount:double>

Native Execution Engine (NEE) & Gluten

NEE is Fabric Spark's vectorized C++ execution path — built on two open-source components: Velox (a C++ database acceleration library open-sourced by Meta) and Apache Gluten (incubating; a middle layer, originated by Intel, that translates Spark's physical plan into a form Velox can execute). General availability: Fabric Runtime 1.3.

Where it sits in the plan — and where it doesn't

NEE integrates after Catalyst's logical and physical optimization (Sec 04) — every existing optimization (predicate pushdown, column pruning, AQE, cost-based rewrites) still runs exactly as it would without NEE. What changes is only how a supported operator executes once the plan is already decided.

• Supported operators offload to the vectorized Velox path.
• Unsupported operators fall back to JVM/Tungsten execution (Sec 05) automatically, with a columnar↔row conversion at the boundary — a real cost, but scoped to just that operator.

Spark JVM vs Native Execution — Fallback Mechanics

NEE is not a switch you flip and forget. Gluten intercepts the physical plan, swaps supported nodes for Velox C++ kernels, and leaves everything else on the JVM — inserting a columnar↔row conversion at every boundary. That cost model has a consequence most teams discover the hard way: a plan that falls back mid-flight can be slower than running with NEE off entirely, because you pay conversion overhead without sustained native execution. Microsoft states the gain is greatest when queries avoid triggering fallback, and that the engine suits compute-intensive rather than I/O-bound work.

Fallback trigger taxonomy — four tiers

Three ways to detect fallback (use all three)

Interactive: fallback simulator

NEE on Runtime 2.0 vs 1.3 — Critical Differences

• NEE supports both Runtime 1.3 (Spark 3.5 / Delta 3.2) and Runtime 2.0 (Spark 4.1 / Delta 4.2) — it is not a 2.0-only feature.
• Runtime 2.0 broadens coverage: vectorized CSV parsing, Python and Scala UDF support, and complex types.
• Native Delta write acceleration extends native execution into the output path rather than read/transform only — relevant for write-heavy silver pipelines.
• The catch on 2.0: ANSI defaults to on and NEE falls back under ANSI, so an untouched Runtime 2.0 environment with NEE "enabled" may be running almost entirely on the JVM. This is the single most important thing to verify after migrating (N003/N007 catch it; nb_nee_fallback_analyzer measures it).

ANSI Mode × NEE Interaction Matrix

Rewrite Patterns — Removing Fallback Operators

Unified Memory Management & Off-Heap Arenas

Every executor is a JVM with a fixed heap (spark.executor.memory), carved up by Spark's UnifiedMemoryManager (the default since Spark 1.6) into regions that compete for the same pool. Drag the sliders — the bar below recomputes exactly the way a real executor's memory manager would.

What each region is for

Partitioning, Sizing & Parallelism

A partition is the unit of parallelism: one task processes exactly one partition. Too few partitions and cores sit idle; too many and per-task scheduling overhead dominates. Two separate settings control partition count depending on where in the plan you are.

The small-files problem

When average source file size is well under the partition target, Spark doesn't automatically launch one task per tiny file — spark.sql.files.openCostInBytes (default 4 MB) estimates the fixed cost of opening a file and packs several small files into one partition/task accordingly. But this only compensates at read time; it doesn't fix the underlying problem of driver-side file-listing overhead and Delta/Parquet metadata bloat from having millions of tiny files in storage. A compaction pass (e.g. Delta OPTIMIZE) upstream is the real fix.

Shuffle Internals, Spill & Sorters

A shuffle physically redistributes data across the cluster so that rows sharing a key end up on the same executor — required by every wide transformation (Sec 03). This is Spark's most expensive common operation, and the one most of the rest of this document (AQE, NEE, Efficient Scaledown) exists to make cheaper or safer.

The write path

Since Spark 2.0, all shuffle behaviour runs through SortShuffleManager, which selects one of three writers per task:

Whichever writer runs, the physical output per task is a data file (compressed, serialized records grouped by destination partition) plus an index file (byte offsets marking where each partition's slice starts/ends) — so a reduce task can seek directly to its range instead of scanning the whole file.

The read path

MapOutputTracker (driver-resident, cached on executors) holds the authoritative map from (shuffleId, partition) to the BlockManagerId that holds it. BlockManager (one per executor) serves the actual bytes over the network on request. A reduce task's BlockStoreShuffleReader looks up locations, fetches blocks in parallel, and returns a sorted iterator over the result.

Broadcast Hash Joins & Driver OOMs

The cheapest way to join two DataFrames is to avoid shuffling either of them. If one side is small enough to fit comfortably in memory, Spark can send a full copy to every executor (a broadcast) and perform the join locally, entirely avoiding a shuffle exchange for that operation.

Dynamic Allocation & Remote Shuffle Manager (RSM)

Full internals (sequence diagrams, config reference, storage/networking constraints) live in the companion document "Efficient Scaledown & the Remote Shuffle Manager — Internals Reference." This section is the condensed version, connecting it back to everything above.

Why elastic scaling and shuffle fight each other

Dynamic allocation (spark.dynamicAllocation.enabled) lets Spark release idle executors and request new ones as load changes. But an executor holding shuffle output (Sec 11) isn't safely idle even with zero running tasks — a not-yet-scheduled task may still need to fetch from it. Left alone, this either blocks scale-down (executor sits idle, billed, protecting data) or risks a costly stage recompute if the executor is removed anyway.

Spark's own fix: graceful decommissioning (3.1+)

Before an executor is torn down, if spark.storage.decommission.shuffleBlocks.enabled is on, Spark migrates its shuffle blocks to a peer executor first — or to configured fallback storage if no peer has room. Only once every block has a new home does the executor actually exit.

Fabric's four Efficient Scaledown capabilities

Configuration Reference & Scopes

Every Spark setting mentioned in this document, in one searchable table — default value, which section explains it, and whether it's safe to change on a running session or needs to be set before one starts (Sec 14 → this exact distinction is what broke the notebook utility during testing; see the companion notebook for the full story).

Interactive Config Advisor (Live Heuristics & Medallion Presets)

Heuristic calculation engine tailored for Microsoft Fabric Spark pools. Generates minimal delta configurations (omitting platform defaults to avoid pinning debt) partitioned by Medallion layer and architectural scope.

// Select advisor recommendations above to generate %%configure JSON

Fabric Compute Reference & Node Sizing

The concrete numbers behind every Fabric-specific recommendation above.

Node sizes (memory-optimized — fixed 8 GB RAM per vCore)

XLarge and XXLarge require a non-trial Fabric SKU. Source: Microsoft Learn, "Apache Spark compute for Data Engineering and Data Science."

Capacity units → Spark vCores

1 Capacity Unit (CU) = 2 Spark vCores. A pool's maximum size is bounded by the capacity SKU hosting it — not every node size is reachable on every SKU.

Starter pools default to Medium node size. Source: Microsoft Learn, "Configure starter pools."

Starter pool vs. custom pool

• Starter pool — pre-hydrated, ~5 second session start, fixed at Medium node size. You can still adjust autoscale bounds and dynamic executor allocation, but not the node size itself.
• Custom pool — you choose node size and full autoscale/dynamic-allocation configuration, at the cost of a normal (slower) session cold start.

Within a node: cores and memory are still adjustable

Choosing a node size fixes the executor's ceiling, not necessarily what you use. A Fabric environment's Compute pane lets you under-populate an executor's core count (e.g. run an 8-vCore node's executor at 4 cores, trading parallelism for per-task memory headroom) and choose among a fixed menu of executor-memory values for that node size — both are session-start settings (Sec 14), configured via the environment or a %%configure cell, not spark.conf.set() mid-session.

Spark 3.5 vs 4.x — Runtime 1.3 vs Runtime 2.0

Fabric currently offers two runtime generations. Runtime 1.3 (Apache Spark 3.5, Delta Lake 3.2) is GA and the recommended production runtime. Runtime 2.0 launched at FabCon Atlanta (March 2026) in Public Preview on Spark 4.0/Delta 4.0 and has since been upgraded in place to Spark 4.1 / Delta Lake 4.2 — as of mid-2026 it is GA: production-ready. Note it is not yet the default runtime, so you must select it explicitly at workspace or Environment level.

Interactive: what actually changed in Spark 4.x

Migration checklist: Runtime 1.3 → 2.0

• Re-publish every Environment that has libraries. The Python 3.11 → 3.13 jump means custom and public libraries must be re-added and Published, or jobs fail with "No module found" / "Class not found". Export the library list first.
• ANSI audit (above) — pre-test on Runtime 1.3 by setting spark.sql.ansi.enabled=true in a dev environment and running your suite before you ever switch runtimes.
• Scala/JVM code: Scala 2.12 → 2.13 and JDK 11 → 21 mean any custom JARs must be recompiled.
• Deprecated paths: WASB protocol → use ABFS; the old EventHubConnector → use the Kafka Spark connector; SparkR is deprecated in Spark 4.x.
• Config default drift from upstream Spark 4.x: ORC compression snappy → zstd, spark.sql.maxSinglePartitionBytes now 128m (was unlimited), speculation is less aggressive (multiplier 1.5 → 3, quantile 0.75 → 0.9), external shuffle service now removes shuffle files for deallocated executors by default.
• Delta 4.x-only table features (e.g. collations, VARIANT columns) are currently Spark-only — enabling them on a table breaks readability from other Fabric engines (SQL endpoint, older readers). Gate them behind interop review; deletion vectors and CDF are fine on both runtimes.

Feature Explorer — Engine Compatibility

Driven by the runtime selector in the top bar (currently: Runtime 1.3). Features are grouped by which engine ships them — SPARK ENGINE DELTA ENGINE FABRIC PLATFORM — because that is what actually gates availability: the Spark version and Delta version come from the runtime you pick (Sec 17); Fabric platform features mostly work on both. Cards that need Runtime 2.0 dim out when 1.3 is selected; switch the selector to see what upgrading unlocks.

Capacity Units, Smoothing, Bursting & Throttling

Every Fabric Spark job draws from a capacity (F-SKU). Understanding the CU model is what turns "the job randomly failed with HTTP 430" into a predictable, monitorable system.

The CU model in three equations

Smoothing — why a Spark spike doesn't instantly throttle you

Fabric classifies Spark operations as background operations, and background usage is smoothed over 24 hours: a burst of consumption is spread across the next day's capacity rather than billed against the minute it ran. This is why a heavy overnight batch doesn't immediately break morning interactive work — but it also means overuse accumulates silently as carryforward overage that must be burned down. Watch cumulative utilisation, not instantaneous.

Job admission: what HTTP 430 actually means

Fabric Spark uses optimistic job admission: a job is admitted if its minimum node count fits, then grows toward its max via autoscale if capacity allows. When even the minimum doesn't fit: interactive (notebook-submitted) jobs are rejected immediately with HTTP 430 — TooManyRequestsForCapacity, while scheduled/pipeline-triggered jobs enter a SKU-dependent queue and start when capacity frees. Practical consequences: set realistic pool minimums (a min of 1 node admits far more often than a min of 10), and put retry-with-backoff on any automation that submits interactively.

Emitting issues early — before users feel them

• Capacity Metrics App — the canonical view of utilisation, smoothing, carryforward overage and per-item CU attribution. Review the overage tab weekly, not just after incidents.
• Surge protection — set a background-rejection threshold on the capacity so runaway background jobs get refused before they push the capacity into the interactive-rejection band that hurts users.
• Activator / alerting on capacity metrics — alert at e.g. 80% smoothed utilisation and on any throttling event, rather than discovering 430s from users.
• Fabric Apache Spark diagnostic emitter — ship Spark driver/executor logs and metrics to Log Analytics, Azure Storage, or Event Hubs; alert on FetchFailedException, OOM patterns, and admission failures centrally (Sec 21).
• Autoscale Billing for Spark — moves Spark to serverless pay-as-you-go compute billed per job, off the capacity entirely: the right lever when Spark contends with Power BI/warehouse workloads on a shared capacity.

Where settings actually live — the configuration hierarchy

Every knob in this document belongs to exactly one of five layers. Knowing which layer owns a setting tells you where to change it and who it affects — and mirrors what you see in the Fabric portal (Workspace settings → Data Engineering/Science → Spark settings, with its Pool / Environment / Jobs / High concurrency tabs).

Interactive: capacity planner

CU Accounting & Per-Entity Chargeback

The Capacity Metrics App tells you which item consumed capacity. It does not tell you which entity in a metadata-driven framework did — one generic notebook running 40 entities appears as one notebook. Closing that gap is a logging decision, not a platform feature.

Extending etl_run_log

Where to write it

Troubleshooting UI, Skew & History Server

Everything Sec 01–Sec 13 explained becomes practical here: each internal mechanism leaves a specific fingerprint in the Spark UI, and reading those fingerprints is how you diagnose slow or failing jobs instead of guessing. In Fabric: Monitoring hub → your application → Spark UI / Spark history server, or from a notebook's run status bar → "View run details". Logs live under the same detail view (driver stdout/stderr, executor logs), and the in-notebook Spark Advisor surfaces common issues inline.

Interactive: symptom → diagnosis → fix

The monitoring detail view, tab by tab — and where Livy fits

Before Spark ever runs your code, Fabric routes the submission through Apache Livy, the REST session layer: notebook or Spark Job Definition → Livy session request → capacity admission (Sec 19) → container acquisition → Environment resolution (runtime version, libraries, Spark properties) → spark-submit → session idle, ready for statements. Each notebook session is a Livy session with an id and state machine (not_started → starting → idle ⇄ busy → dead). This matters diagnostically because a whole class of failures happens at the Livy layer, before any Spark job exists — capacity 430s, library resolution conflicts, un-published Environments, session timeouts — and those appear in the Livy log, not in Spark's UI.

Centralised log emission — catch it before the user does

The Fabric Apache Spark diagnostic emitter (configured per Environment) streams driver/executor logs and metrics to Azure Log Analytics, Azure Storage, or Event Hubs. Once logs land in Log Analytics, standing KQL alerts on OutOfMemoryError, FetchFailedException, spill metrics, and 430 admission failures turn every failure class on this page into a proactive signal instead of a user report. Pair with the emitted Spark event log (see "Observable artifacts", Sec 03) for full-fidelity postmortems.

Spark Logs REST API & Automated Diagnostics

The Spark UI is not actually confusing — it is five separate surfaces that look like one, and nobody tells you which one answers which question. This section maps them, then shows the part most people never discover: every one of them is a documented REST API, so "this cell is slow" can become a script instead of a click-through.

The five surfaces, and what each one actually is

The 6 Application Detail Tabs & Key Production Capabilities

The REST API layer — the same data, scriptable

Fabric exposes Spark application internals as a REST API that mirrors the OSS Spark History Server API contract exactly — same endpoints, same parameters, same JSON shapes for jobs, stages, tasks, executors, storage and streaming. On top of that, Fabric adds endpoints OSS Spark does not have: the Advisor and resource-usage graphs, as APIs.

Turning "this cell is slow" into a diagnosis — the workflow

1. Get the application id + Livy id of the run that felt slow (Recent runs, or captured at session start).
2. List jobs, sort by duration, and note the jobGroupId of the slowest one — that is your cell.
3. List that job's stages, sort by duration; pull the slowest stage's task list.
4. Compute max-vs-median task duration from the task list — a large gap is skew, not general slowness (Sec 21).
5. Check spill in the stage summary metrics — non-zero spill means the fix is partition sizing or executor shape (Sec 09), not more time.
6. Pull Advisor findings for that job/stage — often this step alone names the cause.
7. Cross-check resource usage for the same jobGroup window — confirms whether executors were memory-starved or just busy.

This is exactly what nb_fabric_log_diagnostics.ipynb automates — point it at an application id and it runs steps 2–7 and prints a ranked diagnosis instead of five tabs of manual cross-referencing.

Interactive: which surface answers your question

Log Analytics, KQL & Event Log Analysis

Two distinct things people conflate. The diagnostic emitter streams logs and metrics continuously to an external destination — good for alerting and fleet monitoring. Event logs are the per-application raw event stream the Spark UI is rendered from — good for deep post-mortem analysis of one run. You want both, for different reasons.

Minimum emitter configuration — Log Analytics

Set these as Spark properties on an Environment (not in a notebook — this is session-start scope). Five properties is the whole thing:

spark.synapse.diagnostic.emitters: MyLA
spark.synapse.diagnostic.emitter.MyLA.type: "AzureLogAnalytics"
spark.synapse.diagnostic.emitter.MyLA.categories: "Log,EventLog,Metrics"
spark.synapse.diagnostic.emitter.MyLA.workspaceId: <LOG_ANALYTICS_WORKSPACE_ID>
spark.synapse.diagnostic.emitter.MyLA.secret: <LOG_ANALYTICS_WORKSPACE_KEY>
# Use Key Vault instead of an inline secret (do this in production):
spark.synapse.diagnostic.emitter.MyLA.secret.keyVault: <AZURE_KEY_VAULT_URI>
spark.synapse.diagnostic.emitter.MyLA.secret.keyVault.secretName: <SECRET_NAME>
# REQUIRED when using the default/starter pool - without it the emitter never engages:
spark.fabric.pools.skipStarterPools: "true"

Event logs — the deeper source

The emitter's EventLog category and the per-application event log are the same underlying event stream. The difference is access: the emitter pushes it continuously to a destination you configured in advance; the REST API lets you pull the complete log for a specific past application, whether or not you had an emitter configured. For "this run was slow, why?", pulling the log is the direct route.

GET https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/notebooks|sparkJobDefinitions|lakehouses/{itemId}/livySessions/{livyId}/applications/{appId}/{attemptId}/logs
# Returns binary application/zip stream containing JSON-lines Spark event logs

Runnable PySpark client: Paste any portal Monitoring-hub URL or REST endpoint, acquire the token ambiently via notebookutils, and parse the raw event log stream directly in memory:

import io, json, zipfile, requests, notebookutils
from spark_eventlog_analyzer import resolve_spark_rest_url

# 1. Accepts ANY portal monitoring URL or REST endpoint
portal_or_api_url = "https://app.fabric.microsoft.com/groups/0000.../sparkapplications/application_1741176604085_0001?workspaceId=...&livyId=...&itemId=..."
rest_url = resolve_spark_rest_url(portal_or_api_url, endpoint="logs")

# 2. Acquire ambient Entra ID Bearer token in Fabric Notebook
token = notebookutils.credentials.getToken("pbi")
resp = requests.get(rest_url, headers={"Authorization": f"Bearer {token}"})

# 3. Stream & Unzip in-memory (zero temp files written to disk)
events = []
if resp.status_code == 200:
    with zipfile.ZipFile(io.BytesIO(resp.content)) as z:
        for fname in z.namelist():
            if not fname.endswith("/"):
                with z.open(fname) as zf:
                    events.extend([json.loads(line) for line in zf if line.strip()])
    print(f"Successfully parsed {len(events)} events from Fabric REST API.")

What the event log gives you that the UI does not

Delta Table Optimization — V-Order, Liquid, DV, CDF

Storage-layout decisions have as much performance impact as any Spark setting. This section maps every Delta/Fabric optimization to when to turn it on, per medallion layer and per runtime.

How CDF works internally — and when it drives increments automatically

Without CDF the transaction log already lets you diff two versions, but only at file granularity: you can see which files were added and removed. For appends that is enough to derive inserted rows. For an UPDATE or MERGE it is not — a rewritten file contains changed and untouched rows, and nothing in the log tells you which is which.

Enabling delta.enableChangeDataFeed closes that gap: operations that modify existing rows write additional CDC files into a _change_data/ directory, carrying a _change_type of insert, update_preimage, update_postimage or delete, plus _commit_version and _commit_timestamp. Two efficiency details matter in practice:

• Pure appends write no CDC files. Insert-only changes are derived from the files added in the log, so append-heavy bronze carries almost no CDF overhead — the write amplification lands on updates, deletes and merges.
• Maintenance emits nothing. OPTIMIZE, compaction and VACUUM are not logically data-changing, so they produce no change rows. Your incremental consumer will not see phantom churn after a compaction job (proven in the companion notebook).

Pattern: CDF deltas without MLV, watermarked in Fabric SQL Database

When you need control MLV does not give you — custom batching, bespoke error handling, or a PySpark transform — CDF plus a watermark table is the durable equivalent. The watermark lives in Fabric SQL Database alongside the rest of the metadata (etl_cdf_watermark: entity, source table, last_version, run id, timestamp), upserted forward-only so concurrent runs can never rewind it.

changes = (spark.read.format("delta")
.option("readChangeFeed", "true")
.option("startingVersion", 0) # every version ever
.load(SILVER))
changes = changes.filter(
changes._commit_version > last_version) # filtered AFTER reading

changes = (spark.read.format("delta")
.option("readChangeFeed", "true")
.option("startingVersion", last_version + 1) # exclusive of applied
.option("endingVersion", target_version) # bounded batch
.load(SILVER)
.where(F.col("_change_type") != "update_preimage")
.where("status = 'complete'") # pushes into the scan
.select("order_id","order_date","amount","status",
"_change_type","_commit_version")) # narrow ReadSchema

# one MERGE per change row - a key updated 3 times
# is written 3 times, and ordering bugs are silent
for batch in changes.collect():
apply(batch)

w = Window.partitionBy("order_id").orderBy(
F.col("_commit_version").desc())
net = changes.withColumn("rn", F.row_number().over(w)) \
             .where("rn = 1").drop("rn")
# second pushdown: only the grains this batch touched
touched = [r.order_date for r in
net.select("order_date").distinct().collect()]
recomputed = (spark.read.format("delta").load(SILVER)
.where(F.col("order_date").isin(touched)
& (F.col("status") == "complete"))
.groupBy("order_date")
.agg(F.sum("amount").alias("revenue")))
target_tbl.alias("t").merge(recomputed.alias("s"),
"t.order_date = s.order_date") \
  .whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()
set_watermark(entity, SILVER, target_version, net.count())

earliest = min(h.version for h in
spark.sql(f"DESCRIBE HISTORY delta.`{path}`").collect())
if last_version + 1 < earliest:
raise LookupError(f"watermark {last_version} predates earliest retained "
f"version {earliest} - CDF aged out; full refresh required")

Why your MLV keeps doing a full refresh — and the programmatic equivalent

MLV's Optimal Refresh picks between skip, incremental and full based on what changed and what the definition can express — and it silently downgrades to full rather than failing, which is convenient until it becomes invisible. Five independent conditions force full refresh; only one of them is "unsupported SQL constructs":

Diagnose which condition is firing

-- 1. CDF on every source? (run per source table)
SHOW TBLPROPERTIES silver.orders ('delta.enableChangeDataFeed');
-- 2. Deletes/updates since the last MLV refresh? (the usual culprit)
DESCRIBE HISTORY silver.orders
-- inspect `operation` for UPDATE / DELETE / MERGE since the MLV's last refresh timestamp

# 3. Was the last refresh actually incremental? Check MLV refresh history/lineage in the portal,
# or query the refresh metadata if exposed in your Fabric version - full vs incremental is recorded
# per refresh cycle.

The same pattern in pure Spark SQL

Everything above was PySpark. Here is the identical logic as %%sql — useful in its own right since Sec 30 covers Spark SQL as a first-class surface, and directly relevant here because PySpark-defined MLVs always full-refresh (previous subsection) — a pure-SQL transform is what stays eligible for MLV's optimal refresh in the first place, and is also what you fall back to writing by hand when MLV's other constraints block it.

wm = spark.sql(
"SELECT last_version FROM ctl.cdf_watermark WHERE entity_name = :e",
args={"e": "silver_orders_to_gold"}).collect()[0][0]
target = spark.sql(
"SELECT max(version) FROM (DESCRIBE HISTORY silver_orders)"
).collect()[0][0]

CREATE OR REPLACE TEMP VIEW v_net_changes AS
SELECT order_id, order_date, amount, status
FROM (
SELECT *, ROW_NUMBER() OVER (
PARTITION BY order_id ORDER BY _commit_version DESC) AS rn
FROM table_changes('silver_orders', :wm + 1, :tgt)
WHERE _change_type != 'update_preimage' AND status = 'complete'
) WHERE rn = 1

BEGIN
DECLARE wm BIGINT;
DECLARE tgt BIGINT;
SET VAR wm = (SELECT last_version FROM ctl.cdf_watermark
WHERE entity_name = 'silver_orders_to_gold');
SET VAR tgt = (SELECT max(version) FROM (DESCRIBE HISTORY silver_orders));
CREATE OR REPLACE TEMP VIEW v_net_changes AS
SELECT order_id, order_date, amount, status
FROM table_changes('silver_orders', wm + 1, tgt)
WHERE _change_type != 'update_preimage' AND status = 'complete'
QUALIFY ROW_NUMBER() OVER (
PARTITION BY order_id ORDER BY _commit_version DESC) = 1;
END

The rest of the pipeline — apply and advance, in SQL either way

-- Recompute only the grains this batch touched (predicate pushes into the source scan)
CREATE OR REPLACE TEMP VIEW v_recomputed_gold AS
SELECT s.order_date, SUM(s.amount) AS revenue, COUNT(*) AS orders
FROM silver_orders s
WHERE s.status = 'complete'
AND s.order_date IN (SELECT DISTINCT order_date FROM v_net_changes)
GROUP BY s.order_date;
-- Apply as one MERGE, full business key
MERGE INTO gold_daily t
USING v_recomputed_gold s
ON t.order_date = s.order_date
WHEN MATCHED THEN UPDATE SET t.revenue = s.revenue, t.orders = s.orders
WHEN NOT MATCHED THEN INSERT (order_date, revenue, orders)
VALUES (s.order_date, s.revenue, s.orders);
-- Advance the watermark - forward-only, so concurrent/retried runs can never rewind it
MERGE INTO ctl.cdf_watermark t
USING (SELECT 'silver_orders_to_gold' AS entity_name, {target} AS new_version) s
ON t.entity_name = s.entity_name
WHEN MATCHED AND s.new_version > t.last_version
THEN UPDATE SET last_version = s.new_version, updated_at = current_timestamp();

On 3.5, {target} is a bound parameter like the earlier queries; on 4.x it is literally the tgt session variable, referenced by name with no substitution at all. ctl.cdf_watermark is a plain Delta control table here so the whole example stays self-contained and portable — the equivalent using Fabric SQL Database as the control store follows the pyodbc pattern in Sec 36; only the watermark read/write changes, the CDF logic above is identical either way.

Interactive: table optimization advisor

Column Statistics, Data Skipping & SQL-to-Spark Tuning Masterclass

In traditional relational databases (SQL Server, Oracle, Snowflake), 90% of performance engineering revolves around index seeks, query plan operators, updating statistics, and query decomposition. Over OneLake Delta tables in Microsoft Fabric, achieving sub-second query performance and 10× compute cost reductions requires mastering the 3 tiers of lakehouse statistics, eliminating Delta Transaction Log skipping blind spots, diagnosing Native Execution Engine (NEE / Velox) fallbacks from Spark logs, and systematically refactoring legacy SQL into high-throughput distributed Spark SQL patterns.

1. The 32-Column Data Skipping Trap in the Delta Transaction Log

A critical distinction in Lakehouse architecture: The 32-column ceiling applies strictly to the Delta Transaction Log (_delta_log), NOT to the underlying Parquet files!

By default, Delta Lake writers only collect and serialize min/max statistics in the transaction log for the first 32 columns of a table schema (delta.dataSkippingNumIndexedCols = 32). If your wide fact table has 60 columns and your queries filter on column 35 (e.g. store_region, tenant_id, event_status, or is_active), Delta cannot skip a single Parquet file at the driver level. The driver assumes all files could contain matching records and schedules tasks to scan every file in OneLake!

-- Wide table with 50 columns
CREATE TABLE silver.fact_telemetry (
  device_id STRING, -- col 1 (indexed in _delta_log)
  -- ... cols 2 to 32 (indexed in _delta_log) ...
  col32_meta STRING, -- col 32 (indexed in _delta_log)
  tenant_id STRING,  -- col 33 (NOT in _delta_log! min/max omitted)
  status STRING     -- col 34 (NOT in _delta_log! min/max omitted)
) USING DELTA;

-- 💥 SILENT FULL TABLE SCAN AT DRIVER LEVEL:
-- Driver reads ALL 10,000 files because tenant_id stats are missing in _delta_log!
SELECT * FROM silver.fact_telemetry WHERE tenant_id = 'TENANT_99';

-- OPTION A: Target only high-cardinality filter columns (Best Practice)
ALTER TABLE silver.fact_telemetry SET TBLPROPERTIES (
  'delta.dataSkippingStatsColumns' = 'device_id, tenant_id, status, event_timestamp'
);

-- OPTION B: Raise the global indexed column ceiling
ALTER TABLE silver.fact_telemetry SET TBLPROPERTIES (
  'delta.dataSkippingNumIndexedCols' = '64'
);

-- Backfill stats for existing historical Parquet files:
ANALYZE TABLE silver.fact_telemetry COMPUTE DELTA STATISTICS;

2. How to Genuinely Utilise Spark Logs & Metrics to Diagnose Gaps

When a Spark job is slow or consuming excessive Capacity Units (CUs) in Fabric, do not guess. Inspect the Fabric Monitoring Hub, Spark History Server SQL Graph, and Driver Logs (stderr) for these quantifiable physical signatures:

3. SQL-to-Spark Query Rewriting Masterclass (5 Core Production Patterns)

Migrating traditional T-SQL / Oracle scripts directly into Spark SQL without refactoring frequently leads to 10× slower execution. Here are the 5 architectural query refactorings every data engineer must apply:

Pattern A: Eliminating Multi-Scan CTE Traps (The Recomputation Vulnerability)

In SQL Server and Oracle, CTEs (WITH cte AS (...)) are often materialized or referenced efficiently. In Apache Spark, CTEs and DataFrames are lazily evaluated abstract syntax trees (ASTs). If a CTE is referenced 3 times in downstream joins or unions, Spark parses and executes the entire scan, filter, and aggregation 3 separate times from OneLake!

WITH AggregatedSales AS (
  -- Heavy scan over 500M rows with group by
  SELECT cust_id, SUM(amount) AS total_spend, COUNT(*) AS orders
  FROM silver.fact_orders
  GROUP BY cust_id
)
-- 💥 DISASTER: AggregatedSales is re-scanned and re-aggregated 3 separate times!
SELECT c.cust_name, a1.total_spend
FROM dim_customer c
JOIN AggregatedSales a1 ON c.cust_id = a1.cust_id
LEFT JOIN AggregatedSales a2 ON c.parent_cust_id = a2.cust_id;

# 1. Compute heavy aggregation ONCE and break Catalyst lineage graph
df_agg = (spark.table("silver.fact_orders")
          .groupBy("cust_id")
          .agg(F.sum("amount").alias("total_spend"), F.count("*").alias("orders"))
          .localCheckpoint(eager=True)) # Cut DAG & persist in executor NVMe cache

df_agg.createOrReplaceTempView("agg_sales")

# 2. Downstream query reads pre-materialized buffer from NVMe
spark.sql("""
  SELECT c.cust_name, a1.total_spend
  FROM dim_customer c
  JOIN agg_sales a1 ON c.cust_id = a1.cust_id
  LEFT JOIN agg_sales a2 ON c.parent_cust_id = a2.cust_id
""")

Pattern B: Replacing Python UDFs with Native Velox Vectorized Expressions

Standard Python UDFs (@udf) require serializing rows from JVM off-heap memory through a local Unix socket to a spawned Python worker daemon. This introduces severe serialization overhead, completely breaks Whole-Stage CodeGen, and forces the Native Execution Engine (Velox) to fall back to the slow Java VM.

from pyspark.sql.functions import udf

# 💥 Slow: 1 IPC serialization roundtrip PER ROW
@udf("string")
def clean_phone(phone):
    if not phone: return None
    return "".join([c for c in phone if c.isdigit()])

df.withColumn("clean_phone", clean_phone("raw_phone"))

from pyspark.sql import functions as F

# ✓ 25× Faster: Executes directly in Velox C++ SIMD registers with ZERO serialization!
df_clean = df.withColumn("clean_phone", F.regexp_replace(F.col("raw_phone"), r"\D", ""))

# Explain plan verification receipt:
# +- ProjectExecTransformer [regexp_replace(raw_phone#1, \D, ) AS clean_phone#2]

Pattern C: Resolving Skew Joins via AQE Heuristics & Key Salting

When a join key contains extreme skew (e.g. 50,000,000 rows with cust_id = NULL or a default placeholder ID), a single reducer task receives gigabytes of data and runs for 45 minutes while the remaining 199 tasks finish in 2 seconds.

-- Step 1: Add a random salt (0-9) to the skewed large table
CREATE OR REPLACE TEMP VIEW v_orders_salted AS
SELECT *, CONCAT(cust_id, '_', CAST(FLOOR(RAND() * 10) AS INT)) AS salted_join_key
FROM silver.fact_orders;

-- Step 2: Explode the small dimension table 10× to match all possible salt keys
CREATE OR REPLACE TEMP VIEW v_dim_customer_replicated AS
SELECT c.*, CONCAT(c.cust_id, '_', salt_id) AS salted_join_key
FROM dim_customer c
LATERAL VIEW EXPLODE(ARRAY(0,1,2,3,4,5,6,7,8,9)) t AS salt_id;

-- Step 3: Join on salted key -> Skew is perfectly distributed across 10 parallel tasks!
SELECT o.order_id, c.cust_name, o.amount
FROM v_orders_salted o
JOIN v_dim_customer_replicated c ON o.salted_join_key = c.salted_join_key;

Pattern D: Single-Pass Window Functions vs. Multi-Scan Self-Joins (Top-N per Group)

A classic SQL pattern for finding the most recent record per entity is joining a table back to a grouped subquery on MAX(date). In Spark, this causes 2 full table scans, 2 shuffles, and an expensive SortMergeJoin. A single-pass window function cuts I/O by 50% and executes via native Velox WindowExecTransformer.

-- 💥 Reads fact_orders TWICE + performs SortMergeJoin
SELECT o.*
FROM silver.fact_orders o
JOIN (
  SELECT cust_id, MAX(order_date) AS max_dt
  FROM silver.fact_orders
  GROUP BY cust_id
) m ON o.cust_id = m.cust_id AND o.order_date = m.max_dt;

-- ✓ Scans OneLake exactly ONCE with native Velox WindowExecTransformer
WITH RankedOrders AS (
  SELECT *, 
         ROW_NUMBER() OVER (PARTITION BY cust_id ORDER BY order_date DESC) AS rnk
  FROM silver.fact_orders
)
SELECT * EXCEPT(rnk)
FROM RankedOrders
WHERE rnk = 1;

4. Production Diagnostic Workbook (Run Directly in Fabric Notebook)

Use this automated diagnostic script to audit table skipping health, file fragmentation, deletion vector overhead, and column stats coverage across your Lakehouse:

from pyspark.sql import functions as F

def audit_delta_table_health(table_name):
    """Comprehensive health check for Data Skipping, Small Files & Deletion Vectors"""
    print(f"=== AUDITING TABLE: {table_name} ===")
    
    # 1. Inspect Physical File Sizing & Distribution
    df_files = spark.sql(f"SELECT * FROM {table_name}.`$files`")
    file_metrics = df_files.select(
        F.count("*").alias("total_files"),
        F.round(F.sum("size_in_bytes") / (1024**3), 2).alias("total_gb"),
        F.round(F.avg("size_in_bytes") / (1024**2), 2).alias("avg_file_mb"),
        F.sum(F.when(F.col("size_in_bytes") < 32*1024*1024, 1).otherwise(0)).alias("small_files_lt_32mb"),
        F.sum(F.when(F.col("deletionVector").isNotNull(), 1).otherwise(0)).alias("files_with_dvs")
    ).collect()[0]
    
    print(f"  • Total Parquet Files: {file_metrics['total_files']:,}")
    print(f"  • Total Table Size: {file_metrics['total_gb']} GB")
    print(f"  • Average File Size: {file_metrics['avg_file_mb']} MB (Target: 400-500 MB)")
    print(f"  • Fragmented Small Files (<32MB): {file_metrics['small_files_lt_32mb']:,}")
    print(f"  • Files with Deletion Vectors: {file_metrics['files_with_dvs']:,}")
    
    # 2. Check Table Properties for Data Skipping Indexing in _delta_log
    df_props = spark.sql(f"SHOW TBLPROPERTIES {table_name}").collect()
    props_map = {row['key']: row['value'] for row in df_props}
    stats_cols = props_map.get('delta.dataSkippingStatsColumns', '[NOT SET - Defaults to first 32 cols in _delta_log]')
    print(f"  • Delta Log Indexed Columns: {stats_cols}")
    
    # Actionable Recommendations
    if file_metrics['small_files_lt_32mb'] > 100:
        print(f"  🚨 RECOMMENDATION: Run 'OPTIMIZE {table_name}' to compact small files.")
    if file_metrics['files_with_dvs'] > (file_metrics['total_files'] * 0.25):
        print(f"  🚨 RECOMMENDATION: Run 'REORG TABLE {table_name} APPLY (PURGE)' to purge soft deletion vectors.")

# Run audit across lakehouse tables:
audit_delta_table_health("silver.orders")

Conditional Lakehouse Table Optimization & Health Checks Framework

Running blind, fixed-schedule OPTIMIZE commands across hundreds of Lakehouse tables burns massive Microsoft Fabric Capacity Units (CUs) on tables that are already healthy. The Microsoft Fabric "Check-Then-Act" Conditional Optimization Pattern uses the built-in sys.sp_get_table_health_metrics stored procedure on the SQL Analytics Endpoint to inspect physical table health first, triggering Spark OPTIMIZE and REORG only when anomalies are detected.

1. Potential Anomaly Types Decoded

The sys.sp_get_table_health_metrics procedure analyzes the underlying Delta log and file storage, classifying table health into structured anomaly codes:

2. Reusable Multi-Table Automated Maintenance Suite (PySpark)

For automated, catalog-wide maintenance across all tables in a Fabric Lakehouse, use this production PySpark workbook. It inspects table health metrics dynamically and dispatches maintenance tasks only where needed:

from pyspark.sql import functions as F

def optimize_lakehouse_tables_conditionally(database_name="dbo", small_file_threshold_mb=32, dv_ratio_threshold=0.20):
    """
    Scans all Delta tables in the database, audits health metrics, 
    and executes OPTIMIZE / REORG only when degradation is detected.
    """
    tables = [row.tableName for row in spark.sql(f"SHOW TABLES IN {database_name}").collect()]
    print(f"Auditing {len(tables)} tables in database '{database_name}'...")
    
    for tbl in tables:
        full_table = f"{database_name}.{tbl}"
        try:
            # 1. Read Delta $files metadata
            df_files = spark.sql(f"SELECT * FROM {full_table}.`$files`")
            metrics = df_files.select(
                F.count("*").alias("total_files"),
                F.round(F.avg("size_in_bytes") / (1024**2), 2).alias("avg_file_mb"),
                F.sum(F.when(F.col("size_in_bytes") < small_file_threshold_mb*1024*1024, 1).otherwise(0)).alias("small_files"),
                F.sum(F.when(F.col("deletionVector").isNotNull(), 1).otherwise(0)).alias("files_with_dv")
            ).collect()[0]
            
            total_files = metrics["total_files"]
            small_files = metrics["small_files"]
            dv_files = metrics["files_with_dv"]
            
            needs_optimize = (small_files > 50) or (total_files > 200 and metrics["avg_file_mb"] < 100)
            needs_purge = (total_files > 0) and ((dv_files / total_files) >= dv_ratio_threshold)
            
            if needs_optimize or needs_purge:
                print(f"⚡ MAINTAINING {full_table}: Total={total_files}, Small={small_files}, DV={dv_files}")
                if needs_purge:
                    print(f"  -> Purging Deletion Vectors on {full_table}...")
                    spark.sql(f"REORG TABLE {full_table} APPLY (PURGE)")
                if needs_optimize:
                    print(f"  -> Running V-Order Compaction on {full_table}...")
                    spark.sql(f"OPTIMIZE {full_table}")
            else:
                print(f"✓ SKIPPING {full_table} (Healthy: avg {metrics['avg_file_mb']} MB, 0 DV bloat)")
        except Exception as e:
            print(f"⚠️ Could not audit {full_table}: {str(e)}")

# Run automated conditional maintenance across Lakehouse:
optimize_lakehouse_tables_conditionally("dbo")

Lakehouse Data Modeling Masterclass: Data Vault 2.0 (Silver) & Kimball Dimensional (Gold) Side-by-Side

Modern enterprise Lakehouses face a classic architectural conflict: Data integration requires resilience against schema drift, multi-source ingestion concurrency, and complete historical auditability, while BI and analytics require fast, intuitive star schemas optimized for Power BI Direct Lake mode. Here is how Data Vault 2.0 (in the Silver Layer) and Kimball Dimensional Modeling (in the Gold Layer) sit side-by-side seamlessly over OneLake Delta tables.

1. How Data Vault 2.0 & Kimball Dimensional Sit Side-by-Side

2. Silver Hubs & Links: Physical Delta Tables vs. Materialized Lakehouse Views (MLVs) as Data Scales

A critical architectural decision in Microsoft Fabric: Should Hubs and Links be stored as physical Delta tables or defined as Materialized Lakehouse Views (MLVs)?

3. Why Gold Star Schema Surrogate Keys MUST Be BIGINT (Not String UUIDs)

In Power BI Direct Lake mode and Polaris Distributed Query Execution, joining Fact and Dimension tables on 32-character MD5 / SHA-256 strings is a fatal performance anti-pattern:

4. Production Implementation: Silver Data Vault to High-Performance Gold Star Schema

Here is the production-grade Spark SQL pattern generating Silver Data Vault Delta tables and transforming them into an optimal Gold Star Schema with `BIGINT` surrogate keys, V-Order enabled, and Liquid Clustering:

-- 1. SILVER HUB: Physical Delta table with SHA-256 Hash Key for Ingestion Concurrency
CREATE TABLE IF NOT EXISTS silver.hub_customer (
    customer_hk   STRING NOT NULL, -- SHA-256(customer_id) for multi-source ingest
    customer_id   STRING NOT NULL,
    load_dts      TIMESTAMP NOT NULL,
    record_source STRING NOT NULL
) USING DELTA
CLUSTER BY (customer_hk)
TBLPROPERTIES (
    'delta.enableChangeDataFeed' = 'true',
    'delta.enableDeletionVectors' = 'true'
);

-- 2. SILVER SATELLITE: Attribute History with HashDiffs
CREATE TABLE IF NOT EXISTS silver.sat_customer_profile (
    customer_hk    STRING NOT NULL,
    load_dts       TIMESTAMP NOT NULL,
    hash_diff      STRING NOT NULL, -- SHA-256(name + email + tier + city)
    customer_name  STRING,
    customer_email STRING,
    customer_tier  STRING,
    city           STRING,
    record_source  STRING NOT NULL
) USING DELTA
CLUSTER BY (customer_hk);

-- 3. GOLD CONFORMED DIMENSION (SCD Type 2): Optimal for Power BI Direct Lake
CREATE OR REPLACE TABLE gold.dim_customer
USING DELTA
CLUSTER BY (customer_id, is_current)
TBLPROPERTIES (
    'delta.parquet.vorder.enabled' = 'true', -- Mandatory for Direct Lake
    'delta.dataSkippingStatsColumns' = 'customer_sk, customer_id, customer_tier, is_current'
) AS
WITH CustomerHistory AS (
    SELECT 
        h.customer_id,
        s.customer_name,
        s.customer_email,
        s.customer_tier,
        s.city,
        s.load_dts AS valid_from,
        LEAD(s.load_dts) OVER (PARTITION BY h.customer_hk ORDER BY s.load_dts ASC) AS valid_to
    FROM silver.hub_customer h
    JOIN silver.sat_customer_profile s ON h.customer_hk = s.customer_hk
)
SELECT 
    -- ✓ FASTEST DESIGN: 64-bit BIGINT hash surrogate key (8 bytes vs 32-byte string)
    CAST(xxhash64(CONCAT(customer_id, '_', CAST(valid_from AS STRING))) AS BIGINT) AS customer_sk,
    customer_id,
    customer_name,
    customer_email,
    customer_tier,
    city,
    valid_from,
    COALESCE(valid_to, TIMESTAMP'9999-12-31 23:59:59') AS valid_to,
    CASE WHEN valid_to IS NULL THEN 1 ELSE 0 END AS is_current
FROM CustomerHistory;

-- 4. GOLD FACT TABLE: Joining on BIGINT surrogate keys for maximum throughput
CREATE OR REPLACE TABLE gold.fact_orders
USING DELTA
CLUSTER BY (order_date_id, customer_sk)
TBLPROPERTIES ('delta.parquet.vorder.enabled' = 'true') AS
SELECT 
    o.order_id,
    CAST(DATE_FORMAT(o.order_date, 'yyyyMMdd') AS INT) AS order_date_id,
    c.customer_sk, -- BIGINT foreign key
    o.order_amount,
    o.tax_amount,
    o.discount_amount
FROM silver.fact_orders_raw o
JOIN gold.dim_customer c 
  ON o.customer_id = c.customer_id 
 AND o.order_date >= c.valid_from 
 AND o.order_date < c.valid_to;