---

name: spark-declarative-pipelines description: "Creates, configures, and updates Databricks Lakeflow Spark Declarative Pipelines (SDP/LDP) using serverless compute. Handles streaming tables, materialized views, CDC, SCD Type 2, and Auto Loader ingestion patterns. Use when building data pipelines, working with Delta Live Tables, ingesting streaming data, implementing change data capture, or when the user mentions SDP, LDP, DLT, Lakeflow pipelines, streaming tables, or bronze/silver/gold medallion architectures."

Lakeflow Spark Declarative Pipelines (SDP)

Official Documentation

• Lakeflow Spark Declarative Pipelines Overview - Main documentation hub
• SQL Language Reference - SQL syntax for streaming tables and materialized views
• Python Language Reference - pyspark.pipelines API
• Loading Data - Auto Loader, Kafka, Kinesis ingestion
• Change Data Capture (CDC) - AUTO CDC, SCD Type 1/2
• Developing Pipelines - File structure, testing, validation
• Liquid Clustering - Modern data layout optimization

---

Development Workflow with MCP Tools

Use MCP tools to create, run, and iterate on serverless SDP pipelines. The primary tool is create_or_update_pipeline which handles the entire lifecycle.

IMPORTANT: Always create serverless pipelines (default). Only use classic clusters if user explicitly requires R language, Spark RDD APIs, or JAR libraries.

Step 1: Write Pipeline Files Locally

Create .sql or .py files in a local folder:

Copymy_pipeline/
├── bronze/
│   ├── ingest_orders.sql       # SQL (default for most cases)
│   └── ingest_events.py        # Python (for complex logic)
├── silver/
│   └── clean_orders.sql
└── gold/
    └── daily_summary.sql

SQL Example (bronze/ingest_orders.sql):

CopyCREATE OR REFRESH STREAMING TABLE bronze_orders
CLUSTER BY (order_date)
AS
SELECT
  *,
  current_timestamp() AS _ingested_at,
  _metadata.file_path AS _source_file
FROM read_files(
  '/Volumes/catalog/schema/raw/orders/',
  format => 'json',
  schemaHints => 'order_id STRING, customer_id STRING, amount DECIMAL(10,2), order_date DATE'
);

Python Example (bronze/ingest_events.py):

Copyfrom pyspark import pipelines as dp
from pyspark.sql.functions import col, current_timestamp

@dp.table(name="bronze_events", cluster_by=["event_date"])
def bronze_events():
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .load("/Volumes/catalog/schema/raw/events/")
        .withColumn("_ingested_at", current_timestamp())
        .withColumn("_source_file", col("_metadata.file_path"))
    )

Language Selection:

• Default to SQL unless user specifies Python or task requires it
• Use SQL for: Transformations, aggregations, filtering, joins (90% of use cases)
• Use Python for: Complex UDFs, external APIs, ML inference, dynamic paths
• Generate ONE language per request unless user explicitly asks for mixed pipeline

Step 2: Upload to Databricks Workspace

Copy# MCP Tool: upload_folder
upload_folder(
    local_folder="/path/to/my_pipeline",
    workspace_folder="/Workspace/Users/user@example.com/my_pipeline"
)

Step 3: Create/Update and Run Pipeline

Use create_or_update_pipeline - the main entry point. It:

1. Searches for an existing pipeline with the same name (or uses id from extra_settings)
2. Creates a new pipeline or updates the existing one
3. Optionally starts a pipeline run
4. Optionally waits for completion and returns detailed results

Copy# MCP Tool: create_or_update_pipeline
result = create_or_update_pipeline(
    name="my_orders_pipeline",
    root_path="/Workspace/Users/user@example.com/my_pipeline",
    catalog="my_catalog",
    schema="my_schema",
    workspace_file_paths=[
        "/Workspace/Users/user@example.com/my_pipeline/bronze/ingest_orders.sql",
        "/Workspace/Users/user@example.com/my_pipeline/silver/clean_orders.sql",
        "/Workspace/Users/user@example.com/my_pipeline/gold/daily_summary.sql"
    ],
    start_run=True,           # Start immediately
    wait_for_completion=True, # Wait and return final status
    full_refresh=True,        # Full refresh all tables
    timeout=1800              # 30 minute timeout
)

Result contains actionable information:

Copy{
    "success": True,                    # Did the operation succeed?
    "pipeline_id": "abc-123",           # Pipeline ID for follow-up operations
    "pipeline_name": "my_orders_pipeline",
    "created": True,                    # True if new, False if updated
    "state": "COMPLETED",               # COMPLETED, FAILED, TIMEOUT, etc.
    "catalog": "my_catalog",            # Target catalog
    "schema": "my_schema",              # Target schema
    "duration_seconds": 45.2,           # Time taken
    "message": "Pipeline created and completed successfully in 45.2s. Tables written to my_catalog.my_schema",
    "error_message": None,              # Error summary if failed
    "errors": []                        # Detailed error list if failed
}

Step 4: Handle Results

On Success:

Copyif result["success"]:
    # Verify output tables
    stats = get_table_details(
        catalog="my_catalog",
        schema="my_schema",
        table_names=["bronze_orders", "silver_orders", "gold_daily_summary"]
    )

On Failure:

Copyif not result["success"]:
    # Message includes suggested next steps
    print(result["message"])
    # "Pipeline created but run failed. State: FAILED. Error: Column 'amount' not found.
    #  Use get_pipeline_events(pipeline_id='abc-123') for full details."

    # Get detailed errors
    events = get_pipeline_events(pipeline_id=result["pipeline_id"], max_results=50)

Step 5: Iterate Until Working

1. Review errors from result or get_pipeline_events
2. Fix issues in local files
3. Re-upload with upload_folder
4. Run create_or_update_pipeline again (it will update, not recreate)
5. Repeat until result["success"] == True

---

Quick Reference: MCP Tools

Primary Tool

Pipeline Management

Supporting Tools

---

Reference Documentation (Local)

Load these for detailed patterns:

• 1-ingestion-patterns.md - Auto Loader, Kafka, Event Hub, Kinesis, file formats
• 2-streaming-patterns.md - Deduplication, windowing, stateful operations, joins
• 3-scd-patterns.md - Querying SCD Type 2 history tables, temporal joins
• 4-performance-tuning.md - Liquid Clustering, optimization, state management
• 5-python-api.md - Modern dp API vs legacy dlt API comparison
• 6-dlt-migration.md - Migrating existing DLT pipelines to SDP
• 7-advanced-configuration.md - extra_settings parameter reference and examples

---

Best Practices (2025)

Language Selection

• Default to SQL unless user specifies Python or task clearly requires it
• Use SQL for: Transformations, aggregations, filtering, joins (most cases)
• Use Python for: Complex UDFs, external APIs, ML inference, dynamic paths (use modern pyspark.pipelines as dp)
• Generate ONE language per request unless user explicitly asks for mixed pipeline

Modern Defaults

• CLUSTER BY (Liquid Clustering), not PARTITION BY - see 4-performance-tuning.md
• Raw .sql/.py files, not notebooks
• Serverless compute ONLY - Do not use classic clusters unless explicitly required
• Unity Catalog (required for serverless)
• read_files() for cloud storage ingestion - see 1-ingestion-patterns.md

---

Common Issues

For detailed errors, the result["message"] from create_or_update_pipeline includes suggested next steps. Use get_pipeline_events(pipeline_id=...) for full stack traces.

---

Advanced Pipeline Configuration

For advanced configuration options (development mode, continuous pipelines, custom clusters, notifications, Python dependencies, etc.), see 7-advanced-configuration.md.

---

Platform Constraints

Serverless Pipeline Requirements (Default)

Serverless Limitations (When Classic Clusters Required)

General Constraints

Default to serverless unless user explicitly requires R, RDD APIs, or JAR libraries.