Azure Fabric

Microsoft Fabric

Microsoft Fabric is a unified SaaS analytics platform that combines OneLake storage, lakehouses, data warehouses, data engineering, data science, real-time analytics, and Power BI into a single licensed product billed through capacity units (CUs).

Architecture Overview

Fabric Tenant
└── Capacity (F SKU — F2 to F2048)
    └── Workspace(s)
        ├── Lakehouse       → Delta tables + Files section + SQL analytics endpoint
        ├── Warehouse       → T-SQL warehouse with no physical indexes needed
        ├── Notebook        → Spark compute (PySpark, Scala, SparkSQL, R)
        ├── Spark Job Def   → Headless Spark batch jobs (jar/py)
        ├── Environment     → Library + Spark config management
        ├── Dataflow Gen2   → Power Query M / low-code ETL
        ├── Pipeline        → ADF-compatible orchestration
        ├── KQL Database    → Real-Time Intelligence
        ├── Eventstream     → Real-time event ingestion
        ├── Eventhouse      → KQL database + eventstream bundle
        ├── ML Model        → MLflow-registered model
        ├── ML Experiment   → MLflow experiment tracking
        ├── Semantic Model  → Power BI dataset (Direct Lake capable)
        └── Report          → Power BI report

OneLake is the single logical data lake underpinning everything — one per tenant, structured as onelake://<workspace>/<item>/....

Capacities & Licensing

SKUs

• Trial capacity is free for 60 days (~F64 equivalent); one per tenant per year.
• CUs are shared across all workspaces assigned to the capacity.
• Capacity can be paused (billing stops, items still exist).
• Bursting: short spikes can exceed CU limit; sustained overuse causes throttling then rejection.
• Spark CU consumption: 1 Spark vCore ≈ 0.5 CU/hour. An F64 capacity can run up to 256 concurrent Spark vCores.

Manage via Portal or CLI

# List capacities (Fabric REST API — no az fabric CLI yet for most operations)
curl -sX GET \
  "https://api.fabric.microsoft.com/v1/capacities" \
  -H "Authorization: Bearer $FABRIC_TOKEN" | jq '.value[] | {id, displayName, sku, state}'

# Pause / resume a capacity (requires admin or capacity admin role)
curl -sX POST \
  "https://api.fabric.microsoft.com/v1/capacities/<capacityId>/suspend" \
  -H "Authorization: Bearer $FABRIC_TOKEN"

curl -sX POST \
  "https://api.fabric.microsoft.com/v1/capacities/<capacityId>/resume" \
  -H "Authorization: Bearer $FABRIC_TOKEN"

OneLake

OneLake is ADFS Gen2-compatible. Access paths:

ABFS (Azure Blob Filesystem):
  abfss://<workspace-name>@onelake.dfs.fabric.microsoft.com/<lakehouse-name>.Lakehouse/Tables/
  abfss://<workspace-name>@onelake.dfs.fabric.microsoft.com/<lakehouse-name>.Lakehouse/Files/

OneLake path (internal reference):
  onelake://<workspace-name>/<lakehouse-name>.Lakehouse/Tables/<table-name>/

OneLake File Explorer

• Windows desktop app; mounts OneLake as a drive (O:\ or similar).
• Real-time sync for Files section; Tables are read via SQL endpoint.

Shortcuts

Shortcuts create virtual pointers to external or internal data without copying it.

# Create a shortcut via Fabric REST API
import requests, json

payload = {
    "name": "my_s3_shortcut",
    "path": "Tables",          # or "Files"
    "target": {
        "type": "AmazonS3",
        "amazonS3": {
            "location": "https://s3.amazonaws.com/my-bucket/path/",
            "subpath": "delta_tables/orders"
        }
    }
}
resp = requests.post(
    f"https://api.fabric.microsoft.com/v1/workspaces/{WS_ID}/lakehouses/{LH_ID}/shortcuts",
    headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
    json=payload
)

Gotcha: Shortcuts are read-only from the Delta Lake perspective — you cannot INSERT/UPDATE/DELETE through a shortcut table in the warehouse SQL endpoint.

Lakehouses

Create & Explore

# Via Fabric REST API
curl -sX POST "https://api.fabric.microsoft.com/v1/workspaces/$WS_ID/lakehouses" \
  -H "Authorization: Bearer $FABRIC_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"displayName": "sales_lakehouse", "description": "Sales data lakehouse"}'

Lakehouse Structure

sales_lakehouse.Lakehouse/
├── Tables/            ← Managed Delta tables (visible in SQL endpoint + Explorer)
│   ├── orders/        ← Delta table (_delta_log/ + parquet files)
│   ├── customers/
│   └── products/
└── Files/             ← Unmanaged files (raw zone, staging, CSV, Parquet, images, etc.)
    ├── raw/
    ├── staging/
    └── reference/

Delta Lake Format

All managed tables in the lakehouse are stored as Delta Lake on OneLake.

# In a Fabric notebook (PySpark)
df = spark.read.format("delta").load(
    "abfss://MyWorkspace@onelake.dfs.fabric.microsoft.com/sales_lakehouse.Lakehouse/Tables/orders"
)

# Write a Delta table (creates managed table visible in lakehouse explorer)
df.write.format("delta").mode("overwrite").saveAsTable("sales_lakehouse.orders")

# Time travel
df_v0 = spark.read.format("delta").option("versionAsOf", 0).load("Tables/orders")
df_ts  = spark.read.format("delta").option("timestampAsOf", "2024-01-01").load("Tables/orders")

V-Order Optimization

Fabric writes Delta tables with V-Order by default — a Microsoft-specific Delta writer optimization that improves read performance for Power BI Direct Lake and Fabric engines. It adds ~10-15% write overhead but cuts read time significantly.

# V-Order is on by default in Fabric notebooks. Disable if needed:
spark.conf.set("spark.sql.parquet.vorder.enabled", "false")

Gotcha: V-Order parquet files are valid standard Delta/Parquet but engines outside Fabric (external Spark clusters, AWS Athena) don't benefit from the optimization — they just see normal Parquet.

Spark in Fabric — Comprehensive Guide

Runtime Versions (as of early 2026)

Runtime 2.0 — Key Changes (Public Preview)

Major version jumps: Spark 3.x → 4.0, Delta 2.x/3.x → 4.0, Java 11 → 21, Scala 2.12 → 2.13, Python 3.10/3.11 → 3.12.

Spark 4.0 Highlights

• VARIANT data type — native semi-structured data support (replaces string-based JSON columns)
• SQL user-defined functions — define reusable SQL functions without Python/Scala
• Session variables — DECLARE VARIABLE, SET VARIABLE for session-scoped state
• Pipe syntax — table |> SELECT ... |> WHERE ... for readable query chaining
• String collation — locale-aware string comparison and sorting
• PySpark native plotting API — built-in df.plot() without matplotlib
• Python Data Source API — custom data sources in pure Python
• Python UDTFs — user-defined table functions in Python
• Unified profiling — profile PySpark UDFs with Spark's built-in profiler
• Structured Streaming Arbitrary State API v2 — more flexible stateful processing
• Structured Streaming State Data Source — debug and inspect streaming state
• SparkR is deprecated — may be removed in a future version

Delta Lake 4.0 Highlights

• Cross-format interoperability improvements
• Performance optimizations for large tables
• Parallel Delta snapshot loading (reduces query startup for large tables)

⚠ Important: Delta Lake 4.0-specific features are experimental and only work in Spark experiences (Notebooks, Spark Job Definitions). If you share Delta tables across multiple Fabric workloads (warehouse, Power BI, etc.), do NOT enable Delta 4.0-only features. Check Delta Lake interoperability docs first.

Performance — Native Execution Engine

• Up to 6× faster than open-source Spark on TPC-DS benchmarks
• Vectorized processing — accelerates queries without code changes
• Vectorized CSV parsing — much faster CSV ingestion
• Vectorized JSON parsing and Streaming support planned for future updates
• Enable at environment level so all jobs inherit the performance boost

# Enable Native Execution Engine (Runtime 2.0)
spark.conf.set("spark.native.enabled", "true")
# Or set at Environment level (recommended — applies to all notebooks/jobs)

Compute Management (Runtime 2.0)

• Resource Profiles — predefined resource allocations for sessions (control cost/performance)
• Custom Live Pools (preview) — pre-warmed Spark pools to eliminate cold start

Migration Notes (1.3 → 2.0)

• Scala 2.12 → 2.13: Recompile any custom Scala JARs. Binary incompatible.
• Java 11 → 21: Review deprecated Java APIs. Most code works, but reflection-heavy libraries may need updates.
• Python 3.11 → 3.12: Check third-party library compatibility (most major libs support 3.12).
• WASB protocol deprecated: Use abfss:// (ABFS) instead of wasbs:// for GPv2 storage accounts.
• SparkR deprecated: Migrate R workloads to Python or consider alternatives.
• Delta 4.0 experimental features: Don't enable on tables shared with non-Spark Fabric workloads.
• Initial session startup may be slower during preview — use Custom Live Pools to mitigate.

Spark Compute Model

Fabric Spark is serverless — no persistent clusters to manage. Sessions are created on demand with configurable node sizes.

Node types (Spark pools):

Executor allocation:

• Default: Dynamic allocation enabled. Min 1 executor, max varies by capacity.
• Executors scale up/down within the session as shuffle stages demand.
• Idle sessions auto-terminate after configurable timeout (default 20 min).

Workspace Spark Settings

Configure under Workspace Settings → Data Engineering/Science:

# Or set per-session in notebooks:
spark.conf.set("spark.executor.instances", "4")
spark.conf.set("spark.executor.cores", "4")
spark.conf.set("spark.executor.memory", "28g")
spark.conf.set("spark.driver.memory", "28g")
spark.conf.set("spark.dynamicAllocation.enabled", "true")
spark.conf.set("spark.dynamicAllocation.minExecutors", "1")
spark.conf.set("spark.dynamicAllocation.maxExecutors", "10")

Environments (Library & Configuration Management)

Create a Fabric Environment item to manage:

• PyPI packages — pinned versions, no wheel upload needed for public packages
• Custom libraries — upload .whl, .jar, .tar.gz files
• Conda packages — via environment.yml
• Spark properties — set defaults per environment
• R packages — install from CRAN

# Install libraries in an Environment via REST
curl -sX POST \
  "https://api.fabric.microsoft.com/v1/workspaces/$WS_ID/environments/$ENV_ID/staging/libraries" \
  -H "Authorization: Bearer $FABRIC_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"pypiPackages": [{"name": "scikit-learn", "version": "1.4.0"}, {"name": "xgboost"}]}'

# Publish the environment (triggers build — takes 3-10 minutes)
curl -sX POST \
  "https://api.fabric.microsoft.com/v1/workspaces/$WS_ID/environments/$ENV_ID/staging/publish" \
  -H "Authorization: Bearer $FABRIC_TOKEN"

# environment.yml example for conda-based environments
name: my_env
channels:
  - conda-forge
  - defaults
dependencies:
  - numpy=1.26.*
  - pandas=2.1.*
  - scipy
  - pip:
    - great-expectations==0.18.*
    - delta-spark==3.2.*

Gotcha: After publishing an Environment, existing running sessions do NOT pick up changes. You must restart the session (or start a new notebook session).

Inline Library Installation (Quick & Dirty)

# Install in current session only — lost on session restart
%pip install great-expectations==0.18.0
%pip install /lakehouse/default/Files/libs/my_custom_lib-1.0.0-py3-none-any.whl

# Conda (slower, but available)
%conda install -c conda-forge prophet

High-Concurrency Mode

Enables session sharing across multiple notebooks and users on the same Spark cluster. Reduces cold-start overhead by reusing live sessions.

# Enable via workspace settings or per-session
# In workspace settings → Data Engineering → High Concurrency Mode: ON

# Notebooks share the same SparkContext but get isolated SparkSessions
# Variables and imports in one notebook are NOT visible to others

When to use: Development scenarios with multiple users iterating on notebooks. When NOT to use: Production jobs where isolation matters, or jobs needing full cluster resources.

Native Execution Engine

Fabric's vectorized execution engine that accelerates Spark queries on Delta tables. Available on Runtime 1.2+; up to 6× faster on Runtime 2.0 (TPC-DS benchmarks).

# Enable native execution engine
spark.conf.set("spark.native.enabled", "true")

# Check if a query used native execution
df = spark.sql("SELECT * FROM orders WHERE amount > 100")
df.explain(True)  # Look for "NativeExecution" in the physical plan

Supported: Scan, filter, project, aggregation, join (hash/sort-merge), sort, window. Runtime 2.0 adds vectorized CSV parsing; vectorized JSON and Streaming support planned. Not supported: UDFs (Python/Scala), complex types (struct/array/map operations), some date/timestamp edge cases. Falls back to JVM Spark transparently.

Notebook Patterns

Basic Read/Write

# Read from default lakehouse (attached in notebook metadata)
df = spark.read.format("delta").table("orders")

# Equivalent using spark.sql
df = spark.sql("SELECT * FROM sales_lakehouse.orders")

# Read from another lakehouse in same workspace
df = spark.read.format("delta").load(
    "abfss://MyWorkspace@onelake.dfs.fabric.microsoft.com/other_lakehouse.Lakehouse/Tables/customers"
)

# Write to managed table
df.write.format("delta").mode("overwrite").saveAsTable("orders_enriched")

# Write to Files section (unmanaged / raw zone)
df.write.parquet("Files/raw/orders/2024/01/")

# Append with schema evolution
df.write.format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("orders")

Merge / Upsert (Delta MERGE INTO)

from delta.tables import DeltaTable

target = DeltaTable.forName(spark, "orders")
source = spark.read.format("delta").table("orders_staging")

target.alias("t") \
    .merge(source.alias("s"), "t.order_id = s.order_id") \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .whenNotMatchedBySourceDelete() \
    .execute()

-- Spark SQL equivalent
MERGE INTO orders AS t
USING orders_staging AS s
ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
WHEN NOT MATCHED BY SOURCE THEN DELETE;

Schema Management

# Schema enforcement (default) — rejects writes with mismatched schema
df.write.format("delta").mode("append").saveAsTable("orders")  # Fails if schema differs

# Schema evolution — allows adding new columns
df.write.format("delta").mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("orders")

# Overwrite schema entirely
df.write.format("delta").mode("overwrite") \
    .option("overwriteSchema", "true") \
    .saveAsTable("orders")

# Add/rename/drop columns via ALTER TABLE
spark.sql("ALTER TABLE orders ADD COLUMN (region STRING AFTER customer_id)")
spark.sql("ALTER TABLE orders RENAME COLUMN region TO sales_region")
spark.sql("ALTER TABLE orders DROP COLUMN sales_region")

# Set column comments
spark.sql("ALTER TABLE orders ALTER COLUMN amount COMMENT 'Order total in USD'")

Partitioning Strategies

# Partition by date — good for time-series, daily ingestion
df.write.format("delta") \
    .partitionBy("order_date") \
    .mode("overwrite") \
    .saveAsTable("orders_partitioned")

# Liquid clustering (Delta 3.x / Runtime 1.3) — PREFERRED over partitioning
# Self-optimizing, no need to choose partition columns upfront
spark.sql("""
    CREATE TABLE orders_clustered
    USING DELTA
    CLUSTER BY (customer_id, order_date)
    AS SELECT * FROM orders
""")

# Alter clustering keys (non-breaking — applies incrementally)
spark.sql("ALTER TABLE orders_clustered CLUSTER BY (region, order_date)")

When to partition: Only when partition column has low cardinality (<1000 values) and queries almost always filter on it. Otherwise prefer liquid clustering.

Gotcha: Over-partitioned tables (e.g., partition by high-cardinality column) create thousands of tiny files → severe small-file problem → slow reads and costly OPTIMIZE.

Z-ORDER (for Runtime 1.2 / Delta 2.x)

-- Z-ORDER co-locates data by column values within partitions
OPTIMIZE orders ZORDER BY (customer_id, order_date);

-- Not needed with liquid clustering (Runtime 1.3) — clustering replaces Z-ORDER

Delta Table Maintenance

OPTIMIZE (Compaction)

# Compact small files into larger ones (target ~1 GB per file)
spark.sql("OPTIMIZE orders")

# With predicate — only compact files in matching partitions
spark.sql("OPTIMIZE orders WHERE order_date >= '2024-01-01'")

# In a notebook using Delta API
from delta.tables import DeltaTable
dt = DeltaTable.forName(spark, "orders")
dt.optimize().executeCompaction()

# With Z-ORDER
dt.optimize().executeZOrderBy("customer_id")

VACUUM (Remove Old Files)

# Remove files no longer referenced by Delta log (default retention: 7 days)
spark.sql("VACUUM orders")

# Reduce retention (DANGEROUS — breaks time travel for vacuumed versions)
spark.sql("SET spark.databricks.delta.retentionDurationCheck.enabled = false")
spark.sql("VACUUM orders RETAIN 24 HOURS")

# DeltaTable API
dt = DeltaTable.forName(spark, "orders")
dt.vacuum(retentionHours=168)  # 7 days

Gotcha: VACUUM is not automatic in Fabric. Schedule it via pipeline or cron notebook. Without regular VACUUM, OneLake storage costs grow unbounded from historical file versions.

ANALYZE TABLE (Statistics)

# Collect statistics for query optimization
spark.sql("ANALYZE TABLE orders COMPUTE STATISTICS FOR ALL COLUMNS")

# For specific columns
spark.sql("ANALYZE TABLE orders COMPUTE STATISTICS FOR COLUMNS order_id, customer_id, amount")

Medallion Architecture in Fabric

Files/raw/          → Bronze (raw ingestion, append-only)
Tables/bronze_*     → Bronze Delta tables (schema-enforced raw)
Tables/silver_*     → Silver (cleansed, deduplicated, typed)
Tables/gold_*       → Gold (aggregated, business-ready, Direct Lake-ready)

# Bronze → Silver transformation
bronze_df = spark.read.format("delta").table("bronze_orders")

silver_df = (
    bronze_df
    .dropDuplicates(["order_id"])
    .withColumn("order_date", F.to_date("order_date_str", "yyyy-MM-dd"))
    .withColumn("amount", F.col("amount").cast("decimal(18,2)"))
    .filter(F.col("order_id").isNotNull())
    .withColumn("_ingested_at", F.current_timestamp())
)

silver_df.write.format("delta").mode("overwrite").saveAsTable("silver_orders")

# Silver → Gold aggregation
gold_df = spark.sql("""
    SELECT
        date_trunc('month', order_date) AS month,
        region,
        COUNT(*) AS order_count,
        SUM(amount) AS total_revenue,
        AVG(amount) AS avg_order_value
    FROM silver_orders
    WHERE order_date >= '2024-01-01'
    GROUP BY 1, 2
""")

gold_df.write.format("delta").mode("overwrite").saveAsTable("gold_monthly_revenue")

mssparkutils — Fabric's Utility Library

# Filesystem operations
mssparkutils.fs.ls("Files/raw/")
mssparkutils.fs.cp("Files/raw/data.csv", "Files/staging/data.csv")
mssparkutils.fs.mv("Files/staging/data.csv", "Files/archive/data.csv")
mssparkutils.fs.rm("Files/temp/", recurse=True)
mssparkutils.fs.mkdirs("Files/output/2024/01/")
mssparkutils.fs.head("Files/raw/sample.csv", maxBytes=1024)
mssparkutils.fs.put("Files/config/params.json", '{"key": "value"}', overwrite=True)

# Cross-workspace access
mssparkutils.fs.ls(
    "abfss://OtherWorkspace@onelake.dfs.fabric.microsoft.com/other_lakehouse.Lakehouse/Files/"
)

# Notebook orchestration
mssparkutils.notebook.run("child_notebook", timeout=600, arguments={"date": "2024-01-15"})
mssparkutils.notebook.runMultiple([
    {"name": "etl_orders", "timeout": 600, "args": {"date": "2024-01-15"}},
    {"name": "etl_customers", "timeout": 600},
])
# runMultiple executes notebooks in parallel (respects DAG if dependencies specified)

# Notebook exit with return value
mssparkutils.notebook.exit("success:42")  # returns string to caller

# Credentials / secrets
token = mssparkutils.credentials.getToken("https://storage.azure.com/")
secret = mssparkutils.credentials.getSecret("https://my-kv.vault.azure.net/", "my-secret")
conn_str = mssparkutils.credentials.getConnectionStringOrCred("MyConnection")

# Runtime context
workspace_id = mssparkutils.runtime.context.get("currentWorkspaceId")
lakehouse_id = mssparkutils.runtime.context.get("lakehouseId")

# Lakehouse management (attach/detach within notebook)
mssparkutils.lakehouse.list()
mssparkutils.lakehouse.get("sales_lakehouse")

Spark SQL — Common Patterns in Fabric

-- Show all tables in default lakehouse
SHOW TABLES;

-- Describe table schema with column comments
DESCRIBE EXTENDED orders;

-- Table properties (Delta metadata)
SHOW TBLPROPERTIES orders;

-- Table history (Delta log)
DESCRIBE HISTORY orders;

-- Convert Parquet files to managed Delta table
CREATE TABLE bronze_orders
USING DELTA
AS SELECT * FROM parquet.`Files/raw/orders/`;

-- Create table with explicit schema
CREATE TABLE IF NOT EXISTS orders (
    order_id LONG NOT NULL,
    customer_id LONG,
    order_date DATE,
    amount DECIMAL(18,2),
    status STRING,
    _loaded_at TIMESTAMP
)
USING DELTA
CLUSTER BY (customer_id, order_date)
TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true',
    'delta.autoOptimize.autoCompact' = 'true'
);

-- Insert overwrite by partition (partial overwrite)
INSERT OVERWRITE TABLE orders
PARTITION (order_date = '2024-01-15')
SELECT order_id, customer_id, '2024-01-15' AS order_date, amount, status, current_timestamp()
FROM staging_orders;

-- Delete with time-travel rollback safety
DELETE FROM orders WHERE status = 'cancelled' AND order_date < '2023-01-01';

-- Restore table to previous version
RESTORE TABLE orders TO VERSION AS OF 42;
RESTORE TABLE orders TO TIMESTAMP AS OF '2024-01-15 08:00:00';

-- Clone a table (zero-copy shallow clone)
CREATE TABLE orders_clone SHALLOW CLONE orders;
-- Deep clone (full copy)
CREATE TABLE orders_backup DEEP CLONE orders;

-- Cross-lakehouse query
SELECT a.*, b.customer_name
FROM sales_lakehouse.orders a
JOIN crm_lakehouse.customers b ON a.customer_id = b.id;

Spark Structured Streaming in Fabric

from pyspark.sql import functions as F
from pyspark.sql.types import *

# Read streaming from Files (auto-loader pattern — new files trigger processing)
schema = StructType([
    StructField("order_id", LongType()),
    StructField("customer_id", LongType()),
    StructField("amount", DoubleType()),
    StructField("event_time", TimestampType()),
])

stream_df = (
    spark.readStream
    .format("cloudFiles")           # Auto Loader (Fabric-supported)
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "Files/_checkpoints/orders_schema/")
    .option("cloudFiles.inferColumnTypes", "true")
    .schema(schema)
    .load("Files/raw/orders_stream/")
)

# Transform
enriched = (
    stream_df
    .withColumn("_ingested_at", F.current_timestamp())
    .withColumn("amount_usd", F.col("amount").cast("decimal(18,2)"))
    .withWatermark("event_time", "1 hour")
)

# Write to Delta table (streaming sink)
query = (
    enriched.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "Files/_checkpoints/orders_bronze/")
    .trigger(availableNow=True)      # Process all available, then stop (micro-batch)
    # .trigger(processingTime="5 minutes")  # Continuous micro-batch
    .toTable("bronze_orders")
)
query.awaitTermination()

# Stream-to-stream: read from Delta, aggregate, write to Gold
orders_stream = spark.readStream.format("delta").table("silver_orders")

agg_stream = (
    orders_stream
    .withWatermark("order_date", "1 day")
    .groupBy(F.window("order_date", "1 day"), "region")
    .agg(
        F.count("*").alias("order_count"),
        F.sum("amount").alias("total_revenue"),
    )
    .select("window.start", "region", "order_count", "total_revenue")
)

agg_query = (
    agg_stream.writeStream
    .format("delta")
    .outputMode("complete")
    .option("checkpointLocation", "Files/_checkpoints/gold_daily_revenue/")
    .trigger(availableNow=True)
    .toTable("gold_daily_revenue")
)

Trigger modes:

Gotcha: Streaming notebooks keep the Spark session alive and consume CUs continuously when using processingTime. For cost control, prefer availableNow=True triggered by a pipeline on a schedule.

Spark Job Definitions (Headless Batch)

For production Spark jobs that don't need a notebook UI:

# Create a Spark Job Definition via REST
curl -sX POST "https://api.fabric.microsoft.com/v1/workspaces/$WS_ID/sparkjobdefinitions" \
  -H "Authorization: Bearer $FABRIC_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "displayName": "daily_etl_job",
    "definition": {
      "format": "SparkJobDefinitionV1",
      "parts": [{
        "path": "SparkJobDefinitionV1.json",
        "payload": "{\"executableFile\": \"Files/jobs/daily_etl.py\", \"defaultLakehouse\": {\"name\": \"sales_lakehouse\"}, \"language\": \"python\", \"args\": [\"--date\", \"2024-01-15\"]}"
      }]
    }
  }'

# daily_etl.py — standalone Spark script
import sys
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("daily_etl").getOrCreate()

date_param = sys.argv[sys.argv.index("--date") + 1] if "--date" in sys.argv else None

df = spark.read.format("delta").table("bronze_orders")
if date_param:
    df = df.filter(f"order_date = '{date_param}'")

result = df.groupBy("region").agg({"amount": "sum"})
result.write.format("delta").mode("overwrite").saveAsTable("gold_revenue_by_region")

spark.stop()

Spark Performance Tuning in Fabric

Auto-Optimize Settings

# Per-table auto-optimize (recommended for all tables)
spark.sql("""
    ALTER TABLE orders SET TBLPROPERTIES (
        'delta.autoOptimize.optimizeWrite' = 'true',
        'delta.autoOptimize.autoCompact' = 'true'
    )
""")
# optimizeWrite: coalesces small writes into ~128 MB files
# autoCompact: triggers mini-OPTIMIZE after writes when small files accumulate

Adaptive Query Execution (AQE)

# AQE is ON by default in Fabric. Key settings:
spark.conf.set("spark.sql.adaptive.enabled", "true")                    # default
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true") # default
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")           # default

# Adjust shuffle partition count (default 200 — AQE coalesces automatically)
spark.conf.set("spark.sql.shuffle.partitions", "auto")  # let AQE decide

Broadcast Joins

from pyspark.sql import functions as F

# Force broadcast for small dimension tables (<100 MB)
small_df = spark.read.format("delta").table("dim_products")  # ~50 MB
large_df = spark.read.format("delta").table("fact_orders")   # 100 GB

result = large_df.join(F.broadcast(small_df), "product_id")

# Adjust broadcast threshold (default 10 MB)
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "100m")  # 100 MB

Caching

# Cache hot tables in memory (Spark executor memory)
df = spark.read.format("delta").table("dim_customers")
df.cache()
df.count()  # Materialize cache

# Uncache when done
df.unpersist()

# CACHE TABLE (SQL)
spark.sql("CACHE TABLE dim_customers")
spark.sql("UNCACHE TABLE dim_customers")

Gotcha: Caching uses executor memory. On small capacity SKUs (F2/F4), caching large tables can cause OOM. Use caching sparingly and only for repeatedly-read small tables.

Common Performance Anti-Patterns

Spark 4.0 New Syntax Examples (Runtime 2.0)

-- VARIANT data type — store semi-structured data natively
CREATE TABLE events (
    event_id LONG,
    event_time TIMESTAMP,
    payload VARIANT    -- replaces storing JSON as STRING
);
INSERT INTO events VALUES (1, current_timestamp(), PARSE_JSON('{"action": "click", "page": "/home"}'));
SELECT event_id, payload:action::STRING AS action FROM events;  -- schema-on-read

-- Pipe syntax — readable query chaining (left-to-right)
FROM orders
|> WHERE amount > 100
|> SELECT customer_id, amount
|> ORDER BY amount DESC
|> LIMIT 10;

-- Session variables
DECLARE VARIABLE batch_date DATE DEFAULT CURRENT_DATE();
SET VARIABLE batch_date = DATE '2024-06-15';
SELECT * FROM orders WHERE order_date = batch_date;

-- SQL user-defined functions
CREATE FUNCTION discount(amount DECIMAL(18,2), pct DECIMAL(5,2))
RETURNS DECIMAL(18,2)
RETURN amount * (1 - pct / 100);
SELECT order_id, discount(amount, 10) AS discounted FROM orders;

# PySpark native plotting (Spark 4.0) — no matplotlib needed
df = spark.read.format("delta").table("gold_monthly_revenue")
pdf = df.toPandas()
# In Runtime 2.0 notebooks, df.plot() works natively:
df.select("month", "total_revenue").toPandas().plot(x="month", y="total_revenue", kind="bar")

# Python UDTF — return multiple rows from a function
from pyspark.sql.functions import udtf

@udtf(returnType="word: string, count: int")
class WordCount:
    def eval(self, text: str):
        for word in text.split():
            yield word.lower(), 1

spark.sql("SELECT * FROM WordCount('Hello World Hello')")
# Returns: (hello, 1), (world, 1), (hello, 1)

# Python Data Source API — custom data source in pure Python
from pyspark.sql.datasource import DataSource, DataSourceReader

class MyApiSource(DataSource):
    @classmethod
    def name(cls): return "myapi"
    def reader(self, schema): return MyApiReader(self.options)

class MyApiReader(DataSourceReader):
    def __init__(self, options): self.url = options["url"]
    def read(self, partition):
        import requests
        for row in requests.get(self.url).json():
            yield tuple(row.values())

# Structured Streaming State Data Source — inspect streaming state
state_df = spark.read.format("statestore") \
    .option("checkpointLocation", "Files/_checkpoints/my_stream/") \
    .load()
state_df.show()  # Debug what's in the streaming state store

Data Science & MLflow in Fabric

import mlflow
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split

# MLflow auto-logging (enabled by default in Fabric notebooks)
mlflow.autolog()

# Load data from lakehouse
df = spark.read.format("delta").table("silver_customers").toPandas()
X = df[["age", "income", "tenure"]]
y = df["churn"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

# Train with MLflow tracking
with mlflow.start_run(run_name="rf_churn_v1") as run:
    model = RandomForestClassifier(n_estimators=100, max_depth=10)
    model.fit(X_train, y_train)
    accuracy = model.score(X_test, y_test)
    mlflow.log_metric("accuracy", accuracy)
    mlflow.sklearn.log_model(model, "model")
    print(f"Run ID: {run.info.run_id}, Accuracy: {accuracy:.4f}")

# Register model
mlflow.register_model(f"runs:/{run.info.run_id}/model", "churn_predictor")

# Load and predict with registered model
loaded_model = mlflow.pyfunc.load_model("models:/churn_predictor/latest")
predictions = loaded_model.predict(X_test)

PREDICT function (T-SQL inference)

-- Use MLflow model directly in warehouse T-SQL queries
SELECT *, PREDICT('churn_predictor', 1) AS churn_prediction
FROM customers;

Notebook Scheduling

# Notebooks can be scheduled directly (without a Pipeline) via REST
curl -sX POST \
  "https://api.fabric.microsoft.com/v1/workspaces/$WS_ID/items/$NOTEBOOK_ID/jobs/instances?jobType=RunNotebook" \
  -H "Authorization: Bearer $FABRIC_TOKEN" \
  -H "Content-Type: application/json"

# Or embed in a Pipeline as a Notebook Activity with parameters:
# Pipeline → Add Activity → Notebook → Settings → Base parameters

Fabric Spark vs Databricks — Key Differences

Data Warehouse

The Fabric Warehouse is a serverless, fully managed T-SQL engine layered over Delta Lake in OneLake. No indexes, no statistics management, no traditional DBA tasks.

T-SQL Patterns

-- Cross-database query (within same workspace — no linked server needed)
SELECT o.order_id, c.customer_name
FROM sales_warehouse.dbo.orders o
JOIN sales_lakehouse.dbo.customers c    -- lakehouse SQL endpoint
    ON o.customer_id = c.customer_id;

-- COPY INTO (load from ADLS / OneLake Files section)
COPY INTO dbo.orders
FROM 'https://onelake.dfs.fabric.microsoft.com/MyWorkspace/sales_lakehouse.Lakehouse/Files/raw/orders/'
WITH (
    FILE_TYPE = 'PARQUET',
    CREDENTIAL = (IDENTITY = 'Managed Identity')
);

-- External table over Files section
CREATE EXTERNAL TABLE dbo.orders_ext (
    order_id INT,
    order_date DATE,
    amount DECIMAL(18,2)
)
WITH (
    LOCATION = 'Files/raw/orders/',
    DATA_SOURCE = lakehouse_files_ds,
    FILE_FORMAT = parquet_ff
);

-- DDL differences from Azure SQL: no IDENTITY workaround needed, use sequences
CREATE SEQUENCE dbo.order_seq START WITH 1 INCREMENT BY 1;

Gotcha: The Warehouse does not support traditional clustered indexes or columnstore indexes — the engine decides physical layout. You cannot force distribution (unlike Synapse Dedicated).

Data Factory in Fabric

Fabric Data Factory uses the same pipeline canvas as ADF but runs on Fabric compute.

Pipelines

# Create a pipeline via REST
curl -sX POST "https://api.fabric.microsoft.com/v1/workspaces/$WS_ID/datapipelines" \
  -H "Authorization: Bearer $FABRIC_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"displayName": "ingest_orders_pipeline"}'

Dataflows Gen2

Dataflows Gen2 use Power Query M for transformation and stage data through an internal staging lakehouse before writing to destination.

Key differences from ADF Mapping Data Flows:

• Low-code / no-code Power Query editor
• Uses Fabric compute (no IR to manage)
• Staging adds latency but enables larger datasets than Power Query Online

Gotcha: Dataflows Gen2 always stage data internally — you pay CU cost for the staging write AND the final destination write. For simple copies, a pipeline with Copy Activity is cheaper.

Copy Activity — Common Patterns

{
  "name": "CopyFromBlobToLakehouse",
  "type": "Copy",
  "inputs": [{"referenceName": "BlobDataset", "type": "DatasetReference"}],
  "outputs": [{"referenceName": "LakehouseTableDataset", "type": "DatasetReference"}],
  "typeProperties": {
    "source": {"type": "ParquetSource"},
    "sink": {
      "type": "LakehouseTableSink",
      "tableOption": "autoCreate",
      "writeBehavior": "Append"
    }
  }
}

Real-Time Intelligence

KQL Databases & Eventhouses

// Create table
.create table Telemetry (
    Timestamp: datetime,
    DeviceId: string,
    Temperature: real,
    Humidity: real
)

// Ingest from eventstream (configured via UI) or from storage
.ingest into table Telemetry
    (h'https://mystorageaccount.blob.core.windows.net/mycontainer/data.json.gz;impersonate')
    with (format='json', ingestionMappingReference='TelemetryMapping')

// Common analytics patterns
Telemetry
| where Timestamp > ago(1h)
| summarize avg_temp = avg(Temperature) by bin(Timestamp, 5m), DeviceId
| render timechart

// Join with refer

…(truncated)