Fabric Data Engineering

Fabric Data Engineering

1. Fabric Data Engineering Overview

Microsoft Fabric is a unified analytics platform that brings together data engineering, data science, real-time analytics, and business intelligence into a single SaaS experience. The Data Engineering persona focuses on building lakehouses, authoring Spark notebooks, managing Delta Lake tables, and orchestrating data pipelines.

Core concepts:

Fabric capacity model:

Workspace roles:

Relationship to OneLake: Every Fabric tenant has one OneLake. Every workspace maps to a folder in OneLake. Every lakehouse maps to a subfolder. Data is stored once and accessed by all Fabric engines (Spark, SQL, Power BI, KQL) without duplication. The ADLS Gen2 endpoint is: abfss://<workspace-name>@onelake.dfs.fabric.microsoft.com/<item-name>.<item-type>.

2. Lakehouses

A Fabric lakehouse is a unified analytics store that combines unstructured files with managed Delta tables, automatically exposing a read-only SQL analytics endpoint.

Create a Lakehouse

Via Fabric portal: Workspace > New > Lakehouse > Enter name.

Via REST API:

POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/lakehouses
Content-Type: application/json
Authorization: Bearer {token}

{
  "displayName": "lh_bronze_sales"
}

Response includes the lakehouse id, displayName, and properties including SQL connection string.

Folder Structure

Every lakehouse has two top-level sections:

<lakehouse-name>.Lakehouse/
├── Tables/           # Managed Delta Lake tables (appear in SQL endpoint)
│   ├── customers/    # Delta table (Parquet files + _delta_log/)
│   ├── orders/
│   └── products/
└── Files/            # Unstructured/raw files (not in SQL endpoint)
    ├── landing/      # Raw ingested files (CSV, JSON, Parquet, Excel)
    ├── archive/      # Processed files moved after ingestion
    └── _errors/      # Files that failed processing

• Tables/: Only managed Delta tables. These automatically appear in the SQL analytics endpoint. Created via spark.write.saveAsTable() or CREATE TABLE.
• Files/: Raw data, staging files, exports. Accessed via spark.read.load("Files/...") or OneLake ADLS Gen2 path. NOT visible in the SQL endpoint.

SQL Analytics Endpoint

Every lakehouse auto-generates a read-only T-SQL endpoint:

• Connection string: Found in lakehouse settings > SQL analytics endpoint.
• Supported queries: SELECT, views, table-valued functions. No INSERT/UPDATE/DELETE.
• Use cases: Power BI DirectQuery/Import, SSMS exploration, cross-lakehouse queries.
• Cross-database queries: Reference other lakehouses in the same workspace with three-part naming: SELECT * FROM [other_lakehouse].[dbo].[table_name].

Table Maintenance

Delta tables accumulate small files over time and require periodic maintenance:

-- Compact small files into larger ones (target ~128 MB per file)
OPTIMIZE lakehouse_name.table_name;

-- Remove old file versions (default retention: 7 days)
VACUUM lakehouse_name.table_name;

-- Cluster data by frequently filtered columns
OPTIMIZE lakehouse_name.table_name ZORDER BY (customer_id, order_date);

-- Check table health
DESCRIBE DETAIL lakehouse_name.table_name;
-- Look at: numFiles (should be reasonable), sizeInBytes, numPartitions

Recommended maintenance schedule:

Lakehouse Explorer

The lakehouse explorer in the Fabric portal provides:

• Table preview: View schema, sample data, and statistics for any Delta table.
• File browser: Navigate the Files/ section, upload/download files.
• SQL editor: Run T-SQL queries against the SQL analytics endpoint.
• Table maintenance: Run OPTIMIZE and VACUUM from the UI.

3. Spark Notebooks

Fabric Spark notebooks are interactive development environments for PySpark, SparkSQL, Scala, and R. They run on managed Spark clusters within the Fabric capacity.

Notebook Structure

A Fabric notebook consists of cells, each with a language and execution order:

Cell 1 [markdown]  - Title, description, parameters documentation
Cell 2 [pyspark]   - Configuration and parameters
Cell 3 [pyspark]   - Read source data
Cell 4 [pyspark]   - Transform
Cell 5 [sparksql]  - Validate with SQL query
Cell 6 [pyspark]   - Write to target
Cell 7 [pyspark]   - Log completion

Language magics (switch language within a notebook):

PySpark DataFrame Operations

Reading data:

# Read a Delta table
df = spark.read.table("lakehouse_name.table_name")

# Read from the Files section
df = spark.read.format("csv").option("header", "true").load("Files/landing/*.csv")
df = spark.read.format("parquet").load("Files/staging/data.parquet")
df = spark.read.format("json").option("multiLine", "true").load("Files/landing/data.json")

# Read from OneLake path
df = spark.read.format("delta").load("abfss://workspace@onelake.dfs.fabric.microsoft.com/lakehouse.Lakehouse/Tables/table_name")

# Read from a shortcut
df = spark.read.table("shortcut_table_name")

Transformations:

from pyspark.sql.functions import col, when, lit, trim, lower, upper, concat, date_format, current_timestamp
from pyspark.sql.functions import sum, count, avg, min, max, row_number, dense_rank
from pyspark.sql.window import Window

# Column operations
df = df.select("id", "name", "amount")
df = df.withColumn("amount_usd", col("amount") * col("exchange_rate"))
df = df.withColumnRenamed("old_name", "new_name")
df = df.drop("unwanted_column")

# Filtering
df = df.filter(col("status") == "active")
df = df.filter(col("amount").between(100, 1000))
df = df.filter(col("name").isNotNull())

# Aggregation
summary = df.groupBy("region", "category") \
    .agg(
        sum("amount").alias("total_amount"),
        count("id").alias("record_count"),
        avg("amount").alias("avg_amount")
    )

# Joins
result = orders.join(customers, orders.customer_id == customers.id, "left")
result = orders.join(broadcast(lookup_table), "status_code", "inner")  # Broadcast small tables

# Window functions
window = Window.partitionBy("customer_id").orderBy(col("order_date").desc())
df = df.withColumn("row_num", row_number().over(window))
df = df.withColumn("rank", dense_rank().over(window))

# Deduplication
df = df.dropDuplicates(["id"])
# or keep latest by date
df = df.withColumn("rn", row_number().over(
    Window.partitionBy("id").orderBy(col("updated_at").desc())
)).filter(col("rn") == 1).drop("rn")

Writing data:

# Write as managed Delta table
df.write.format("delta").mode("overwrite").saveAsTable("table_name")

# Write with partitioning
df.write.format("delta").mode("overwrite").partitionBy("year", "month").saveAsTable("table_name")

# Append
df.write.format("delta").mode("append").saveAsTable("table_name")

# Write to Files section (non-Delta)
df.write.format("parquet").mode("overwrite").save("Files/staging/output.parquet")
df.write.format("csv").option("header", "true").mode("overwrite").save("Files/exports/output.csv")

SparkSQL Magic

Use %%sql cells for SQL queries that leverage Spark SQL:

%%sql
-- Query Delta tables directly
SELECT customer_id, SUM(amount) as total
FROM lh_silver_sales.orders
WHERE order_date >= '2024-01-01'
GROUP BY customer_id
ORDER BY total DESC
LIMIT 100;

-- Create a temp view from a previous DataFrame
-- (run in a PySpark cell first: df.createOrReplaceTempView("my_view"))
SELECT * FROM my_view WHERE status = 'active';

-- CTAS (Create Table As Select)
CREATE OR REPLACE TABLE lh_gold_sales.customer_summary AS
SELECT customer_id, COUNT(*) as order_count, SUM(amount) as lifetime_value
FROM lh_silver_sales.orders
GROUP BY customer_id;

Notebook Parameters and Scheduling

Notebooks can receive parameters when called from pipelines:

# Define parameters at the top of the notebook
# These are overridden by pipeline Notebook Activity parameters
source_table = "raw_orders"       # type: string
target_table = "clean_orders"     # type: string
run_date = "2024-01-15"           # type: string
batch_size = 10000                # type: int

Pipeline integration: When a pipeline Notebook Activity calls this notebook, it passes parameter values that override the defaults. The parameter cell must be tagged as a "Parameters" cell in the notebook UI.

mssparkutils

mssparkutils is a built-in utility library available in all Fabric notebooks:

File system operations (mssparkutils.fs):

# List files
files = mssparkutils.fs.ls("Files/landing/")
for f in files:
    print(f.name, f.size, f.isDir)

# Copy files
mssparkutils.fs.cp("Files/landing/data.csv", "Files/archive/data.csv")

# Move files
mssparkutils.fs.mv("Files/landing/data.csv", "Files/archive/data.csv")

# Delete files
mssparkutils.fs.rm("Files/archive/old_data.csv")

# Create directory
mssparkutils.fs.mkdirs("Files/staging/temp/")

Credentials (mssparkutils.credentials):

# Get secret from Azure Key Vault
secret = mssparkutils.credentials.getSecret("https://myvault.vault.azure.net/", "secret-name")

# Get access token for an Azure service
token = mssparkutils.credentials.getToken("https://graph.microsoft.com")
token = mssparkutils.credentials.getToken("https://database.windows.net")
token = mssparkutils.credentials.getToken("https://storage.azure.com")

Notebook orchestration (mssparkutils.notebook):

# Run another notebook (synchronous)
result = mssparkutils.notebook.run("child_notebook", timeout_seconds=600, arguments={"param1": "value1"})

# Run multiple notebooks in parallel
mssparkutils.notebook.runMultiple([
    {"name": "notebook_a", "timeoutPerCellInSeconds": 300, "args": {"param1": "val1"}},
    {"name": "notebook_b", "timeoutPerCellInSeconds": 300, "args": {"param1": "val2"}}
])

# Exit notebook with a value (returned to pipeline or parent notebook)
mssparkutils.notebook.exit("success")

Session Configuration

Configure Spark behavior at the notebook level:

# Shuffle partitions (auto-tune in Fabric)
spark.conf.set("spark.sql.shuffle.partitions", "auto")

# Delta write optimization
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

# V-Order optimization (Fabric-specific, optimizes for Power BI)
spark.conf.set("spark.sql.parquet.vorder.enabled", "true")

# Adaptive query execution
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")

# Memory and overhead
spark.conf.set("spark.driver.memory", "8g")
spark.conf.set("spark.executor.memory", "16g")

4. Delta Lake Tables

Delta Lake is the default and required table format in Fabric lakehouses. It stores data as Parquet files with a JSON transaction log (_delta_log/) that provides ACID transactions, time travel, schema evolution, and audit history.

Delta Format Features

CREATE TABLE

-- Managed table (stored in Tables/ of the lakehouse)
CREATE TABLE IF NOT EXISTS customers (
    customer_id STRING NOT NULL,
    first_name STRING,
    last_name STRING,
    email STRING,
    segment STRING,
    created_at TIMESTAMP,
    updated_at TIMESTAMP,
    _ingested_at TIMESTAMP DEFAULT current_timestamp()
)
USING DELTA
COMMENT 'Customer master data from CRM system'
TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true',
    'delta.autoOptimize.autoCompact' = 'true'
);

-- Partitioned table
CREATE TABLE IF NOT EXISTS orders (
    order_id STRING NOT NULL,
    customer_id STRING NOT NULL,
    order_date DATE,
    amount DECIMAL(18,2),
    status STRING,
    year_month STRING GENERATED ALWAYS AS (date_format(order_date, 'yyyy-MM'))
)
USING DELTA
PARTITIONED BY (year_month)
TBLPROPERTIES (
    'delta.autoOptimize.optimizeWrite' = 'true'
);

-- Table with liquid clustering (alternative to partitioning + Z-ORDER)
CREATE TABLE IF NOT EXISTS events (
    event_id STRING NOT NULL,
    user_id STRING NOT NULL,
    event_type STRING,
    event_timestamp TIMESTAMP,
    properties MAP<STRING, STRING>
)
USING DELTA
CLUSTER BY (user_id, event_type);

MERGE INTO (Upsert)

MERGE INTO silver.customers AS target
USING bronze.raw_customers AS source
ON target.customer_id = source.customer_id
WHEN MATCHED AND target._hash != source._hash THEN
    UPDATE SET
        target.first_name = source.first_name,
        target.last_name = source.last_name,
        target.email = source.email,
        target.segment = source.segment,
        target.updated_at = current_timestamp()
WHEN NOT MATCHED THEN
    INSERT (customer_id, first_name, last_name, email, segment, created_at, updated_at)
    VALUES (source.customer_id, source.first_name, source.last_name, source.email, source.segment, current_timestamp(), current_timestamp())
WHEN NOT MATCHED BY SOURCE AND target.is_active = true THEN
    UPDATE SET target.is_active = false, target.updated_at = current_timestamp();

PySpark MERGE:

from delta.tables import DeltaTable

target = DeltaTable.forName(spark, "silver.customers")
target.alias("t").merge(
    source_df.alias("s"),
    "t.customer_id = s.customer_id"
).whenMatchedUpdate(
    condition="t._hash != s._hash",
    set={
        "first_name": "s.first_name",
        "last_name": "s.last_name",
        "email": "s.email",
        "updated_at": "current_timestamp()"
    }
).whenNotMatchedInsertAll() \
 .execute()

UPDATE and DELETE

-- Update specific rows
UPDATE silver.customers
SET segment = 'premium', updated_at = current_timestamp()
WHERE lifetime_value > 10000;

-- Delete rows
DELETE FROM silver.customers
WHERE is_active = false AND updated_at < dateadd(YEAR, -2, current_timestamp());

Schema Enforcement vs Evolution

Schema enforcement (default): Any write that adds, removes, or changes column types is rejected.

# This FAILS if df has columns not in the table schema
df.write.format("delta").mode("append").saveAsTable("customers")
# AnalysisException: A schema mismatch detected when writing to the Delta table.

Schema evolution: Explicitly allow schema changes:

# Merge schema (add new columns from source)
df.write.format("delta").mode("append").option("mergeSchema", "true").saveAsTable("customers")

# Overwrite schema entirely (replace with new schema)
df.write.format("delta").mode("overwrite").option("overwriteSchema", "true").saveAsTable("customers")

Best practices:

• Bronze layer: mergeSchema = true (accept whatever the source sends).
• Silver layer: Enforce schema by default; use mergeSchema only in planned migration notebooks.
• Gold layer: Strict schema enforcement always.

Partitioning Strategy

Anti-patterns to avoid:

• Partitioning by high-cardinality columns (>10,000 values) creates too many small files.
• Partitioning by columns not used in WHERE clauses adds overhead with no benefit.
• Nested partitioning with more than 2 levels (e.g., year/month/day) often creates excessive directories.

Liquid Clustering

Liquid clustering is a newer alternative to partitioning + Z-ORDER that dynamically reorganizes data:

-- Create a table with liquid clustering
CREATE TABLE events CLUSTER BY (user_id, event_date) ...;

-- Change clustering columns without rewriting data
ALTER TABLE events CLUSTER BY (event_type, event_date);

-- Remove clustering
ALTER TABLE events CLUSTER BY NONE;

-- Trigger clustering (happens automatically during writes with autoCompact)
OPTIMIZE events;

Advantages over partitioning + Z-ORDER:

• No need to choose partition columns upfront.
• Clustering columns can be changed without full table rewrite.
• Works well with high-cardinality columns.
• Incremental clustering (only reorganizes new data during OPTIMIZE).

V-Order Optimization

V-Order is a Fabric-specific write-time optimization that reorders Parquet row groups for faster reads by Power BI and SQL endpoints:

# Enable V-Order (recommended for Gold-layer tables consumed by Power BI)
spark.conf.set("spark.sql.parquet.vorder.enabled", "true")

df.write.format("delta").mode("overwrite").saveAsTable("gold.sales_summary")

V-Order adds minimal write overhead (~5%) but can improve Power BI DirectQuery scan performance by 2-10x.

5. Data Pipelines

Fabric data pipelines orchestrate data movement and transformation. They are built on the same engine as Azure Data Factory with Fabric-specific activities.

Pipeline Activities

Parameters and Expressions

Pipeline parameters (set at pipeline level, passed at trigger time):

{
  "parameters": {
    "environment": { "type": "string", "defaultValue": "prod" },
    "run_date": { "type": "string", "defaultValue": "" },
    "source_path": { "type": "string", "defaultValue": "Files/landing" }
  }
}

System variables:

Expression functions:

@utcNow()                                    # Current UTC datetime
@utcNow('yyyy-MM-dd')                        # Formatted date string
@addDays(utcNow(), -7)                       # 7 days ago
@concat('prefix_', pipeline().parameters.env) # String concatenation
@if(equals(pipeline().parameters.env, 'prod'), 'lh_prod', 'lh_dev')  # Conditional
@activity('Copy_Data').output.rowsCopied      # Output from previous activity
@activity('Lookup_Watermark').output.firstRow.last_value  # Lookup result

Triggers

Schedule trigger (cron-based recurring):

{
  "type": "ScheduleTrigger",
  "typeProperties": {
    "recurrence": {
      "frequency": "Day",
      "interval": 1,
      "startTime": "2024-01-01T06:00:00Z",
      "timeZone": "UTC",
      "schedule": {
        "hours": [6],
        "minutes": [0]
      }
    }
  }
}

Tumbling window trigger (fixed intervals with backfill support):

{
  "type": "TumblingWindowTrigger",
  "typeProperties": {
    "frequency": "Hour",
    "interval": 1,
    "startTime": "2024-01-01T00:00:00Z",
    "delay": "00:05:00",
    "maxConcurrency": 1,
    "retryPolicy": { "count": 3, "intervalInSeconds": 300 }
  }
}

Event trigger (fires when files arrive in OneLake):

• Supported event: Blob created in a lakehouse Files/ path.
• Use for real-time ingestion: file lands in Files/landing/ -> trigger fires -> pipeline processes file.

Pipeline JSON Structure via REST API

# Create a pipeline
POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/dataPipelines
Content-Type: application/json
Authorization: Bearer {token}

{
  "displayName": "pl_ingest_sales_daily",
  "definition": {
    "parts": [
      {
        "path": "pipeline-content.json",
        "payload": "<base64-encoded pipeline JSON>",
        "payloadType": "InlineBase64"
      }
    ]
  }
}

The pipeline JSON payload follows the ADF pipeline schema with activities, dependencies, parameters, and triggers.

6. Data Transformations

Common PySpark transformation patterns for Fabric data engineering.

Cleaning Patterns

from pyspark.sql.functions import col, trim, lower, regexp_replace, when, lit, coalesce

# Trim whitespace from all string columns
for c in [f.name for f in df.schema.fields if f.dataType == StringType()]:
    df = df.withColumn(c, trim(col(c)))

# Normalize email
df = df.withColumn("email", lower(trim(col("email"))))

# Remove special characters from phone
df = df.withColumn("phone", regexp_replace(col("phone"), r"[^\d+]", ""))

# Replace empty strings with null
for c in df.columns:
    df = df.withColumn(c, when(col(c) == "", None).otherwise(col(c)))

# Fill nulls with defaults
df = df.fillna({"status": "unknown", "amount": 0.0})

# Coalesce multiple columns
df = df.withColumn("full_name", coalesce(col("display_name"), concat(col("first_name"), lit(" "), col("last_name"))))

Join Patterns

from pyspark.sql.functions import broadcast

# Standard join
result = orders.join(customers, orders.customer_id == customers.id, "left")

# Broadcast join (small lookup table < 100 MB)
result = orders.join(broadcast(status_lookup), "status_code", "inner")

# Anti join (find unmatched records)
orphaned = orders.join(customers, orders.customer_id == customers.id, "left_anti")

# Cross join with filter (use sparingly)
combos = products.crossJoin(regions).filter(col("region") != "excluded")

Window Functions

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, lag, lead, sum as spark_sum, avg as spark_avg

# Running total
window_running = Window.partitionBy("customer_id").orderBy("order_date").rowsBetween(Window.unboundedPreceding, Window.currentRow)
df = df.withColumn("running_total", spark_sum("amount").over(window_running))

# Previous and next values
window_ordered = Window.partitionBy("customer_id").orderBy("order_date")
df = df.withColumn("prev_amount", lag("amount", 1).over(window_ordered))
df = df.withColumn("next_amount", lead("amount", 1).over(window_ordered))

# Moving average (7-day window)
window_moving = Window.partitionBy("product_id").orderBy("date").rowsBetween(-6, 0)
df = df.withColumn("moving_avg_7d", spark_avg("sales").over(window_moving))

# Rank and dedup
window_dedup = Window.partitionBy("id").orderBy(col("updated_at").desc())
df = df.withColumn("rn", row_number().over(window_dedup)).filter(col("rn") == 1).drop("rn")

UDFs (Use Sparingly)

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# Only use UDFs when NO built-in function exists
# UDFs disable Catalyst optimization and require serialization

@udf(returnType=StringType())
def custom_hash(val):
    """Example: custom business logic that has no built-in equivalent."""
    import hashlib
    return hashlib.sha256(str(val).encode()).hexdigest()[:12] if val else None

df = df.withColumn("short_hash", custom_hash(col("id")))

Prefer built-ins: concat(), sha2(), regexp_replace(), when(), coalesce() — these run natively in Spark's JVM and are 10-100x faster than Python UDFs.

Schema Drift Handling

from pyspark.sql.types import StructType

# Read with a defined schema — extra columns are dropped, missing columns become null
expected_schema = StructType([...])
df = spark.read.schema(expected_schema).format("csv").load(path)

# Or detect drift and log it
actual_cols = set(df.columns)
expected_cols = set([f.name for f in expected_schema.fields])
new_cols = actual_cols - expected_cols
missing_cols = expected_cols - actual_cols

if new_cols:
    print(f"WARNING: New columns detected: {new_cols}")
if missing_cols:
    print(f"WARNING: Missing columns: {missing_cols}")
    for mc in missing_cols:
        df = df.withColumn(mc, lit(None))

SCD Type 2 Pattern

from delta.tables import DeltaTable
from pyspark.sql.functions import current_timestamp, lit, sha2, concat_ws, col

# Compute hash for change detection
hash_cols = ["first_name", "last_name", "email", "segment"]
source = source_df.withColumn("_hash", sha2(concat_ws("||", *[col(c) for c in hash_cols]), 256))

target = DeltaTable.forName(spark, "silver.customers_scd2")

# Step 1: Close changed records
target.alias("t").merge(
    source.alias("s"),
    "t.customer_id = s.customer_id AND t._is_current = true"
).whenMatchedUpdate(
    condition="t._hash != s._hash",
    set={"_is_current": "false", "_valid_to": "current_timestamp()"}
).execute()

# Step 2: Insert new current records
changed_or_new = source.alias("s").join(
    target.toDF().filter(col("_is_current") == True).alias("t"),
    col("s.customer_id") == col("t.customer_id"),
    "left_anti"  # Records not in target (new or just closed)
).withColumn("_is_current", lit(True)) \
 .withColumn("_valid_from", current_timestamp()) \
 .withColumn("_valid_to", lit(None).cast("timestamp"))

changed_or_new.write.format("delta").mode("append").saveAsTable("silver.customers_scd2")

7. Medallion Architecture

The medallion architecture organizes data into three layers — Bronze, Silver, and Gold — each with a specific purpose and quality level.

Bronze Layer (Raw Ingestion)

• Purpose: Land data exactly as received from source systems.
• Quality: Raw, unvalidated, schema-on-read.
• Format: Delta tables with minimal transformation (add audit columns only).
• Retention: Keep raw data for reprocessing (30-90 days typical).
• Schema: Accept everything — use mergeSchema = true.

# Bronze ingestion pattern
df = spark.read.format("csv").option("header", "true").schema(source_schema).load("Files/landing/*.csv")
df = df.withColumn("_ingested_at", current_timestamp()) \
       .withColumn("_source_file", input_file_name()) \
       .withColumn("_batch_id", lit(batch_id))

df.write.format("delta").mode("append").option("mergeSchema", "true").saveAsTable("lh_bronze_sales.raw_orders")

Silver Layer (Cleaned and Conformed)

• Purpose: Cleaned, deduplicated, conformed data with business-level naming.
• Quality: Validated, typed, null-handled, deduplicated.
• Format: Delta tables with enforced schema.
• Transformations: Type casting, dedup, null handling, standardization, SCD2.

# Silver transformation pattern
bronze_df = spark.read.table("lh_bronze_sales.raw_orders")

silver_df = bronze_df \
    .dropDuplicates(["order_id"]) \
    .withColumn("order_date", col("order_date").cast("date")) \
    .withColumn("amount", col("amount").cast("decimal(18,2)")) \
    .withColumn("customer_email", lower(trim(col("customer_email")))) \
    .filter(col("order_id").isNotNull()) \
    .withColumn("_processed_at", current_timestamp())

silver_df.write.format("delta").mode("overwrite").saveAsTable("lh_silver_sales.orders")

Gold Layer (Business Aggregates)

• Purpose: Business-ready aggregations and metrics for Power BI and reporting.
• Quality: Fully validated, aggregated, business KPIs.
• Format: Delta tables with V-Order enabled for Power BI performance.
• Consumers: Power BI datasets, SQL analytics endpoint, ad-hoc queries.

# Gold aggregation pattern
spark.conf.set("spark.sql.parquet.vorder.enabled", "true")

silver_df = spark.read.table("lh_silver_sales.orders")

gold_df = silver_df \
    .withColumn("month", date_trunc("month", col("order_date"))) \
    .groupBy("month", "region", "product_category") \
    .agg(
        sum("amount").alias("total_revenue"),
        count("order_id").alias("order_count"),
        avg("amount").alias("avg_order_value"),
        countDistinct("customer_id").alias("unique_customers")
    )

gold_df.write.format("delta").mode("overwrite").saveAsTable("lh_gold_sales.monthly_sales_summary")

Implementation in Fabric

One lakehouse per layer (recommended for separation of concerns):

Workspace: ws_sales_analytics
├── lh_bronze_sales     # Raw ingested data
├── lh_silver_sales     # Cleaned and conformed
├── lh_gold_sales       # Business aggregates (Power BI connects here)
├── nb_ingest_bronze    # Ingestion notebook
├── nb_transform_silver # Cleaning notebook
├── nb_aggregate_gold   # Aggregation notebook
└── pl_sales_daily      # Pipeline orchestrating all three

Cross-lakehouse references in Spark:

# Read from bronze lakehouse while attached to silver
bronze_df = spark.read.table("lh_bronze_sales.raw_orders")

# Write to the current (attached) lakehouse
silver_df.write.format("delta").saveAsTable("orders")

# Or reference by full OneLake path
df = spark.read.format("delta").load(
    "abfss://ws_sales_analytics@onelake.dfs.fabric.microsoft.com/lh_bronze_sales.Lakehouse/Tables/raw_orders"
)

Cross-lakehouse references in SQL analytics endpoint:

-- Three-part naming within the same workspace
SELECT * FROM [lh_silver_sales].[dbo].[orders]
WHERE order_date >= '2024-01-01';

8. REST API

The Fabric REST API manages Fabric items (lakehouses, notebooks, pipelines) programmatically.

Authentication

# Get token via Azure CLI
TOKEN=$(az account get-access-token --resource https://api.fabric.microsoft.com --query accessToken -o tsv)

# Or via service principal (MSAL)
# POST https://login.microsoftonline.com/{tenantId}/oauth2/v2.0/token
# grant_type=client_credentials&client_id={clientId}&client_secret={clientSecret}&scope=https://api.fabric.microsoft.com/.default

Workspace Management

# List workspaces
GET https://api.fabric.microsoft.com/v1/workspaces

# Create workspace
POST https://api.fabric.microsoft.com/v1/workspaces
{ "displayName": "ws_sales_analytics", "capacityId": "{capacityId}" }

# Get workspace items
GET https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items

Item Operations (Lakehouses, Notebooks, Pipelines)

# List lakehouses
GET https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/lakehouses

# Create lakehouse
POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/lakehouses
{ "displayName": "lh_bronze_sales" }

# Get lakehouse details (includes SQL endpoint connection string)
GET https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/lakehouses/{lakehouseId}

# Create notebook
POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/notebooks
{ "displayName": "nb_ingest_csv", "definition": { "format": "ipynb", "parts": [...] } }

# Create pipeline
POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/dataPipelines
{ "displayName": "pl_daily_ingest", "definition": { "parts": [...] } }

# Delete an item
DELETE https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{itemId}

Long-Running Operations

Some operations (e.g., creating large items, running notebooks) return a 202 Accepted with a Location header:

# Poll for completion
GET https://api.fabric.microsoft.com/v1/operations/{operationId}

# Response when complete:
{ "status": "Succeeded", "createdTimeUtc": "...", "lastUpdatedTimeUtc": "..." }

Run a Pipeline

POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{pipelineId}/jobs/instances?jobType=Pipeline
{
  "executionData": {
    "parameters": {
      "run_date": "2024-01-15",
      "environment": "prod"
    }
  }
}

9. Connectivity

OneLake ADLS Gen2 Endpoint

Every Fabric item is accessible via the standard ADLS Gen2 endpoint:

abfss://<workspace-name>@onelake.dfs.fabric.microsoft.com/<item-name>.<item-type>

Examples:

# Read a Delta table from another workspace
df = spark.read.format("delta").load(
    "abfss://ws_shared_data@onelake.dfs.fabric.microsoft.com/lh_reference.Lakehouse/Tables/countries"
)

# Read raw files
df = spark.read.format("csv").load(
    "abfss://ws_sales@onelake.dfs.fabric.microsoft.com/lh_bronze.Lakehouse/Files/landing/orders.csv"
)

External access (from outside Fabric — Azure Databricks, Synapse, ADF):

• Authenticate with Azure AD (OAuth 2.0).
• Use the same abfss:// path with onelake.dfs.fabric.microsoft.com as the account name.
• No storage account keys — OneLake uses Azure AD only.

Shortcuts

Shortcuts allow a lakehouse to reference data from external sources without copying:

Create a shortcut (REST API):

POST https://api.fabric.microsoft.com/v1/workspaces/{workspaceId}/items/{lakehouseId}/shortcuts
{
  "name": "external_customers",
  "path": "Tables",
  "target": {
    "type": "AdlsGen2",
    "adlsGen2": {
      "location": "https://storageaccount.dfs.core.windows.net",
      "subpath": "/container/path/to/delta/table",
      "connectionId": "{connectionId}"
    }
  }
}

Shortcuts appear as regular tables or folders in the lakehouse. Spark reads them transparently.

Gateway Connections

For on-premises or private-network data sources, use a Fabric data gateway:

1. Install the on-premises data gateway on a machine with network access to the source.
2. Register the gateway in the Fabric portal (Settings > Manage gateways).
3. Create a connection in the Fabric workspace that uses the gateway.
4. Reference the connection in Copy activities or notebooks.

10. Performance

Spark Pool Sizing

Starter pools (default, shared):

• Auto-created per workspace.
• Pool warms up in ~30 seconds (vs. 2-3 minutes for custom pools).
• Suitable for development and small-to-medium workloads.
• Node count auto-scales based on capacity.

Custom pools (dedicated, larger):

• Created in workspace settings > Data Engineering > Spark Settings > Custom Pools.
• Choose node family (Memory Optimized, etc.) and min/max nodes.
• Use for production workloads requiring predictable performance.
• Can configure autoscale min/max.

V-Order

V-Order is a write-time optimization that reorganizes Parquet row groups for faster reads:

spark.conf.set("spark.sql.parquet.vorder.enabled", "true")

• Apply to Gold-layer tables consumed by Power BI.
• Adds ~5% write overhead for 2-10x read improvement.
• Automatically applied when writing through Power BI dataflows.

Partition Pruning

Spark automatically prunes partitions when WHERE clauses match partition columns:

# This reads only the 2024-01 partition (not the entire table)
df = spark.read.table("orders").filter(col("year_month") == "2024-01")

Verify partition pruning is working:

df.explain(True)  # Look for "PartitionFilters" in the physical plan

Predicate Pushdown

Spark pushes filter conditions down to the storage layer:

# Predicate pushdown — Spark reads only matching Parquet row groups
df = spark.read.table("customers").filter(col("segment") == "premium")

Works best with:

• Z-ORDER or liquid clustering on the filter column.
• Delta statistics (min/max per file) — maintained automatically.

Cache

# Cache a DataFrame that will be used multiple times
df = spark.read.table("large_table").filter(col("year") == 2024)
df.cache()  # Stores in executor memory
df.count()  # Triggers materialization

# Use after multiple operations on the same DataFrame
result1 = df.groupBy("region").count()
result2 = df.groupBy("category").sum("amount")

# Unpersist when done
df.unpersist()

Do NOT cache:

• DataFrames used only once.
• Very large DataFrames that exceed executor memory (causes spilling to disk).
• DataFrames in a simple read-transform-write pipeline.

Autotune

Fabric Spark includes autotune, which automatically adjusts:

• spark.sql.shuffle.partitions — auto-sizes based on data volume.
• spark.sql.files.maxPartitionBytes — auto-sizes file splits.
• Query plan optimization via adaptive query execution (AQE).

Enable with:

spark.conf.set("spark.sql.adaptive.enabled", "true")

This is enabled by default in Fabric. Avoid overriding shuffle partitions manually unless you have a specific reason.

11. Monitoring

Spark Application Monitoring

In the Fabric portal, navigate to Monitor Hub or the notebook's Spark application view:

• Stages: See task distribution, skew, and spill metrics per stage.
• SQL/DataFrame: View the logical and physical query plans.
• Storage: See data read/written per stage.
• Environment: Spark configuration and lib

…(truncated)