Spark CLI

Data Engineering Patterns — Skill Resource

Essential patterns and principles for PySpark data engineering in Microsoft Fabric.

Recommended patterns

Must

1. Always define explicit schemas for production data ingestion — avoid inferSchema=true which adds overhead and inconsistency

2. Use Delta Lake format for all managed tables — provides ACID guarantees, time travel, and optimized reads

3. Validate data quality at ingestion boundaries — check nulls, data types, and business rules before persisting

4. Add metadata columns to track lineage — ingestion_timestamp, source_system, pipeline_run_id for debugging

5. Handle errors gracefully — wrap ingestion/transformation logic in try-except with proper logging and recovery

6. Use MERGE for upserts — leverage Delta Lake's MERGE INTO for incremental updates based on merge keys

7. Partition large tables — use date or category columns for partition pruning to improve query performance

8. If a real public source URL is provided, ingest from that source — download/copy into lakehouse Files/ first, then load with Spark from lakehouse paths (do not replace with synthetic inline rows)

Prefer

1. Batch processing over streaming unless real-time requirements exist — simpler to debug and monitor

2. Read-optimized writes for analytical workloads — use .coalesce() or .repartition() to right-size output files

3. Window functions over self-joins — more efficient for ranking, running totals, and lag/lead operations

4. Broadcast joins for small dimensions — use .broadcast() hint when one table fits in memory (<100MB)

5. Columnar operations over row-wise — leverage DataFrame/SQL API instead of UDFs when possible

6. Lazy evaluation mindset — build transformation chains, then execute with actions (.write(), .count())

Avoid

1. Don't use .collect() on large DataFrames — brings all data to driver, causes OOM errors

2. Don't chain multiple .count() calls — each triggers a full scan; cache DataFrame if needed

3. Don't ignore skew — salting keys or adaptive query execution prevents straggler tasks

4. Don't skip Delta optimization — run OPTIMIZE and VACUUM regularly to prevent small file problem

5. Don't hardcode paths or credentials — use parameters and secure configuration patterns

6. Don't mix append and overwrite carelessly — understand partition scope for .mode("overwrite")

Data Ingestion Principles

Schema Management

Guide LLM to define explicit schemas with nullable constraints, data type validation, and business context comments.

Note: This section refers to data schemas (DataFrame structure). For lakehouse schemas (databases/namespaces for organizing tables), see SPARK-AUTHORING-CORE.md Lakehouse Schema Organization.

Source Format Handling

• CSV/TSV: Explicit schema, header option

• Parquet/ORC: Columnar formats with embedded schema

• JSON: multiLine option for nested objects

• ADLS Gen2: abfss://[email protected]/path

• OneLake: abfss://[email protected]/lakehouse.Lakehouse/Files/path

• Public HTTP/HTTPS datasets: Download/copy to lakehouse Files/... first, then spark.read from lakehouse paths for stable runtime behavior

Validation Patterns

• Completeness: Filter nulls in required fields

• Referential integrity: Join with dimensions, flag orphans

• Business rules: Domain-specific checks (amount > 0, date ranges)

• Duplicates: dropDuplicates or groupBy to identify

Error Handling Strategy

• Try-except blocks with specific exceptions

• Contextual logging

• Dead letter queues for invalid records

• Retry logic for transient failures

Transformation Patterns

This section helps you choose the right transformation approach for your data pipeline. Each pattern follows a consistent structure: WHAT (definition), WHY (benefits/trade-offs), WHEN (decision criteria), and HOW (examples).

Pattern 1: Aggregations (GROUP BY + Aggregate Functions)

WHAT:
Summarize data by grouping rows and applying aggregate functions (SUM, COUNT, AVG, MIN, MAX, COLLECT_LIST).

WHY:

• ✅ Single-pass computation (efficient)

• ✅ Built-in Spark optimization (adaptive query execution)

• ✅ Can combine multiple aggregates in one .agg() call

• ⚠️ Large cardinality GROUP BY (millions of unique keys) may cause memory pressure

WHEN to use:

• Summarizing metrics by dimension (e.g., revenue by region, counts by product)

• Calculating totals, averages, or other statistical measures

• Creating reporting tables or dashboards

• Preparing data for visualization

Examples:

from pyspark.sql import functions as F

# Daily sales summary
df.groupBy("order_date") \
  .agg(
    F.sum("amount").alias("total_sales"),
    F.count("order_id").alias("order_count"),
    F.countDistinct("customer_id").alias("unique_customers")
  )

# Multi-dimensional aggregation
df.groupBy("product_category", "region", "order_date") \
  .agg(F.avg("discount_percent").alias("avg_discount"))

Alternative: If aggregation + periodic refresh → consider Materialized Lake View (see Pattern 4 below).

Pattern 2: Window Functions (Partitioned Analytics)

WHAT:
Perform calculations across rows related to the current row without collapsing them into groups. Common functions: ROW_NUMBER(), RANK(), LAG(), LEAD(), SUM() OVER (...), AVG() OVER (...).

WHY:

• ✅ More efficient than self-joins for ranking/running totals

• ✅ Preserves row-level detail while adding analytical columns

• ✅ Supports complex ordering and partitioning logic

• ⚠️ Large partitions can cause shuffle overhead (consider repartitioning by partition key first)

WHEN to use:

• Ranking: Top N per category, percentile ranks

• Running calculations: Cumulative sums, moving averages

• Lag/Lead comparisons: Month-over-month change, previous transaction

• Deduplication: ROW_NUMBER() + filter(rank == 1) to pick most recent (note: QUALIFY and window functions block MLV incremental refresh — keep window functions OUT of MLV; apply ranking in downstream notebook for IR-eligible pipelines. See mlv-incremental-refresh-patterns.md § Pattern 1)

Examples:

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# Top 3 products per category by sales
window_spec = Window.partitionBy("category").orderBy(F.desc("sales"))
df.withColumn("rank", F.row_number().over(window_spec)) \
  .filter(F.col("rank") <= 3)

# Running total and previous value
window_running = Window.partitionBy("customer_id").orderBy("order_date") \
                       .rowsBetween(Window.unboundedPreceding, Window.currentRow)
df.withColumn("cumulative_spent", F.sum("amount").over(window_running)) \
  .withColumn("previous_order", F.lag("order_date", 1).over(window_running))

Alternative: Self-joins are less efficient but sometimes unavoidable for complex multi-step logic.

Pattern 3: Joins (Combining Datasets)

WHAT:
Merge two DataFrames based on a common key. Types: inner, left (left outer), right, full (full outer), cross, left_semi, left_anti.

WHY:

• ✅ Standard relational operation for combining data

• ✅ Broadcast joins (<100MB dimension table) avoid shuffle — very fast

• ⚠️ Large-to-large joins can cause shuffle and spill to disk

• ⚠️ Skewed keys (one key with millions of rows) create straggler tasks

WHEN to use:

Examples:

# Inner join: transactions + customer details
transactions.join(customers, on="customer_id", how="inner")

# Broadcast join: large fact table + small dimension
from pyspark.sql.functions import broadcast
fact_table.join(broadcast(dim_table), on="product_id")

# Left join: keep all orders, add customer name if exists
orders.join(customers, on="customer_id", how="left") \
      .select("order_id", "amount", "customer_name")

Tips:

• For skewed joins, use salting (add random suffix to key)

• Use df.explain() to verify broadcast join actually happened

• Consider pre-filtering both DataFrames before joining to reduce shuffle

Pattern 4: Materialized Lake Views (Declarative Transformations)

WHAT:
A Fabric-native feature that materializes transformation results as Delta tables with automatic refresh, data quality constraints, and query optimization. Two authoring modes: Spark SQL (with incremental refresh) or PySpark (full refresh only).

WHY:

• ✅ Declarative: Define WHAT to compute, not HOW (Fabric handles refresh scheduling)

• ✅ Auto-refresh: Spark SQL MLVs support incremental refresh; PySpark MLVs use full refresh

• ✅ Query-optimized: Pre-aggregated, indexed, faster than querying raw tables

• ✅ Data quality: Built-in CHECK constraints with ON MISMATCH DROP|FAIL

• ⚠️ Spark SQL MLVs: No Python UDFs, but support incremental refresh

• ⚠️ PySpark MLVs: Support UDFs and complex logic, but full refresh only (no incremental)

• ⚠️ Batch-only: Not suitable for real-time streaming

• ⚠️ Incremental limitations: Some SQL patterns block incremental refresh (see mlv-incremental-refresh-patterns.md)

WHEN to use:

Use this decision tree to choose between MLVs and PySpark notebooks:

┌─ Is transformation logic pure SQL (SELECT/WHERE/JOIN/GROUP BY)?
│
├─ YES ─┐
│       ├─ Does it need periodic refresh (not real-time)?
│       │
│       ├─ YES ─┐
│       │       ├─ Would incremental refresh save cost (source is append-only + CDF enabled + query uses supported SQL constructs)?
│       │       │
│       │       ├─ YES → ✅ **Use Materialized Lake View**
│       │       │          (Declarative, auto-incremental, query-optimized)
│       │       │
│       │       └─ NO  → ⚠️  **MLV or Notebook**
│       │                   (Small tables: MLV. Large full-refresh: consider notebook with optimizations)
│       │
│       └─ NO (real-time) → ❌ **Use Structured Streaming Notebook/Job Definition + Eventstream**
│                               (MLVs are batch-only; use Spark Structured Streaming for real-time ingestion)
│
└─ NO (Python/complex logic) → ❌ **Use PySpark Notebook**
                                   (UDFs, iterative algorithms, external APIs, custom validation)

Concrete examples:

Cost/performance considerations:

When to switch from MLV to Notebook:

• MLV hits SQL limitations (no UDFs, complex Python logic needed)

• Incremental refresh blockers appear (see mlv-incremental-refresh-patterns.md)

• Debugging requires inspecting intermediate steps (MLV is declarative, no step-by-step visibility)

When to switch from Notebook to MLV:

• Notebook logic simplifies to pure SQL after refactoring

• Manual watermark/incremental logic becomes complex — let MLV handle it

• Transformation is stable and needs less frequent iteration

See also:

• Materialized Lake View patterns — design patterns, scheduling, when to use vs Delta tables

• MLV incremental refresh patterns — IR blocker catalog, safe rewrites, CDF prerequisites

• the mlv mode — schedule, trigger, monitor, and cancel MLV refreshes via REST API (use when user asks to automate refresh, not author SQL)

Quick Decision Matrix

Example Approaches

Customer Segmentation: Use window functions for lifetime metrics, when().otherwise() for classification, temporal dimensions for recency

Product Analytics: Join with dimensions, aggregate by category, rank with row_number(), compute percentiles

Delta Lake Best Practices

MERGE Operations (Upserts)

When to use:

• Incremental loads where source sends changed/new records

• Slowly changing dimensions (SCD Type 1 or Type 2)

• Deduplication scenarios

Guide LLM to generate MERGE with:

• .merge(source_df, "target.id = source.id") on unique key

• .whenMatchedUpdateAll() to update existing records

• .whenNotMatchedInsertAll() to insert new records

• Optional: .whenMatchedDelete() for hard deletes based on condition

Optimization Strategies

Tell LLM to include:

• Liquid Clustering (preferred over Z-Order and partitioning in Runtime 2.0+): Declarative, change-friendly clustering that replaces static partitioning and manual Z-Order maintenance

  • Create: CREATE TABLE t (...) CLUSTER BY (col1, col2) or ALTER TABLE t CLUSTER BY (col1, col2)

  • Apply: OPTIMIZE table_name (columns stored in metadata, no need to specify)

  • Full recluster: OPTIMIZE table_name FULL (rebuilds all Z-Cubes)

  • Change columns anytime: ALTER TABLE t CLUSTER BY (new_col) — applies on next OPTIMIZE

  • Remove: ALTER TABLE t CLUSTER BY NONE

  • Runtime 2.0 incremental mode: Only reclusters unclustered/unhealthy/small files (eliminates full-rewrite amplification)

  • Pick 1–4 columns from frequent WHERE clauses; Hilbert curve for 2+ columns

  • CAUTION (Runtime 1.3): Full Z-Cube rewrite every run — do NOT combine with auto compaction

• Z-Ordering (legacy/Runtime 1.3): OPTIMIZE table_name ZORDER BY (frequently_filtered_column) — use only when liquid clustering is unavailable

• VACUUM: VACUUM table_name RETAIN 168 HOURS cleans up old file versions after retention period

• Partition pruning: Query with partition columns in WHERE clause to skip irrelevant data

• File compaction: Run OPTIMIZE to combine small files into right-sized files (128MB-1GB)

Time Travel

Use cases:

• Point-in-time queries: spark.read.format("delta").option("versionAsOf", 5).load(path)

• Rollback bad writes: Restore to previous version with RESTORE TABLE table_name TO VERSION AS OF 10

• Audit trail: Query historical data for compliance, debugging

Spark Session Configurations for Performance

Guide LLM to configure Spark sessions based on workload type:

Write-Heavy Workloads (Bronze Layer - High-Volume Ingestion):

• spark.microsoft.delta.parquet.vorder.enabled = false — Disable V-Order for faster writes

• spark.databricks.delta.optimizeWrite.binSize = 1073741824 — Target 1GB file size for fewer small files

• spark.databricks.delta.autoCompact.enabled = true — Automatic compaction during writes

• spark.microsoft.delta.optimize.fast.enabled = true — Fast optimization algorithms

• spark.databricks.delta.properties.defaults.enableDeletionVectors = true — Efficient delete tracking

• spark.microsoft.delta.targetFileSize.adaptive.enabled = true — Adaptive file sizing

• spark.native.enabled = true — Use native execution engine (Velox)

• spark.gluten.delta.columnMapping.name.enabled = true — Column mapping for schema evolution

Balanced Workloads (Silver Layer - Mixed Read/Write):

• spark.microsoft.delta.parquet.vorder.enabled = true — Enable V-Order for better read performance

• spark.databricks.delta.optimizeWrite.enabled = true — Balance write optimization with read efficiency

• spark.microsoft.delta.snapshot.driverMode.enabled = true — Faster snapshot reads

• spark.sql.adaptive.enabled = true — Adaptive query execution

• spark.sql.adaptive.coalescePartitions.enabled = true — Dynamic partition coalescing

Read-Heavy Workloads (Gold Layer - Analytics & Reporting):

• spark.microsoft.delta.parquet.vorder.enabled = true — V-Order for maximum read performance

• spark.databricks.delta.optimizeWrite.enabled = false — No write optimization overhead

• spark.sql.parquet.enableVectorizedReader = true — Vectorized Parquet reads

• spark.sql.files.maxPartitionBytes = 134217728 — 128MB partition size for optimal parallelism

• spark.sql.adaptive.enabled = true — Optimize query plans based on runtime stats

• spark.databricks.delta.stalenessLimit = 0 — Always use latest snapshot

When to apply these configs:

• Pass during Livy session creation: "conf": {"spark.config.key": "value"}

• Set in notebook first cell before any Spark operations

• Configure at workspace level for consistent defaults

• Override per-job for specific workload requirements

Runtime 2.0 Performance Features (Spark 4.1, Delta 4.2)

⚠️ Public Preview: Fabric Runtime 2.0 (Spark 4.1, Delta Lake 4.x) is in Public Preview. Delta Lake 4.x features are experimental and should not be enabled for tables that interoperate across Fabric workloads. Only recommend the features below once the customer has explicitly accepted preview risk and validated on a non-production workspace — do not present them as default production guidance.

Native Execution Engine (NEE): Significant performance improvement over open-source Spark via vectorized C++ execution. See NEE documentation for current benchmarks.

NEE must be enabled before the session starts — either via the environment-level Acceleration setting or with %%configure in the notebook's first cell. A post-start spark.conf.set("spark.native.enabled", ...) does not reliably enable it.

%%configure
{
    "driverMemory": "56g",
    "executorMemory": "56g",
    "conf": {
        "spark.native.enabled": "true"
    }
}

Keep Livy resource fields such as driverMemory and executorMemory at the top level. Use conf only for Spark properties.

# Supported automatically once enabled:
# - Python UDFs (scalar and vectorized/Pandas UDFs)
# - Scala UDFs
# - Complex types: arrays, maps, structs (explode, array_contains, map_keys, etc.)
# - Vectorized CSV parsing
# - OPTIMIZE on liquid clustered tables

Efficient Scaledown (Remote Shuffle Manager): Decouples shuffle data from executor lifetime for improved compute efficiency. Requires NEE enabled (set at session start, above). Set the full stack at session start via %%configure so it applies before Spark initializes. See RSM documentation for the complete option list and benchmarks.

%%configure
{
    "conf": {
        "spark.remote.shuffle.enabled": "true",
        "spark.sql.rsm.decisionlayer.enabled.level": "stage",
        "spark.sql.adaptive.shuffleWrite.enabled": "true",
        "spark.storage.decommission.shuffleBlocks.enabled": "true",
        "spark.storage.decommission.shuffleBlocks.cleanup": "true",
        "spark.storage.decommission.shuffleBlocks.migrateToFallbackStorage": "true",
        "spark.storage.decommission.fallbackStorage.cleanUp": "true",
        "spark.dynamicAllocation.preventShutdownExecutorWithCache": "false",
        "spark.dynamicAllocation.excludeDeltaSnapshotCache": "true"
    }
}

Key benefits:

• Large shuffles → Azure Blob Storage (fault-tolerant, eliminates FetchFailedException)

• Small shuffles → local disk (low latency), migrated on decommission

• Autoscaler releases nodes immediately after task completion

• Prerequisites: NEE enabled + Autoscale recommended

Incremental Liquid Clustering (Runtime 2.0+ only):

Incremental clustering and auto-reclustering are enabled by default in Runtime 2.0, so you normally don't set anything — OPTIMIZE reclusters only new, changed, or unhealthy files instead of rewriting whole Z-Cubes. The documented session toggles (both default true) are:

# Master switch — set to False only to revert to full-rewrite behavior
spark.conf.set("spark.microsoft.delta.optimize.clustering.strategy.incremental", "true")
# Auto-detect and recluster overlapping files (applies only when incremental is on)
spark.conf.set("spark.microsoft.delta.optimize.clustering.strategy.incremental.autoRecluster", "true")

See Liquid clustering — Configuration reference for the full set of tuning options.

Runtime 2.0 Key Components:

• Apache Spark 4.1, Delta Lake 4.2, Python 3.13, Java 21, Scala 2.13

• PySpark native plotting API, Python Data Source API, Python UDTFs

• Structured Streaming Arbitrary State API v2

• Breaking change: Python upgrade requires re-publishing all Environments with libraries

• Deprecated: WASB protocol for GPv2 accounts — use ABFS instead

• Deprecated: SparkR (may be removed in future versions)

Quality Assurance Strategies

Testing Levels

Guide LLM to implement:

Unit Testing (local Spark):

• Test transformation logic with small sample DataFrames

• Use pytest fixtures to create test Spark session

• Assert row counts, column values, schema correctness

• Focus on business logic in isolation

Integration Testing (Fabric API):

• Validate workspace/lakehouse creation succeeded

• Test notebook deployment via REST API

• Verify Livy session creation and code execution

• Check end-to-end data flow through bronze → silver → gold

Data Quality Checks (production):

• Row count validation: compare source vs target

• Schema validation: ensure expected columns exist with correct types

• Null checks: flag unexpected nulls in required fields

• Range checks: validate numeric values within expected bounds

• Freshness checks: ensure data updated within SLA timeframe

Quality Gates

Define when pipelines should fail:

• Critical failures: schema mismatch, zero rows ingested, primary key violations

• Warnings: elevated null rate, data volume anomaly (>20% change), late arrival

• Monitoring: track ingestion lag, transformation duration, error rates over time

Logging and Observability

Prompt LLM to generate:

• Structured logging: JSON-formatted logs with timestamp, severity, context

• Metrics emission: log key counts (rows processed, errors, duration) for monitoring

• Error context: capture input values, stack traces, environment details for debugging