Databricks Pipelines

Lakeflow Spark Declarative Pipelines Development

FIRST: Use the parent databricks-core skill for CLI basics, authentication, profile selection, and data discovery commands.

Decision Tree

Use this tree to determine which dataset type and features to use. Multiple features can apply to the same dataset — e.g., a Streaming Table can use Auto Loader for ingestion, Append Flows for fan-in, and Expectations for data quality. Choose the dataset type first, then layer on applicable features.

User request → What kind of output?
├── Intermediate/reusable logic (not persisted) → Temporary View
│   ├── Preprocessing/filtering before Auto CDC → Temporary View feeding CDC flow
│   ├── Shared intermediate streaming logic reused by multiple downstream tables
│   ├── Pipeline-private helper logic (not published to catalog)
│   └── Published to UC for external queries → Persistent View (SQL only)
├── Persisted dataset
│   ├── Source is streaming/incremental/continuously growing → Streaming Table
│   │   ├── File ingestion (cloud storage, Volumes) → Auto Loader
│   │   ├── Message bus (Kafka, Kinesis, Pub/Sub, Pulsar, Event Hubs) → streaming source read
│   │   ├── Existing streaming/Delta table → streaming read from table
│   │   ├── CDC / upserts / track changes / keep latest per key / SCD Type 1 or 2 → Auto CDC
│   │   ├── Multiple sources into one table → Append Flows (NOT union)
│   │   ├── Historical backfill + live stream → one-time Append Flow + regular flow
│   │   └── Windowed aggregation with watermark → stateful streaming
│   └── Source is batch/historical/full scan → Materialized View
│       ├── Aggregation/join across full dataset (GROUP BY, SUM, COUNT, etc.)
│       ├── Gold layer aggregation from streaming table → MV with batch read (spark.read / no STREAM)
│       ├── JDBC/Federation/external batch sources
│       └── Small static file load (reference data, no streaming read)
├── Output to external system (Python only) → Sink
│   ├── Existing external table not managed by this pipeline → Sink with format="delta"
│   │   (prefer fully-qualified dataset names if the pipeline should own the table — see Publishing Modes)
│   ├── Kafka / Event Hubs → Sink with format="kafka" + @dp.append_flow(target="sink_name")
│   ├── Custom destination not natively supported → Sink with custom format
│   ├── Custom merge/upsert logic per batch → ForEachBatch Sink (Public Preview)
│   └── Multiple destinations per batch → ForEachBatch Sink (Public Preview)
└── Data quality constraints → Expectations (on any dataset type)

Common Traps

• Names → SDP = LDP = Lakeflow Declarative Pipelines = (formerly) DLT. All interchangeable when the user mentions them.
• "Create a table" without specifying type → ask whether the source is streaming or batch. Streaming source → Streaming Table; batch source → Materialized View. Mismatched pairs error at validation.
• Aggregation over a streaming source → use a Materialized View with a batch read (spark.read.table / SELECT FROM without STREAM). STs are append-only and don't recompute aggregates when source rows change; MVs do.
• Intermediate logic → default to a Temporary View. Even for shared logic reused by multiple downstream tables. Use a Private MV/ST (private=True / CREATE PRIVATE ...) only when materializing once saves significant reprocessing. For preprocessing before Auto CDC, the temp view is required — the CDC flow reads from STREAM(view_name) (SQL) or spark.readStream.table("view_name") (Python).
• Union of streams → use multiple Append Flows. UNION across streaming sources is an anti-pattern.
• Changing dataset type → cannot change ST→MV or MV→ST in place. Full refresh does NOT help. Drop the existing table manually or rename the new dataset.
• `CREATE OR REFRESH` vs `CREATE` → both parse for SQL datasets, but CREATE OR REFRESH is the idiomatic convention. For PRIVATE datasets: CREATE OR REFRESH PRIVATE STREAMING TABLE / ... MATERIALIZED VIEW.
• Kafka/Event Hubs sink serialization → the value column is mandatory; serialize the row with to_json(struct(*)) AS value. See sink-python.md.
• Multi-column Auto CDC sequencing → SQL: SEQUENCE BY STRUCT(col1, col2). Python: sequence_by=struct("col1", "col2"). See the auto-cdc references.
• Auto CDC TRUNCATE (SCD Type 1 only) → SQL: APPLY AS TRUNCATE WHEN condition. Python: apply_as_truncates=expr("condition"). Do NOT claim truncate is unsupported.
• Python-only features → Sinks, ForEachBatch Sinks, CDC from snapshots, and custom data sources are Python-only. When the user is working in SQL, clarify this and suggest switching to Python.
• Recommend ONE clear approach → present a single recommended path. Don't list anti-patterns or inferior alternatives — they confuse. Only mention alternatives when they genuinely offer different trade-offs.

Common Issues

Error → cause/fix mappings agents hit constantly. For DAB-bundle vs CLI-iteration deploy issues, see the workflow-specific reference files.

Publishing Modes

Pipelines use a default catalog and schema configured in the pipeline settings. All datasets are published there unless overridden.

• Fully-qualified names: Use catalog.schema.table in the dataset name to write to a different catalog/schema than the pipeline default. The pipeline creates the dataset there directly — no Sink needed.
• USE CATALOG / USE SCHEMA: SQL commands that change the current catalog/schema for all subsequent definitions in the same file.
• LIVE prefix: Deprecated. Ignored in the default publishing mode.
• When reading or defining datasets within the pipeline, use the dataset name only — do NOT use fully-qualified names unless the pipeline already does so or the user explicitly requests a different target catalog/schema.

API Reference

Before writing pipeline code for any feature, read the linked reference file. Each table below maps the feature to the exact API and to the detail file for that (feature, language).

Some features sit on top of others — read both:

• Auto Loader / Auto CDC / Sinks target a streaming table → also read streaming-table-python.md / streaming-table-sql.md.
• Expectations attach to a dataset → also read the dataset definition file (streaming-table / materialized-view / temporary-view).

Dataset Definition APIs

Flow and Sink APIs

CDC APIs

For querying SCD Type 2 history tables (__START_AT / __END_AT, point-in-time, joining facts with historical dimensions), see scd-2-querying.md.

Data Quality APIs

Reading Data APIs

Table/Schema Feature APIs

Legacy DLT Syntax — always migrate

The tables above show only the modern API. If you see any of the following in user code, it is the legacy DLT syntax — always migrate to the modern form, do not extend it. Read references/dlt-migration.md before suggesting changes so the conversion is correct (especially around apply_changes → create_auto_cdc_flow semantics and partition_cols → cluster_by).

Language Selection (Python vs SQL)

Decide before scaffolding — the choice picks template files (.py vs .sql) and which reference docs apply. Both can coexist, but pick a primary. When unsure, default to SQL for simplicity.

If ambiguous, ask. Stick with the chosen language unless the user explicitly switches.

Choose Your Workflow

Three project shapes exist — pick before scaffolding. Default to A for production-bound work and C for exploration / demo scaffolding.

• A: Standalone new pipeline project (DAB) — pipeline IS the project, no existing databricks.yml. Scaffold with databricks pipelines init --output-dir . --config-file init-config.json. → 1-project-initialization-with-dab.md
• B: Pipeline in an existing bundle (DAB) — databricks.yml already exists. Add a resources/<name>.pipeline.yml pointing at src/. → 1-project-initialization-with-dab.md#workflow-b-pipeline-in-existing-bundle
• C: Rapid CLI iteration (no bundle) — prototyping. databricks pipelines create / start-update / list-pipeline-events; formalise into a bundle later if the work goes to production. → 2-rapid-iteration-with-cli.md

Pipeline Structure

• Follow the medallion pattern (Bronze → Silver → Gold) unless the user says otherwise. Keep it simple by default — just a few tables.
• One dataset per file, named after the dataset. Transformation files live in src/ or transformations/.
• Gold layer: preserve key business dimensions. When aggregating into Gold, keep the dimensions analysts will filter / slice by (location, department, product line, customer segment, time period). Over-aggregating loses information that can't be recovered downstream. If a dashboard is mentioned, every filter on it needs to be a column in the Gold table. Easier to aggregate further in queries than to recover lost dimensions.

Running a Pipeline

Picking the right run command depends on the workflow chosen above.

• Workflow A / B (DAB) — Code changes only take effect after databricks bundle deploy. Always deploy before any run, dry run, or selective refresh.

  databricks bundle validate --profile <profile>
  databricks bundle deploy -t dev --profile <profile>
  databricks bundle run <pipeline_name> -t dev --profile <profile>
  databricks pipelines get <pipeline_id> --profile <profile>      # status

→ Full DAB run + iteration details: references/1-project-initialization-with-dab.md#running-a-pipeline-workflow-a--b

• Workflow C (CLI, no bundle) — Upload files to the workspace, then drive the pipeline directly. Re-upload after every code change.

  databricks workspace import-dir ./my_pipeline /Workspace/Users/<user>/my_pipeline --overwrite
  databricks pipelines start-update <pipeline_id>

→ Full CLI run + polling pattern: references/2-rapid-iteration-with-cli.md

Refresh modes (both workflows):

• Selective refresh is preferred when you only need to run one table. Dependencies must already be materialized.
• Full refresh is the most expensive and dangerous option and can lead to data loss (it reprocesses streaming sources from scratch, destroying streaming state). Use only when really necessary. Always suggest it as a follow-up the user must explicitly approve.

Always poll the update, not top-level pipeline state — see the polling rationale in 2-rapid-iteration-with-cli.md#step-4-start-an-update-and-poll-that-update. Same rule applies to bundle runs.

Reference Index

Project & lifecycle:

• 1-project-initialization-with-dab.md — Workflows A and B.
• 2-rapid-iteration-with-cli.md — Workflow C; start-update + polling + error-extraction.
• pipeline-configuration.md — Full create/update JSON reference + variant snippets + multi-schema + platform constraints.
• performance.md — Liquid Clustering, state management, joins, pre-aggregation, monitoring.
• dlt-migration.md — DLT → SDP conversions.

Cross-cutting patterns:

• streaming-patterns.md — Dedup, windowed aggregations, late data, rescue-data quarantine, anomaly detection, lag monitoring.
• scd-2-querying.md — Current-state, point-in-time, joining facts with historical dims.
• kafka.md — Kafka / Event Hubs ingestion.

Auto Loader format-specific options: JSON · CSV · XML · Parquet · Avro · Text · ORC.

Dataset, flow, CDC, expectation, Auto Loader, and sink references are listed per (feature, language) in the API Reference tables above.

Project Initialization with DAB

Two DAB-based workflows for creating Spark Declarative Pipelines:

• Workflow A: Standalone new project (the pipeline is the project).
• Workflow B: Adding a pipeline to an existing bundle (the pipeline is part of a larger app + jobs + dashboards).

For prototyping without a bundle, see 2-rapid-iteration-with-cli.md.

---

Workflow A: Standalone Bundle (pipelines init)

Use when the user wants a new project where the pipeline is the project (no existing databricks.yml).

Non-interactive (recommended for agents)

databricks pipelines init --output-dir . --config-file init-config.json

init-config.json:

{
  "project_name": "customer_pipeline",
  "initial_catalog": "prod_catalog",
  "use_personal_schema": "no",
  "initial_language": "sql"
}

Interactive

databricks pipelines init --output-dir .

Prompts for the same fields.

Alternative: databricks bundle init lakeflow-pipelines

The older template-based scaffolding also works:

databricks bundle init lakeflow-pipelines \
  --config-file <(echo '{"project_name": "my_pipeline", "language": "python", "serverless": "yes"}') \
  --profile <PROFILE> < /dev/null

Both produce DAB-shaped projects; pipelines init is the newer, more focused command.

Generated structure

project_root/
├── databricks.yml                     # Bundle config
├── pyproject.toml                     # Python only
├── resources/
│   ├── <name>_etl.pipeline.yml        # Pipeline resource
│   └── sample_job.job.yml             # Optional scheduled job
└── src/
    └── <name>_etl/
        ├── explorations/              # Ad-hoc notebooks (NOT pipeline code)
        └── transformations/           # Pipeline transformations
            ├── sample_*.sql           # or .py
            └── ...

Key rule: Pipeline transformations are raw .sql / .py files. Notebooks go in explorations/ for ad-hoc work only.

Customize and deploy

1. Replace sample_* files in transformations/ with real datasets (1 dataset per file). 2. Edit databricks.yml to set per-target catalog/schema variables and workspace host. 3. Edit resources/<name>_etl.pipeline.yml for pipeline-level settings (serverless on by default). 4. databricks bundle validate → databricks bundle deploy [-t <target>] → databricks bundle run <pipeline_name>.

databricks.yml essentials

bundle:
  name: customer_pipeline

include:
  - resources/*.yml
  - resources/*/*.yml

variables:
  catalog: { description: The catalog to use }
  schema:  { description: The schema to use }

targets:
  dev:
    mode: development           # prefixes resources with [dev <user>], pauses schedules
    default: true
    workspace:
      host: https://<workspace>.cloud.databricks.com
    variables:
      catalog: dev_catalog
      schema: ${workspace.current_user.short_name}

  prod:
    mode: production            # no prefix, schedules active
    workspace:
      host: https://<workspace>.cloud.databricks.com
      root_path: /Workspace/Users/<owner>/.bundle/${bundle.name}/${bundle.target}
    variables:
      catalog: prod_catalog
      schema: production
    permissions:
      - user_name: <owner>
        level: CAN_MANAGE

Pipeline resource (resources/<name>.pipeline.yml)

resources:
  pipelines:
    customer_pipeline_etl:
      name: customer_pipeline_etl
      catalog: ${var.catalog}
      schema: ${var.schema}
      serverless: true
      continuous: false          # explicit — true auto-retries failed updates forever
      root_path: "../src/customer_pipeline_etl"
      libraries:
        - glob:
            include: ../src/customer_pipeline_etl/transformations/**
      environment:                # serverless Python deps (optional)
        dependencies:
          - --editable ${workspace.file_path}

Scheduling Pipelines

To schedule a pipeline, add a job that triggers it in resources/<name>.job.yml:

resources:
  jobs:
    my_pipeline_job:
      trigger:
        periodic:
          interval: 1
          unit: DAYS
      tasks:
        - task_key: refresh_pipeline
          pipeline_task:
            pipeline_id: ${resources.pipelines.my_pipeline.id}

Python project dependencies

Python projects ship a standard pyproject.toml. Runtime deps in [project].dependencies, dev-only in [project.optional-dependencies].dev (e.g. databricks-connect>=15.4,<15.5, pytest, ruff). The --editable ${workspace.file_path} line in the pipeline resource installs the package on serverless compute at deploy time.

Multi-environment workflow

databricks bundle deploy                          # dev (default target) — resources prefixed [dev <user>]
databricks bundle deploy --target prod            # prod — no prefix, schedules active
databricks bundle run customer_pipeline_etl [--target prod]

---

Workflow B: Pipeline in Existing Bundle

Use when databricks.yml already exists for a larger project (app + jobs + dashboards) and a pipeline is being added to it.

Step 1: Add a pipeline resource file

resources/my_pipeline.pipeline.yml:

resources:
  pipelines:
    my_pipeline:
      name: my_pipeline
      catalog: ${var.catalog}
      schema: ${var.schema}
      serverless: true
      continuous: false
      libraries:
        - glob:
            include: ../src/pipelines/my_pipeline/**

Step 2: Add source files

src/pipelines/my_pipeline/
├── bronze_ingest.sql
├── silver_clean.sql
└── gold_summary.sql

Step 3: Deploy

databricks bundle deploy
databricks bundle run my_pipeline

The pipeline picks up the bundle's existing targets / variables / permissions.

---

Running a Pipeline (Workflow A / B)

You must deploy before running. In local development, code changes only take effect after databricks bundle deploy. Always deploy before any run, dry run, or selective refresh.

Development workflow

# 1. Validate the bundle config
databricks bundle validate --profile <profile>

# 2. Deploy to a target (dev is default)
databricks bundle deploy -t dev --profile <profile>

# 3. Trigger the pipeline
databricks bundle run <pipeline_name> -t dev --profile <profile>

# 4. Check status (capture the update_id from step 3 and poll it — not top-level state)
databricks pipelines get <pipeline_id> --profile <profile>
databricks pipelines get-update <pipeline_id> <update_id> --profile <profile>

For the rationale on polling the update (not the pipeline) and the FAILED-extraction jq pattern, see 2-rapid-iteration-with-cli.md#step-4-start-an-update-and-poll-that-update. It applies to bundle runs too.

Refresh modes

• Selective refresh is preferred when only one table needs to run. Dependencies must already be materialized.
• Full refresh is the most expensive option and can lead to data loss — it reprocesses streaming sources from scratch and destroys streaming state. Use only when necessary, and always surface it as a follow-up the user must explicitly approve. CLI: databricks bundle run <pipeline_name> --full-refresh-all or --refresh <table> for selective.

Editing pipeline code

Edit .sql / .py files under src/, then re-run databricks bundle deploy + databricks bundle run. Bundle deploy uploads changed files as raw FILE entries. Don't mix databricks workspace import --format SOURCE into a bundle-managed pipeline — that creates a NOTEBOOK entry and subsequent bundle deploys fail with type mismatch (asked: FILE, actual: NOTEBOOK).

---

Migrating from a Manual Folder Structure

If the user already has bronze/, silver/, gold/ folders without a bundle, migrate to Workflow A by wrapping them in a databricks.yml and a pipeline resource pointing at the existing folders via a glob. No file moves required — the medallion folders work as-is under transformations/**.

For detailed pipeline configuration options (development mode, continuous, custom event log, notifications, Python deps, classic clusters), see pipeline-configuration.md.

---

Common Initialization Issues

Rapid Iteration with CLI (no DAB)

Use the databricks pipelines CLI to create, run, and iterate on a pipeline without managing a bundle. Fastest path for prototyping. Production-bound work belongs in a bundle — see 1-project-initialization-with-dab.md.

Default to serverless. Only use classic clusters if the user explicitly requires R, Spark RDD APIs, or JAR libraries.

---

Step 1: Write pipeline files locally

.sql or .py files in a folder. See python-basics.md or sql-basics.md for syntax.

Step 2: Upload to the workspace

databricks workspace import-dir ./my_pipeline /Workspace/Users/<user>/my_pipeline

Re-upload with --overwrite after every code change.

Step 3: Create the pipeline

databricks pipelines create --json '{
  "name": "my_pipeline",
  "catalog": "my_catalog",
  "schema": "my_schema",
  "serverless": true,
  "continuous": false,
  "development": true,
  "channel": "PREVIEW",
  "configuration": {
    "pipelines.numUpdateRetryAttempts": "0",
    "pipelines.maxFlowRetryAttempts": "0"
  },
  "libraries": [{"glob": {"include": "/Workspace/Users/<user>/my_pipeline/**"}}]
}'

These flags are the canonical dev/iteration defaults — fail fast. Tuned for demo / iteration. For production pipelines, drop "development" and the two pipelines.*RetryAttempts overrides so the platform's retry defaults (5 / 2) can absorb transient infra failures. Per-field rationale in pipeline-configuration.md#canonical-create-dev--iteration-defaults.

libraries: use "glob" for a directory (recommended for medallion folders), "file" for a single .sql/.py (folder paths fail with Paths must end with .py or .sql), or enumerated "file" entries when ordering matters. "notebook" is deprecated — never use.

"libraries": [
  {"file": {"path": "/Workspace/.../bronze/ingest_orders.sql"}},
  {"file": {"path": "/Workspace/.../silver/clean_orders.sql"}}
]

Capture the returned pipeline_id.

Step 4: Start an update and poll that update

UPDATE_ID=$(databricks pipelines start-update <pipeline_id> | jq -r .update_id)
# Or with full refresh (destructive on streaming state — omit for incremental):
# UPDATE_ID=$(databricks pipelines start-update <pipeline_id> --full-refresh | jq -r .update_id)

while :; do
  STATE=$(databricks pipelines get-update <pipeline_id> "$UPDATE_ID" | jq -r '.update.state')
  echo "$(date +%H:%M:%S) update=$UPDATE_ID state=$STATE"
  case "$STATE" in COMPLETED|FAILED|CANCELED) break;; esac
  sleep 30
done

Why poll the update, not the pipeline. Top-level pipeline state flips back to RUNNING on RETRY_ON_FAILURE, so a loop watching the pipeline (or latest_updates[0]) can spin past a real FAILED update forever. Poll the captured update_id and stop on the first terminal state — including FAILED.

On `FAILED`: read the events log, don't re-run. The real error is in `error.exceptions[0].message`, not in the top-level `.message` — that one just says "Update X is FAILED", which is useless. Extract both:

databricks pipelines list-pipeline-events <pipeline_id> \
  | jq '[.[] | select(.level=="ERROR") | {
      event_type,
      summary: (.message // "")[0:200],
      exception: ((.error.exceptions[0].message // "no exception body") | .[0:800])
    }] | .[0:5]'

If you only see "Update X is FAILED" in your output, you're not extracting error.exceptions[0].message — fix the jq and re-run.

If the pipeline is already RUNNING, start-update queues the new update. Force-stop with databricks pipelines stop <pipeline_id> first if needed.

Step 5: Edit → re-upload → restart

# Re-upload (whole dir)
databricks workspace import-dir ./my_pipeline /Workspace/Users/<user>/my_pipeline --overwrite

# Or a single file
databricks workspace import /Workspace/Users/<user>/my_pipeline/gold.sql \
  --file ./my_pipeline/gold.sql --format RAW --overwrite

# Restart
databricks pipelines start-update <pipeline_id>

Use `--format RAW` for raw .sql / .py FILE entries. --format SOURCE --language SQL|PYTHON uploads a workspace notebook — and notebooks are deprecated for pipelines. Mixing the two on the same path fails with Cannot overwrite the asset ... due to type mismatch (asked: NOTEBOOK, actual: FILE).

Step 6: Validate output data

Even on COMPLETED, verify the data:

databricks experimental aitools tools discover-schema \
  my_catalog.my_schema.bronze_orders \
  my_catalog.my_schema.silver_orders \
  my_catalog.my_schema.gold_summary

Returns columns/types, 5 sample rows, total row count, and null counts per column per table.

Check for: empty tables (ingestion or filter problems), unexpected row counts (broken joins), missing columns (schema mismatch), nulls in key columns (data quality).

If validation reveals problems, trace upstream: run discover-schema on the source table of the problematic dataset, then its source, until you hit the layer where the issue originates. Bronze empty = source path wrong or files missing; silver empty = filter too aggressive or join condition mismatched; gold wrong counts = aggregation/grouping bug or duplicate keys in source.

---

Quick reference: CLI commands

Pipeline lifecycle

Run management

Supporting commands

---

Python SDK alternative

Same JSON shape via databricks.sdk.WorkspaceClient: w.pipelines.create(name=..., catalog=..., schema=..., serverless=True, continuous=False, development=True, channel="PREVIEW", configuration={...}, libraries=[...]). Capture pipeline.pipeline_id. Trigger with w.pipelines.start_update(pipeline_id=..., full_refresh=...) and poll w.pipelines.get_update(pipeline_id=..., update_id=update.update_id).update.state until it hits COMPLETED/FAILED/CANCELED. Prefer the CLI for interactive setup; the SDK is for programmatic / scripted workflows.

Auto CDC (Python)

CDC from streaming events (dp.create_auto_cdc_flow) or periodic snapshots (dp.create_auto_cdc_from_snapshot_flow). Both write into a pre-created streaming table.

Use streaming when CDC events arrive continuously (transaction logs, Kafka, Delta change feeds). Use snapshot when the source is a full dump compared to the previous state (daily extracts, batch exports).

Legacy aliases dp.apply_changes() / dp.apply_changes_from_snapshot() still parse but should be migrated (see SKILL.md Legacy DLT Syntax).

For querying SCD Type 2 history tables, see scd-2-querying.md.

dp.create_auto_cdc_flow(...)

Call at top level — does NOT return a value.

dp.create_auto_cdc_flow(
    target="<target_table>",                  # required — pre-created via dp.create_streaming_table()
    source="<source_table_or_view>",          # required — string name (table or @dp.temporary_view)
    keys=["key1", "key2"],                    # required — primary key columns
    sequence_by="<col>",                       # required — string col name, or col("ts"), or struct("ts","id")
    stored_as_scd_type=1,                     # 1 (default) = latest values; 2 = history with __START_AT/__END_AT
    ignore_null_updates=False,                # NULL values won't overwrite non-NULL existing
    apply_as_deletes=None,                    # expr("op = 'D'") or "op = 'D'"
    apply_as_truncates=None,                  # SCD Type 1 only
    column_list=None,                          # include list — mutually exclusive with except_column_list
    except_column_list=None,                  # exclude list
    track_history_column_list=None,           # SCD Type 2: cols that trigger new history rows
    track_history_except_column_list=None,    # SCD Type 2: cols that DON'T trigger new history rows
    name=None,                                 # flow name (multiple flows to one target)
    once=False,
)

source must be a table/view identifier (string) — NOT a DataFrame. To pre-filter, define a @dp.temporary_view() and reference its name. Don't materialize a streaming table just for filtering — temp view is preferred.

dp.create_auto_cdc_from_snapshot_flow(...)

dp.create_auto_cdc_from_snapshot_flow(
    target="<target_table>",
    source="<snapshot_table>",                # OR callable (see below)
    keys=["product_id"],
    stored_as_scd_type=1,
    track_history_column_list=None,
    track_history_except_column_list=None,
)

source accepts a string (most common — name of a table holding the latest snapshot) or a callable for historical snapshot replay:

def next_snapshot_and_version(latest_version: Optional[int]) -> Optional[Tuple[DataFrame, int]]:
    # Receives the last processed snapshot version (None on first run).
    # Return (DataFrame, version) for the next snapshot, or None when caught up.
    if latest_version is None:
        return (spark.read.load("products_v1.csv"), 1)
    return None

Version must be a comparable scalar (int, str, float, bytes, datetime, date, Decimal).

Patterns

Basic (SCD Type 1)

dp.create_streaming_table(name="users")
dp.create_auto_cdc_flow(target="users", source="user_changes",
                        keys=["user_id"], sequence_by="updated_at")

With pre-filtering via temp view

@dp.temporary_view()
def filtered_user_changes():
    return spark.readStream.table("raw_user_changes").filter("user_id IS NOT NULL")

dp.create_streaming_table(name="users")
dp.create_auto_cdc_flow(target="users", source="filtered_user_changes",
                        keys=["user_id"], sequence_by="updated_at")

Explicit deletes + truncates + ignore-null

from pyspark.sql.functions import expr

dp.create_auto_cdc_flow(
    target="orders", source="order_events", keys=["order_id"],
    sequence_by="event_timestamp",
    apply_as_deletes=expr("operation = 'DELETE'"),
    apply_as_truncates=expr("operation = 'TRUNCATE'"),    # SCD Type 1 only
    ignore_null_updates=True,
)

SCD Type 2 with selective history tracking

dp.create_auto_cdc_flow(
    target="accounts", source="account_changes", keys=["account_id"],
    sequence_by="modified_at",
    stored_as_scd_type=2,
    track_history_column_list=["balance", "status"],   # only these trigger new history rows
)

Use track_history_except_column_list=[...] for the inverse.

Snapshot-based (table source)

@dp.materialized_view(name="product_snapshot")
def product_snapshot():
    return spark.read.table("source.daily_product_dump")

dp.create_streaming_table(name="products")
dp.create_auto_cdc_from_snapshot_flow(
    target="products", source="product_snapshot",
    keys=["product_id"], stored_as_scd_type=1,
)

Key rules

• Create the target with dp.create_streaming_table() first.
• dp.create_auto_cdc_flow() does NOT return a value — call at top level.
• source is a string table/view name, never a DataFrame. Pre-process via @dp.temporary_view().
• SCD Type 2 adds __START_AT / __END_AT columns with the same type as sequence_by. If you supply an explicit target schema, include them.
• sequence_by accepts string column name OR col("ts") — both work. Use struct("ts", "id") for multi-column ordering.

Auto CDC (SQL)

AUTO CDC INTO processes CDC events from a streaming source into a target streaming table. SCD Type 1 (latest) or Type 2 (history). The target must be pre-created.

SQL only supports CDC from streaming sources (AUTO CDC INTO). For periodic-snapshot CDC, use Python's dp.create_auto_cdc_from_snapshot_flow() — see auto-cdc-python.md.

Syntax

CREATE OR REFRESH STREAMING TABLE <target_table>;

CREATE FLOW <flow_name> AS AUTO CDC INTO <target_table>
FROM STREAM(<source_table_or_view>)
KEYS (<key1>, <key2>, ...)
[IGNORE NULL UPDATES]
[APPLY AS DELETE WHEN <condition>]
[APPLY AS TRUNCATE WHEN <condition>]                 -- SCD Type 1 only
SEQUENCE BY <col_or_struct>
[COLUMNS {<col_list> | * EXCEPT (<col_list>)}]
[STORED AS {SCD TYPE 1 | SCD TYPE 2}]                -- default Type 1
[TRACK HISTORY ON {<col_list> | * EXCEPT (<col_list>)}]   -- SCD Type 2 only

Clause notes:

• FROM STREAM(...) accepts only a table/view identifier — NOT a subquery. Pre-filter via a temp view if needed.
• KEYS — required primary key columns for row identification.
• IGNORE NULL UPDATES — NULL values won't overwrite existing non-NULL values.
• APPLY AS DELETE WHEN / APPLY AS TRUNCATE WHEN — order matters in the SQL: put both before SEQUENCE BY or the parser fails.
• SEQUENCE BY — single column, or STRUCT(ts_col, tiebreaker_col) for multi-column ordering.
• COLUMNS * EXCEPT (...) — only list columns that exist in the source (omit _rescued_data unless bronze rescued data).
• STORED AS SCD TYPE 2 adds __START_AT and __END_AT system columns to the target. If you supply an explicit target schema, include them with the same type as SEQUENCE BY.
• TRACK HISTORY ON cols — Type 2 only; only listed columns trigger new history rows. Others get in-place Type-1 updates.

For querying Type 2 history tables, see scd-2-querying.md.

Patterns

Basic (SCD Type 1, default)

CREATE OR REFRESH STREAMING TABLE users;

CREATE FLOW user_flow AS AUTO CDC INTO users
FROM STREAM(user_changes)
KEYS (user_id)
SEQUENCE BY updated_at;

Pre-filter via temporary view (when the source needs transformation)

CREATE OR REFRESH TEMPORARY VIEW filtered_changes AS
SELECT * FROM source_table WHERE status = 'active';

CREATE OR REFRESH STREAMING TABLE active_records;

CREATE FLOW active_flow AS AUTO CDC INTO active_records
FROM STREAM(filtered_changes)
KEYS (record_id)
SEQUENCE BY updated_at;

Explicit deletes + ignore NULL updates

CREATE FLOW order_flow AS AUTO CDC INTO orders
FROM STREAM(order_events)
KEYS (order_id)
IGNORE NULL UPDATES
APPLY AS DELETE WHEN operation = 'DELETE'
SEQUENCE BY event_timestamp;

SCD Type 2 (full history)

CREATE FLOW customer_flow AS AUTO CDC INTO customer_history
FROM STREAM(customer_changes)
KEYS (customer_id)
SEQUENCE BY changed_at
STORED AS SCD TYPE 2;

Variants: TRACK HISTORY ON balance, status (only those columns trigger new rows) or TRACK HISTORY ON * EXCEPT (last_login, view_count) (track everything except).

Selective columns

COLUMNS account_id, balance, status (include list) or COLUMNS * EXCEPT (internal_notes, temp_field) (exclude list).

Multi-column sequencing

SEQUENCE BY STRUCT(event_timestamp, event_id)    -- order by ts first, break ties with id

TRUNCATE support (SCD Type 1 only)

APPLY AS TRUNCATE WHEN operation = 'TRUNCATE'
SEQUENCE BY event_timestamp
STORED AS SCD TYPE 1;

Auto Loader (Python)

spark.readStream.format("cloudFiles") for incremental ingestion from cloud storage. Returns a streaming DataFrame; use inside @dp.table() or @dp.append_flow().

@dp.table()
def my_table():
    return (spark.readStream.format("cloudFiles")
                 .option("cloudFiles.format", "json")     # json, csv, parquet, avro, orc, xml, text, binaryFile
                 .load("s3://bucket/path"))

Rules

• Don't set `cloudFiles.schemaLocation` — the pipeline manages schema location and checkpoint automatically.
• Use spark.readStream (streaming), not spark.read (batch). Auto Loader is streaming by definition.
• If you provide an explicit schema=, include the rescued-data column (default name _rescued_data STRING; configurable via rescuedDataColumn option).
• Look up the official Databricks docs for any option before use — every option has subtle semantics not captured here.

Schema handling

• cloudFiles.inferColumnTypes — enable type inference (default: all-string for JSON/CSV/XML).
• cloudFiles.schemaHints — partial typing, e.g. "id INT, amount DECIMAL(10,2)".
• cloudFiles.schemaEvolutionMode — how to handle new columns (addNewColumns, rescue, failOnNewColumns, none).
• Quarantine malformed rows via the rescued-data pattern in streaming-patterns.md#rescue-data-quarantine.

Common format-agnostic options

Generic file options (apply to all formats): ignoreCorruptFiles, ignoreMissingFiles, modifiedAfter, modifiedBefore, pathGlobFilter / fileNamePattern, recursiveFileLookup.

Listing strategy:

• Directory listing (default for small/medium volumes): cloudFiles.useIncrementalListing.
• File notification (recommended at scale): cloudFiles.useNotifications, cloudFiles.useManagedFileEvents, cloudFiles.fetchParallelism, cloudFiles.pathRewrites, cloudFiles.resourceTag.

Cloud-specific auth options

All clouds accept databricks.serviceCredential to reference a UC service credential — prefer this over inline keys.

• AWS: cloudFiles.region, cloudFiles.queueUrl, cloudFiles.awsAccessKey / awsSecretKey, cloudFiles.roleArn / roleExternalId / roleSessionName, cloudFiles.stsEndpoint.
• Azure: cloudFiles.resourceGroup, cloudFiles.subscriptionId, cloudFiles.clientId / clientSecret, cloudFiles.connectionString, cloudFiles.tenantId, cloudFiles.queueName.
• GCP: cloudFiles.projectId, cloudFiles.client, cloudFiles.clientEmail, cloudFiles.privateKey / privateKeyId, cloudFiles.subscription.

Format-specific options

See JSON, CSV, Parquet, Avro, ORC, XML, Text.

Auto Loader (SQL)

read_files() for incremental ingestion from cloud storage. Use inside a streaming table as FROM STREAM read_files(...).

-- In a streaming table definition
CREATE OR REFRESH STREAMING TABLE my_table
AS SELECT * FROM STREAM read_files('s3://bucket/path', format => 'json');

-- Or via a flow into a pre-created target
CREATE OR REFRESH STREAMING TABLE target_table;

CREATE FLOW ingest_flow
AS INSERT INTO target_table BY NAME
SELECT * FROM STREAM read_files('s3://bucket/path', format => 'json');

Rules

• FROM STREAM read_files(...) (no extra parens around the function) — that's the canonical form for function sources. Without STREAM, read_files is a batch read and fails inside a streaming table.
• inferColumnTypes defaults to true for read_files (opposite of cloudFiles in Python). Set false to force string types.
• Use schemaHints => 'col1 TYPE, ...' for production tables; schemaEvolutionMode => '...' to control schema-drift behavior.
• Unity Catalog pipelines must use external locations to load files.
• Look up the official Databricks docs for any option before use.

Common format-agnostic options

Generic file options: ignoreCorruptFiles, ignoreMissingFiles, modifiedAfter, modifiedBefore, pathGlobFilter / fileNamePattern, recursiveFileLookup.

Format-specific options

See JSON, CSV, Parquet, Avro, ORC, XML, Text.

Migration Guide: DLT → SDP

Two migration paths:

1. DLT Python → SDP Python (dlt → dp): same language, new API. 2. DLT Python → SDP SQL: convert to SQL when the logic is mostly relational.

If 80%+ of the pipeline is SQL-expressible (filters, aggregations, joins, CDC, Auto Loader), prefer SDP SQL. Stay in Python when there are complex UDFs, external API calls, custom libraries, or ML inference.

---

Migration Path 1: DLT Python → SDP Python

Mapping

Behavioral changes to watch for in CDC

• apply_changes(...) → create_auto_cdc_flow(...). Same parameters EXCEPT:
• sequence_by accepts string OR col(...); either works.
• stored_as_scd_type is integer `2` for Type 2, string `"1"` for Type 1.

# Legacy
dlt.apply_changes(target="dim_customers", source="customers_cdc",
                  keys=["customer_id"], sequence_by="updated_at",
                  stored_as_scd_type="2")

# Modern
dp.create_auto_cdc_flow(target="dim_customers", source="customers_cdc",
                        keys=["customer_id"], sequence_by=F.col("updated_at"),
                        stored_as_scd_type=2)

---

Migration Path 2: DLT Python → SDP SQL

Streaming table with Auto Loader

# DLT Python
@dlt.table(name="bronze_sales", comment="Raw sales")
def bronze_sales():
    return (spark.readStream.format("cloudFiles")
                 .option("cloudFiles.format", "json")
                 .load("/Volumes/cat/sch/raw/sales")
                 .withColumn("_ingested_at", F.current_timestamp()))

-- SDP SQL
CREATE OR REFRESH STREAMING TABLE bronze_sales
COMMENT 'Raw sales' AS
SELECT *, current_timestamp() AS _ingested_at
FROM STREAM read_files('/Volumes/cat/sch/raw/sales', format => 'json');

Filter / cast / select

DLT Python dlt.read_stream("bronze_sales").withColumn("amount", ...cast("decimal(10,2)")).filter(...) becomes:

CREATE OR REFRESH STREAMING TABLE silver_sales AS
SELECT sale_id, customer_id,
       CAST(amount AS DECIMAL(10,2)) AS amount,
       CAST(sale_date AS DATE) AS sale_date
FROM STREAM(bronze_sales)
WHERE amount > 0 AND sale_id IS NOT NULL;

SCD Type 2

CREATE OR REFRESH STREAMING TABLE customers_history;

CREATE FLOW customers_scd2_flow AS
AUTO CDC INTO customers_history
FROM STREAM(customers_cdc_clean)
KEYS (customer_id)
APPLY AS DELETE WHEN operation = 'DELETE'
SEQUENCE BY event_timestamp
COLUMNS * EXCEPT (operation, _ingested_at, _source_file)
STORED AS SCD TYPE 2;

Put APPLY AS DELETE WHEN before SEQUENCE BY. Only list columns in COLUMNS * EXCEPT (...) that exist in the source — _rescued_data should only appear if bronze uses rescue data.

Expectations

Three options for @dlt.expect_or_drop("valid_amount", "amount > 0"):

-- 1. Constraint (closest equivalent, with metrics)
CREATE OR REFRESH STREAMING TABLE silver_sales (
  CONSTRAINT valid_amount EXPECT (amount > 0) ON VIOLATION DROP ROW
) AS SELECT * FROM STREAM(bronze_sales);

-- 2. WHERE filter (no metrics, simplest)
... WHERE amount > 0

-- 3. Quarantine pattern (full audit trail; route bad rows to a side table) —
-- see streaming-patterns.md#rescue-data-quarantine

UDFs

Simple UDFs (categorisation, math) translate to SQL CASE:

SELECT *,
  CASE WHEN amount > 1000 THEN 'High'
       WHEN amount > 100  THEN 'Medium'
       ELSE 'Low' END AS category
FROM sales;

Keep complex UDFs (external APIs, custom algorithms, ML inference) in Python with the modern dp API.

---

Migration Order (by layer)

1. Bronze (ingestion) — cloudFiles → read_files() (or keep cloudFiles with dp). 2. Silver (cleansing) — dlt.expect* → WHERE clause or dp.expect*. 3. Gold (aggregations) — usually straightforward port. 4. CDC/SCD — apply_changes(...) → AUTO CDC INTO (SQL) or dp.create_auto_cdc_flow(...) (Python).

Run old and new in parallel during cutover and diff outputs before retiring the old pipeline.

---

Common Issues

---

Related

• python-basics.md — modern dp API reference
• auto-cdc-python.md / auto-cdc-sql.md — full CDC API
• SKILL.md — Legacy DLT Syntax mapping table

Expectations (Python)

Data-quality constraints stacked above @dp.materialized_view() / @dp.table() / @dp.temporary_view() functions. Each constraint is a SQL Boolean string evaluated per row.

Legacy @dlt.expect* decorators still parse but should be migrated to @dp.expect* (same names, same semantics) — see SKILL.md Legacy DLT Syntax.

Decorators

• name (str) — unique within the dataset; appears in metrics.
• condition (str) — a SQL Boolean expression. Built-ins are fine. No Python UDFs, external calls, or subqueries.

Patterns

Single decorator

@dp.materialized_view()
@dp.expect_or_drop("valid_email", "email IS NOT NULL AND email LIKE '%@%'")
def customer_contacts():
    return spark.read.table("raw_contacts")

@dp.expect("name", "cond") (warn) and @dp.expect_or_fail(...) (fail) follow the same shape.

Multiple expectations, same action — use expect_all

@dp.materialized_view()
@dp.expect_all({
    "valid_age":     "age >= 0 AND age <= 120",
    "valid_country": "country_code IN ('US', 'CA', 'MX')",
    "recent_date":   "created_date >= '2020-01-01'",
})
def validated_customers():
    return spark.read.table("raw_customers")

Multiple expectations, mixed actions — stack decorators

@dp.materialized_view(comment="Clean customer data")
@dp.expect_or_drop("valid_email", "email LIKE '%@%'")
@dp.expect_or_fail("required_id", "id IS NOT NULL")
@dp.expect("valid_age", "age BETWEEN 0 AND 120")
def customers_clean():
    return spark.read.table("raw_customers")

Temporary view with expectations

@dp.temporary_view(name="high_value_customers")
@dp.expect("valid_total", "total_purchases > 0")
def high_value_view():
    return (spark.read.table("orders")
                 .groupBy("customer_id")
                 .agg(F.sum("amount").alias("total_purchases"))
                 .filter("total_purchases > 1000"))

Best Practices

• Unique, descriptive names — they appear in metrics.
• expect_or_fail for critical business invariants.
• expect_or_drop for cleansing operations.
• expect (warn) for measuring soft quality without blocking.
• Group same-action constraints in expect_all* rather than stacking many decorators.
• Predicate is a SQL string — no Python UDFs, subqueries, external calls.

Expectations (SQL)

Data-quality constraints inside CREATE OR REFRESH STREAMING TABLE / MATERIALIZED VIEW / CREATE LIVE VIEW. Each constraint is a SQL Boolean expression evaluated per row; the action on violation is (default) warn, DROP ROW, or FAIL UPDATE.

CREATE TEMPORARY VIEW does NOT support CONSTRAINT clauses. Use CREATE LIVE VIEW for the edge case of "temporary view with expectations" — see temporary-view-sql.md#using-expectations-with-temporary-views.

Syntax

CREATE OR REFRESH STREAMING TABLE table_name (
  CONSTRAINT name1 EXPECT (cond1),                                  -- warn (default)
  CONSTRAINT name2 EXPECT (cond2) ON VIOLATION DROP ROW,            -- drop violating rows
  CONSTRAINT name3 EXPECT (cond3) ON VIOLATION FAIL UPDATE          -- fail pipeline on first violation
) AS SELECT ...

• constraint_name must be unique within the dataset; describes what's validated.
• condition is a SQL Boolean expression. Built-in functions (year(...), current_date(), CASE, ...) are fine. No Python UDFs, external calls, or subqueries.
• Multiple CONSTRAINT clauses are stacked comma-separated and each can have a different action.
• Action semantics:
• warn (default): violations logged, invalid rows still written to the target. Metrics collected.
• `DROP ROW`: violating rows dropped before write. Metrics collected.
• `FAIL UPDATE`: first violation fails the pipeline atomically; transaction rolls back. Requires manual fix.

Patterns

Mixed actions in one dataset

CREATE OR REFRESH STREAMING TABLE customers_clean (
  CONSTRAINT valid_email EXPECT (email LIKE '%@%')      ON VIOLATION DROP ROW,
  CONSTRAINT required_id EXPECT (id IS NOT NULL)        ON VIOLATION FAIL UPDATE,
  CONSTRAINT valid_age   EXPECT (age BETWEEN 0 AND 120)                          -- warn only
) AS SELECT * FROM STREAM(raw_customers);

With SQL functions / complex predicates

CREATE OR REFRESH STREAMING TABLE transactions (
  CONSTRAINT valid_date          EXPECT (year(transaction_date) >= 2020),
  CONSTRAINT non_negative_price  EXPECT (price >= 0),
  CONSTRAINT recent_purchase     EXPECT (transaction_date <= current_date())
) AS SELECT * FROM STREAM(raw_transactions);

CREATE OR REFRESH MATERIALIZED VIEW active_subscriptions (
  CONSTRAINT valid_dates EXPECT (
    start_date <= end_date
    AND end_date <= current_date()
    AND start_date >= '2020-01-01'
  ) ON VIOLATION DROP ROW
) AS SELECT * FROM subscriptions WHERE status = 'active';

Temporary view + expectation (only via CREATE LIVE VIEW)

CREATE LIVE VIEW high_value_customers (
  CONSTRAINT valid_total EXPECT (total_purchases > 0)
) AS
SELECT customer_id, SUM(amount) AS total_purchases
FROM orders
GROUP BY customer_id
HAVING total_purchases > 1000;

Monitoring

Metrics show up in the pipeline UI Data quality tab and the event log. Available for warn and DROP ROW actions. Unavailable if the pipeline fails before completion.

Best Practices

• Unique, descriptive constraint names — they appear in metrics.
• FAIL UPDATE for critical business invariants (anything that should never reach downstream consumers).
• DROP ROW for data-cleansing operations where you accept some loss.
• Default (warn) for soft quality metrics you want to measure without blocking.
• Keep the predicate simple — no Python, no subqueries, no UDFs.

ForEachBatch Sinks (Python, Public Preview)

Process the stream as micro-batches with custom Python logic — for things the built-in delta / kafka sinks can't do: MERGE/upsert into Delta, fan out to multiple destinations per batch, or write to unsupported targets (JDBC, etc.).

@dp.foreach_batch_sink(name="...")

@dp.foreach_batch_sink(name="<sink_name>")        # name optional; defaults to function name
def my_sink(df, batch_id):
    # df: micro-batch DataFrame
    # batch_id: int, increments per trigger. 0 = first run OR start of full refresh.
    # Access SparkSession via df.sparkSession (NOT the module-level `spark`)
    ...

The handler doesn't return a value. Write to a sink via @dp.append_flow(target="<sink_name>") — multiple flows can target the same sink, each with its own checkpoint.

Patterns

MERGE/upsert into an existing Delta table

@dp.foreach_batch_sink(name="upsert_sink")
def upsert_sink(df, batch_id):
    df.createOrReplaceTempView("batch_data")
    df.sparkSession.sql("""
        MERGE INTO target_catalog.schema.target_table AS t
        USING batch_data AS s ON t.id = s.id
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)

@dp.append_flow(target="upsert_sink")
def upsert_flow():
    return spark.readStream.table("source_events")

The target Delta table must exist before the MERGE runs — create it externally or in the handler on batch_id == 0.

Fan out to multiple destinations (idempotent)

Use txnVersion + txnAppId so partial-failure retries don't double-write.

APP_ID = "my-app-name"   # unique per application writing to the same target

@dp.foreach_batch_sink(name="multi_target_sink")
def multi_target_sink(df, batch_id):
    df.persist()         # avoid re-reading the source for each destination
    df.write.format("delta").mode("append") \
        .option("txnVersion", batch_id).option("txnAppId", APP_ID) \
        .saveAsTable("my_catalog.my_schema.table_a")
    df.write.format("json").mode("append") \
        .option("txnVersion", batch_id).option("txnAppId", APP_ID) \
        .save("/tmp/json_target")

@dp.append_flow(target="multi_target_sink")
def multi_target_flow():
    return spark.readStream.table("processed_events")

Key rules

• Streaming only — append flows only. No batch DataFrames, no Auto CDC.
• The pipeline does NOT track sink data. On full refresh, checkpoints reset and batch_id restarts at 0 but your target is NOT cleaned up — truncate/drop manually if you want a clean slate.
• Access the session via df.sparkSession, not the module-level spark.
• Multiple @dp.append_flows can target the same sink; each maintains its own checkpoint.
• For Delta writes use txnVersion/txnAppId for idempotency. For multi-destination handlers, df.persist() / df.cache() to avoid re-reading the source.
• Keep handlers small — no threading, no heavy libraries, no large in-memory work.
• databricks-connect: the handler must be serializable and must not call dbutils. Capture dbutils.widgets.get(...) values into variables outside the handler. Non-serializable handlers log a warning but may fail at runtime.

Kafka Ingestion

Ingest from Apache Kafka into a streaming table. Same shape works for Azure Event Hubs (Kafka protocol on port 9093) — only the connection string and SASL config differ.

For Kinesis, Pub/Sub, and Pulsar, use the analogous read_kinesis / read_pubsub / read_pulsar SQL functions or spark.readStream.format("kinesis|pubsub|pulsar") — same overall shape as below.

Basic Read

Kafka returns rows with binary key and value columns plus topic, partition, offset, timestamp. Cast to STRING/BINARY and parse downstream — don't carry raw bytes.

CREATE OR REFRESH STREAMING TABLE bronze_kafka_events AS
SELECT CAST(key AS STRING)   AS event_key,
       CAST(value AS STRING) AS event_value,
       topic, partition, offset,
       timestamp           AS kafka_timestamp,
       current_timestamp() AS _ingested_at
FROM read_kafka(
  bootstrapServers => '${kafka_brokers}',
  subscribe        => 'events-topic',
  startingOffsets  => 'latest'
);

Python equivalent: spark.readStream.format("kafka").option("kafka.bootstrap.servers", spark.conf.get("kafka_brokers")).option("subscribe", "events-topic").option("startingOffsets", "latest").load().selectExpr(...) + .withColumn("_ingested_at", F.current_timestamp()).

`read_kafka` reference.

Common options

Parse JSON Payloads

value is a blob. Extract structured columns with from_json against an explicit schema — JSON-schema inference from a streaming Kafka source is not supported.

CREATE OR REFRESH STREAMING TABLE silver_events AS
SELECT data.*, kafka_timestamp, _ingested_at
FROM (
  SELECT from_json(event_value,
                   'event_id STRING, event_type STRING, timestamp TIMESTAMP') AS data,
         kafka_timestamp, _ingested_at
  FROM STREAM(bronze_kafka_events)
);

Python: build a StructType and .withColumn("data", F.from_json("event_value", event_schema)).select("data.*", ...). Keep the schema in code, versioned alongside the pipeline.

For Avro / Protobuf payloads, swap from_json for from_avro / from_protobuf (with Schema Registry config). Same overall pattern.

Authentication

Use {{secrets/scope/key}} interpolation in SQL or dbutils.secrets.get(scope, key) in Python. Never hard-code credentials.

-- SASL/PLAIN
FROM read_kafka(
  bootstrapServers          => '${kafka_brokers}',
  subscribe                 => 'events-topic',
  `kafka.security.protocol` => 'SASL_SSL',
  `kafka.sasl.mechanism`    => 'PLAIN',
  `kafka.sasl.jaas.config`  =>
    'org.apache.kafka.common.security.plain.PlainLoginModule required ' ||
    'username="{{secrets/kafka/username}}" ' ||
    'password="{{secrets/kafka/password}}";'
);

For mTLS, add kafka.ssl.truststore.* and kafka.ssl.keystore.* options pointing at files in a UC volume; pass paths via pipeline config.

Event Hubs (via Kafka protocol)

Same Kafka source — change the connection target and auth:

bootstrapServers          => '<namespace>.servicebus.windows.net:9093',
subscribe                 => '<event-hub-name>',
`kafka.security.protocol` => 'SASL_SSL',
`kafka.sasl.mechanism`    => 'PLAIN',
`kafka.sasl.jaas.config`  =>
    'org.apache.kafka.common.security.plain.PlainLoginModule required '
    'username="$ConnectionString" '
    'password="{{secrets/eventhub/connection-string}}";'

The username is the literal $ConnectionString; the password is the namespace- or entity-level connection string (with SharedAccessKey=...).

Pipeline Configuration

Pass brokers, topics, consumer-group identity through pipeline config so dev/prod differ without code changes.

# resources/<name>.pipeline.yml
resources:
  pipelines:
    my_pipeline:
      configuration:
        kafka_brokers: "broker-1:9092,broker-2:9092,broker-3:9092"
        kafka_topic:   "events-topic"

Read with spark.conf.get("kafka_brokers") (Python) or ${kafka_brokers} (SQL).

Writing to Kafka (sinks)

Sinks are Python-only. Create a sink with format="kafka" and write via @dp.append_flow. The value column is mandatory — use to_json(struct(*)) to serialize the row. See sink-python.md.

Best Practices

1. Cast value to STRING / BINARY and parse with from_json / from_avro against an explicit schema. 2. Add _ingested_at — see streaming-patterns.md#monitoring-lag. 3. Tune maxOffsetsPerTrigger if downstream operations bottleneck. 4. Don't set failOnDataLoss = false unless you accept retention-window gaps.

Common Issues

Materialized Views (Python)

Batch processing with full refresh or incremental computation. For streaming tables, see streaming-table-python.md. For the incremental-refresh operation-support table, see materialized-view-sql.md.

@dp.materialized_view() — preferred

@dp.materialized_view(
    name="<name>",
    comment="<comment>",
    spark_conf={...},
    table_properties={...},
    path="<storage-location>",
    cluster_by=["<col>", ...],          # Liquid Clustering — preferred
    cluster_by_auto=True,                # let Databricks pick keys
    partition_cols=["<col>"],            # legacy, prefer cluster_by — see performance.md#liquid-clustering
    schema="col1 TYPE, ...",             # supports GENERATED ALWAYS AS, MASK clauses, PK/FK constraints
    row_filter="ROW FILTER my_catalog.my_schema.func ON (col)",
    private=False,                       # True = pipeline-scoped, not published to UC
)
def my_mv():
    return spark.read.table("source.data")     # must be a batch DataFrame

@dp.table() with a batch DataFrame return type also creates a materialized view (legacy DLT shape), but @dp.materialized_view() is the recommended decorator. Use @dp.table only for streaming tables now.

For the detailed semantics of row_filter (UC SQL UDF returning BOOLEAN; forces full refresh of downstream MVs; cannot define the UDF inside the pipeline), see streaming-table-python.md.

Incremental refresh

Requires serverless + Delta row tracking on source tables (delta.enableRowTracking = true). Falls back to full recompute otherwise. For the supported-operations matrix, see materialized-view-sql.md — same support applies to the Python DataFrame equivalents.

For exactly-once semantics (Kafka, Auto Loader), use a streaming table instead.

Patterns

Aggregation with clustering

@dp.materialized_view(name="daily_sales_summary", cluster_by=["sale_date", "region"])
def daily_sales_summary():
    return (spark.read.table("raw.orders")
                 .withColumn("sale_date", F.to_date("order_timestamp"))
                 .groupBy("sale_date", "region")
                 .agg(F.count("*").alias("order_count"),
                      F.sum("amount").alias("total_revenue")))

Generated columns

@dp.materialized_view(
    schema="""
        order_datetime STRING,
        order_day_of_week STRING GENERATED ALWAYS AS (dayofweek(order_datetime)),
        customer_id BIGINT,
        amount DECIMAL(10,2)
    """,
    cluster_by=["order_day_of_week", "customer_id"],
)
def orders_with_day():
    return spark.read.table("raw.orders")

Row filter / column masking (UC, Public Preview)

@dp.materialized_view(
    name="employees",
    schema="emp_id INT, emp_name STRING, dept STRING, salary DECIMAL(10,2)",
    row_filter="ROW FILTER my_catalog.my_schema.filter_by_dept ON (dept)",
)
def employees():
    return spark.read.table("source.employees")

Column masking uses MASK ... USING COLUMNS (...) inside the schema= string — same form as in SQL.

Key rules

• MVs use spark.read (batch); streaming tables use spark.readStream.
• Never .write, .save(), .saveAsTable(), .toTable() — Databricks manages writes.
• Generated columns, PK/FK constraints, and MASK clauses require an explicit schema=.
• Row filters on source tables force full refresh of downstream MVs.

Materialized Views (SQL)

Batch processing with full refresh or incremental computation. For streaming tables (incremental streaming), see streaming-table-sql.md.

Syntax

CREATE OR REFRESH [PRIVATE] MATERIALIZED VIEW view_name
  [ ( col_name col_type [NOT NULL] [COMMENT '...'] [column_constraint | MASK clause]
      [, ...]
      [, CONSTRAINT name EXPECT (cond) [ON VIOLATION DROP ROW | FAIL UPDATE]]
      [, table_constraint] ) ]
  [ PARTITIONED BY (col, ...) | CLUSTER BY (col, ...) ]   -- prefer CLUSTER BY
  [ LOCATION path ]                                       -- Hive metastore only
  [ COMMENT '...' ]
  [ TBLPROPERTIES (key = value, ...) ]
  [ WITH ROW FILTER func_name ON (col, ...) ]
  AS query

Clause notes (same semantics as streaming tables — see streaming-table-sql.md for the detailed treatment of PRIVATE, MASK, WITH ROW FILTER, and informational table constraints):

• query must NOT use STREAM(...) — MVs are batch. Streaming reads belong in a streaming table.
• PRIMARY KEY requires explicit NOT NULL.
• Generated columns supported via col TYPE GENERATED ALWAYS AS (expr).
• Identity columns, default columns, and explicit OPTIMIZE / VACUUM are not supported (the pipeline handles maintenance).
• Non-column expressions in the SELECT list require explicit aliases.
• Sum aggregates over a nullable column return 0 (not NULL) when only NULLs remain.

Incremental refresh

MVs use incremental refresh automatically when possible. Falls back to full recompute otherwise.

Requirements: serverless pipeline, source is Delta / MV / streaming table, row tracking enabled on sources (for ops marked below).

Enable delta.enableRowTracking = true, delta.enableChangeDataFeed = true, and deletion vectors on source tables for the best incremental coverage. For exactly-once semantics (Kafka, Auto Loader), use a streaming table instead.

Patterns

Aggregation with Liquid Clustering

CREATE OR REFRESH MATERIALIZED VIEW daily_sales_summary
CLUSTER BY (sale_date, region)
AS
SELECT DATE(order_timestamp) AS sale_date, region,
       COUNT(*) AS order_count, SUM(amount) AS total_revenue
FROM raw.orders
GROUP BY DATE(order_timestamp), region;

Generated column

CREATE OR REFRESH MATERIALIZED VIEW orders_with_day (
  order_datetime    STRING,
  order_day_of_week STRING GENERATED ALWAYS AS (dayofweek(order_datetime)),
  customer_id       BIGINT,
  amount            DECIMAL(10,2)
)
CLUSTER BY (order_day_of_week, customer_id)
AS SELECT order_datetime, customer_id, amount FROM raw.orders;

Row filter (UC, Public Preview)

CREATE OR REFRESH MATERIALIZED VIEW employees (
  emp_id INT, emp_name STRING, dept STRING, salary DECIMAL(10,2)
)
WITH ROW FILTER my_catalog.my_schema.filter_by_dept ON (dept)
AS SELECT * FROM source.employees;

Column masking (UC, Public Preview)

CREATE OR REFRESH MATERIALIZED VIEW users_with_masked_ssn (
  user_id BIGINT,
  ssn     STRING MASK catalog.schema.ssn_mask_fn USING COLUMNS (region),
  region  STRING
)
AS SELECT user_id, ssn, region FROM raw.users;

AVRO-Specific Options

CSV-Specific Options

JSON-Specific Options

ORC-Specific Options

PARQUET-Specific Options

TEXT-Specific Options

XML-Specific Options

Performance Tuning

Liquid Clustering, state management for streaming, join strategy, query optimization, pre-aggregation. SQL is shown as canonical; Python equivalents use the obvious @dp.table + DataFrame translation (cluster_by=[...], table_properties={...}).

---

Liquid Clustering

Recommended for data layout. Replaces PARTITION BY + ZORDER. Adaptive, multi-dimensional, self-optimizing — no manual OPTIMIZE needed.

CREATE OR REFRESH STREAMING TABLE bronze_events
CLUSTER BY (event_type, event_date)
AS SELECT *, current_timestamp() AS _ingested_at
FROM STREAM read_files('/Volumes/cat/sch/raw/events/', format => 'json');

Python: @dp.table(cluster_by=["event_type", "event_date"]).

Use CLUSTER BY (AUTO) / cluster_by=["AUTO"] while learning the workload, prototyping, or when access patterns are unclear. Pick keys manually for production once query patterns are stable.

Cluster key data types

Numeric, string, date, or timestamp only. BOOLEAN, ARRAY, MAP, STRUCT, BINARY fail at first write with DELTA_CLUSTERING_COLUMNS_DATATYPE_NOT_SUPPORTED (no data-skipping stats). Low-cardinality flags also don't benefit from clustering — leave them out.

Cluster key selection by layer

Rules of thumb: most-selective key first, second-most-common filter second; order matters; cap at 4 keys (diminishing returns beyond). Use AUTO if unsure.

Migrating from PARTITION BY + ZORDER

Replace:

PARTITIONED BY (date DATE)
TBLPROPERTIES ('pipelines.autoOptimize.zOrderCols' = 'user_id,event_type')

with:

CLUSTER BY (date, user_id, event_type)

Typical wins: 20–50% query improvement, no small-file problem, automatic optimization. Keep `PARTITION BY` only for: regulatory physical separation, data lifecycle requiring DROP PARTITION, DBR < 13.3 compatibility, or huge existing tables where migration cost > benefit.

---

Table Properties

TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',     -- right-size new files on write
  'delta.autoOptimize.autoCompact'   = 'true',     -- compact small files automatically
  'delta.enableChangeDataFeed'       = 'true',     -- if downstream needs CDF
  'delta.logRetentionDuration'        = '7 days',  -- high-volume tables only
  'delta.deletedFileRetentionDuration' = '7 days'  -- shortens time-travel window
)

Python: table_properties={"delta.autoOptimize.optimizeWrite": "true", ...}.

Short retention windows break time-travel queries beyond the window — only set on high-volume tables where storage cost dominates.

---

Materialized View Refresh

CREATE OR REFRESH MATERIALIZED VIEW gold_live_metrics
REFRESH EVERY 5 MINUTES                  -- or REFRESH EVERY 1 DAY for batch reports
AS SELECT metric_name, AVG(metric_value) AS avg_value, MAX(last_updated) AS freshness
   FROM silver_metrics GROUP BY metric_name;

Incremental refresh

MVs use incremental refresh automatically when possible. Requirements:

• Serverless pipeline (incremental refresh for aggregations is serverless-only).
• Source has Delta row tracking enabled (delta.enableRowTracking = true).
• No row-level filters on the source.
• Aggregation/expression pattern is supported.

Falls back to full recompute if any requirement isn't met.

---

State Management for Streaming

Higher cardinality → more state. Watch the combinations in GROUP BY.

-- High state: every unique combination creates state
SELECT user_id, product_id, session_id, COUNT(*)
FROM STREAM(bronze_events)
GROUP BY user_id, product_id, session_id;   -- 1M × 10K × 100M — massive

Three strategies to bound state:

1. Reduce cardinality — group by coarser keys.

-- 100 categories instead of 10K products
GROUP BY user_id, product_category, DATE(event_time)

2. Use time windows — explicit retention boundary.

GROUP BY user_id, window(event_time, '1 hour')

3. Materialize daily then aggregate batch monthly — move state from streaming to batch.

CREATE OR REFRESH STREAMING TABLE user_daily_stats AS
SELECT user_id, DATE(event_time) AS event_date, COUNT(*) AS event_count
FROM STREAM(bronze_events)
GROUP BY user_id, DATE(event_time);

CREATE OR REFRESH MATERIALIZED VIEW user_monthly_stats AS
SELECT user_id, DATE_TRUNC('month', event_date) AS month, SUM(event_count) AS total_events
FROM user_daily_stats
GROUP BY user_id, DATE_TRUNC('month', event_date);

---

Join Optimization

Stream-to-static (efficient)

Small static dimensions broadcast naturally — no special config needed.

CREATE OR REFRESH STREAMING TABLE sales_enriched AS
SELECT s.sale_id, s.product_id, s.amount, p.product_name, p.category
FROM STREAM(bronze_sales) s
LEFT JOIN dim_products p ON s.product_id = p.product_id;

Python: sales = spark.readStream.table("bronze_sales") / products = spark.read.table("dim_products") (static, broadcastable) / sales.join(products, "product_id", "left").

Rule: keep static dimensions small (< 10K rows) so they broadcast.

Stream-to-stream (stateful, time-bounded)

Always bound by event-time interval. Without bounds, state grows unbounded.

CREATE OR REFRESH STREAMING TABLE orders_with_payments AS
SELECT o.order_id, o.amount AS order_amount, p.payment_id, p.amount AS payment_amount
FROM STREAM(bronze_orders) o
INNER JOIN STREAM(bronze_payments) p
  ON o.order_id = p.order_id
 AND p.payment_time BETWEEN o.order_time AND o.order_time + INTERVAL 1 HOUR;

Python: same shape, time-bound predicate as (p.payment_time >= o.order_time) & (p.payment_time <= o.order_time + F.expr("INTERVAL 1 HOUR")).

---

Query Optimization

Filter early — push filters into the streaming read so downstream MV inputs stay small. The anti-pattern is wide-open silver tables filtered later in gold MVs — every row is processed twice.

CREATE OR REFRESH STREAMING TABLE silver_recent AS
SELECT * FROM STREAM(bronze_events)
WHERE event_date >= CURRENT_DATE() - INTERVAL 7 DAYS;

*Skip `SELECT `** once schema is stable. Narrow projections enable Delta column pruning and shrink wire/state size for stateful operations.

---

Pre-Aggregation

When the same coarse aggregation is queried frequently, materialize it.

CREATE OR REFRESH MATERIALIZED VIEW orders_monthly AS
SELECT customer_id, YEAR(order_date) AS year, MONTH(order_date) AS month,
       SUM(amount) AS total
FROM large_orders_table
GROUP BY customer_id, YEAR(order_date), MONTH(order_date);

Querying orders_monthly is far cheaper than re-aggregating the underlying table.

---

Compute Configuration

Default to serverless. Switch to classic only when R, Spark RDD APIs, JAR/Maven libraries, or other serverless-incompatible features are required — see pipeline-configuration.md.

---

Monitoring Freshness

SELECT table_name,
       MAX(event_timestamp) AS latest_event,
       TIMESTAMPDIFF(MINUTE, MAX(event_timestamp), CURRENT_TIMESTAMP()) AS lag_minutes
FROM pipeline_monitoring.table_metrics
GROUP BY table_name;

Watch for slow streaming tables (high processing lag), large state ops (memory), expensive joins (long batch durations), small-file accumulation (raise auto-optimize).

---

Common Issues

Pipeline Configuration

JSON field reference for databricks pipelines create --json '{...}' and databricks pipelines update <id> --json '{...}', plus variant snippets for common configurations.

Defaults to serverless + Unity Catalog. Don't set serverless: false unless the user explicitly needs R, Spark RDD APIs, or JAR / Maven libraries.

Canonical Create (dev / iteration defaults)

For dev, demo, and iteration work, always pass these fields:

databricks pipelines create --json '{
  "name": "my_pipeline",
  "catalog": "my_catalog",
  "schema": "my_schema",
  "serverless": true,
  "continuous": false,
  "development": true,
  "channel": "PREVIEW",
  "configuration": {
    "pipelines.numUpdateRetryAttempts": "0",
    "pipelines.maxFlowRetryAttempts": "0"
  },
  "libraries": [{"glob": {"include": "/Workspace/Users/<user>/my_pipeline/**"}}]
}'

Tuned for demo / iteration. The pipelines.*RetryAttempts: "0" overrides disable retries so a broken update fails fast (~30s) instead of retrying for 10+ min on the same root cause. For production, drop these overrides so the platform's retry defaults (5 update / 2 flow) absorb transient infra failures.

Per-field rationale:

• `continuous: false` — triggered runs. true auto-restarts failed updates forever (cause: RETRY_ON_FAILURE), burning cost and trapping polling loops. Only true when the user explicitly asks for always-on streaming.
• `development: true` — faster startup, relaxed validation, no retry-on-failure. Required for any edit/re-run loop.
• `pipelines.numUpdateRetryAttempts: "0"` + `maxFlowRetryAttempts: "0"` — belt-and-suspenders against retries. Even with development, some configs still retry. Drop for prod.
• `channel: "PREVIEW"` — latest features. "CURRENT" (default) for production stability.

Variant snippets below show only the deltas to add/replace in the canonical JSON.

---

Top-Level Fields

Edition Comparison

---

clusters[] — Classic Cluster Config

Required when serverless: false. Each cluster object:

---

event_log — Custom Event Log Table

---

notifications[] — Alerts

---

configuration — Spark / Pipeline Config

All values must be strings.

Any key here is also accessible from pipeline code via spark.conf.get("key") — use this to parameterize transformations.

---

run_as — Execution Identity

Only one of these:

---

restart_window — Continuous-Pipeline Restart Window

For continuous pipelines, the 5-hour window when daily restarts may occur:

---

filters — Path Filtering

---

environment — Serverless Python Deps

---

deployment — Deployment Method

---

Variant Snippets

Each block shows what to add to (or replace in) the canonical create JSON.

Production mode (remove dev defaults)

The canonical create above is tuned for iteration. For production, remove "development": true and the two pipelines.*RetryAttempts overrides so the platform's retry defaults (5 / 2) can absorb transient infra failures. Add ownership tags:

"channel": "CURRENT",
"tags": {"environment": "production", "owner": "data-team"}

Switch "channel" to "CURRENT" for stable runtime behavior.

Non-serverless / dedicated cluster

Required only for R, Spark RDD APIs, or JAR/Maven libraries.

"serverless": false,
"photon": true,
"edition": "ADVANCED",
"clusters": [{
  "label": "default",
  "autoscale": {"min_workers": 2, "max_workers": 8, "mode": "ENHANCED"},  // or "num_workers": 4 for fixed
  "node_type_id": "i3.xlarge",
  "spark_conf": {"spark.sql.adaptive.enabled": "true"},
  "custom_tags": {"environment": "production"}
}]

Continuous streaming

"continuous": true,
"configuration": {"spark.sql.shuffle.partitions": "auto"}

Email notifications

"notifications": [{
  "email_recipients": ["team@example.com", "oncall@example.com"],
  "alerts": ["on-update-failure", "on-update-fatal-failure", "on-flow-failure"]
}]

Serverless Python dependencies

"environment": {
  "dependencies": ["scikit-learn==1.3.0", "pandas>=2.0.0", "requests"]
}

Continuous with restart window

Combine "continuous": true with:

"restart_window": {
  "start_hour": 2,
  "days_of_week": ["SATURDAY", "SUNDAY"],
  "time_zone_id": "America/Los_Angeles"
}

Custom event-log location

"event_log": {
  "catalog": "audit_catalog",
  "schema": "pipeline_logs",
  "name": "my_pipeline_events"
}

---

Updating an Existing Pipeline

update takes the same JSON shape as create:

databricks pipelines update <pipeline_id> --json '{
  "name": "updated_name",
  "development": false,
  "notifications": [{"email_recipients": ["team@example.com"], "alerts": ["on-update-failure"]}]
}'

Then trigger a new run with databricks pipelines start-update <pipeline_id> [--full-refresh]. See 2-rapid-iteration-with-cli.md for the polling pattern — never poll top-level pipelines get state for run completion.

---

Multi-Schema Patterns

Preferred: one pipeline, multiple schemas via fully-qualified table names. Simpler than running multiple pipelines. For trivial cases where all tables share one schema, use name prefixes (bronze_*, silver_*, gold_*).

Set pipeline defaults to one schema (e.g. bronze); pull the rest from configuration:

silver_schema = spark.conf.get("silver_schema")  # add silver_catalog too for cross-catalog
gold_schema   = spark.conf.get("gold_schema")

@dp.table(name="orders_bronze")                                  # uses pipeline default schema
def orders_bronze(): ...

@dp.table(name=f"{silver_schema}.orders_clean")                  # other schema, same catalog
def orders_clean(): ...

@dp.materialized_view(name=f"{gold_schema}.orders_by_date")
def orders_by_date(): ...

For cross-catalog: use three-part f"{cat}.{schema}.{table}" in name=. SQL uses the same fully-qualified form in CREATE OR REFRESH ....

---

Platform Constraints

Serverless requirements

Serverless limitations (force classic clusters)

General constraints

Python Basics

Setup

• from pyspark import pipelines as dp — required at the top. Legacy import dlt still parses but should be migrated (see SKILL.md Legacy DLT Syntax).
• spark (SparkSession) is pre-imported in pipeline files. In utility modules, import it normally.

Core decorators

• @dp.materialized_view() — batch table. See materialized-view-python.md.
• @dp.table() — streaming table when the function returns a streaming DataFrame. (Returns-batch-DataFrame is legacy DLT shape — use @dp.materialized_view instead.) See streaming-table-python.md.
• @dp.temporary_view() — pipeline-scoped view. See temporary-view-python.md.
• @dp.expect*() — quality constraints. See expectations-python.md.
• @dp.append_flow(target=..., once=...) — fan multiple sources into one target. See streaming-table-python.md.
• @dp.foreach_batch_sink() — custom per-batch Python sink (Public Preview). See foreach-batch-sink-python.md.

Core functions

• dp.create_streaming_table() — empty target for @dp.append_flow / dp.create_auto_cdc_flow. See streaming-table-python.md.
• dp.create_auto_cdc_flow() / dp.create_auto_cdc_from_snapshot_flow() — CDC. See auto-cdc-python.md.
• dp.create_sink() — external Delta / Kafka / Event Hubs sinks. See sink-python.md.

Reading datasets

• Batch sibling table: spark.read.table("name").
• Streaming sibling table: spark.readStream.table("name").
• Never use the LIVE. prefix — fully deprecated, errors in modern pipelines.
• dp.read() / dp.read_stream() are legacy — always use spark.read.table(...) / spark.readStream.table(...).

Critical rules

• ✅ Dataset functions return a Spark DataFrame.
• ✅ Use the modern auto_cdc API, not apply_changes.
• ✅ Look up parameter docs when unsure — many decorators have nuanced options.
• ❌ Never call .collect(), .count(), .toPandas(), .save(), .saveAsTable(), .start(), .toTable() inside a dataset function. The pipeline owns the write side.
• ❌ No custom monitoring or side effects in dataset functions — they may be evaluated multiple times. Keep them pure DataFrame definitions.
• ❌ No star imports.

skipChangeCommits

When a downstream streaming table reads from an upstream streaming table that has updates/deletes (GDPR purges, Auto CDC targets), set skipChangeCommits to ignore the change commits — without it, they cause errors:

@dp.table()
def downstream():
    return spark.readStream.option("skipChangeCommits", "true").table("upstream_table")

Querying SCD Type 2 Tables

How to read SCD Type 2 history tables produced by Auto CDC: current-state views, point-in-time queries, change analysis, and joining facts with historical dimensions. SQL is shown as canonical; Python translates via spark.read.table(...).filter(F.col("__END_AT").isNull()) etc.

For the CDC flow that writes these tables, see auto-cdc-python.md / auto-cdc-sql.md.

Temporal Columns

SCD Type 2 tables (from stored_as_scd_type=2 / STORED AS SCD TYPE 2) include two system columns:

Both have the same type as the SEQUENCE BY / sequence_by column (usually TIMESTAMP).

Rule of thumb: WHERE __END_AT IS NULL selects only current rows. That's the most common filter — bake it into a materialized view if you query it often.

Current State

CREATE OR REFRESH MATERIALIZED VIEW dim_customers_current AS
SELECT customer_id, customer_name, email, phone, address,
       __START_AT AS valid_from
FROM dim_customers
WHERE __END_AT IS NULL;

For a single entity: WHERE customer_id = '12345' AND __END_AT IS NULL.

Point-in-Time Queries

State as it existed on a specific date. Boundary convention: [__START_AT, __END_AT) — start inclusive, end exclusive. Get this wrong and you'll either drop the seam row or double-count it.

CREATE OR REFRESH MATERIALIZED VIEW products_as_of_2024_01_01 AS
SELECT product_id, product_name, price, category, __START_AT, __END_AT
FROM products_history
WHERE __START_AT <= '2024-01-01'
  AND (__END_AT > '2024-01-01' OR __END_AT IS NULL);

Change Analysis

All versions of one entity

SELECT customer_id, customer_name, email, phone,
       __START_AT, __END_AT,
       COALESCE(DATEDIFF(DAY, __START_AT, __END_AT),
                DATEDIFF(DAY, __START_AT, CURRENT_TIMESTAMP())) AS days_active
FROM dim_customers
WHERE customer_id = '12345'
ORDER BY __START_AT DESC;

Changes within a period (excluding the original version per entity)

SELECT customer_id, customer_name,
       __START_AT AS change_timestamp,
       'UPDATE'   AS change_type
FROM dim_customers c
WHERE __START_AT BETWEEN '2024-01-01' AND '2024-03-31'
  AND __START_AT != (SELECT MIN(__START_AT) FROM dim_customers c2
                     WHERE c2.customer_id = c.customer_id)
ORDER BY __START_AT;

Joining Facts with Historical Dimensions

As-of-transaction-time (canonical for revenue-correct gold)

For each fact row, pick the dimension version that was active at the transaction's event time.

CREATE OR REFRESH MATERIALIZED VIEW sales_with_historical_prices AS
SELECT s.sale_id, s.product_id, s.sale_date, s.quantity,
       p.product_name,
       p.price AS unit_price_at_sale_time,
       s.quantity * p.price AS calculated_amount,
       p.category
FROM sales_fact s
INNER JOIN products_history p
  ON s.product_id = p.product_id
 AND s.sale_date  >= p.__START_AT
 AND (s.sale_date < p.__END_AT OR p.__END_AT IS NULL);

With the current dimension (ignore history)

When attributes are labels (always-current product name, region label), not values that drive the math.

CREATE OR REFRESH MATERIALIZED VIEW sales_with_current_prices AS
SELECT s.sale_id, s.product_id, s.sale_date, s.quantity,
       s.amount        AS amount_at_sale,
       p.product_name  AS current_product_name,
       p.price         AS current_price
FROM sales_fact s
INNER JOIN products_history p
  ON s.product_id = p.product_id
 AND p.__END_AT IS NULL;

When to use which: as-of-time for revenue, billing, and audit; current-dim for operational dashboards where attributes are labels.

Optimization

Pre-filter into MVs for repeated queries on history tables:

CREATE OR REFRESH MATERIALIZED VIEW dim_products_current AS
SELECT * FROM products_history WHERE __END_AT IS NULL;

CREATE OR REFRESH MATERIALIZED VIEW product_change_stats AS
SELECT product_id, COUNT(*) AS version_count,
       MIN(__START_AT) AS first_seen, MAX(__START_AT) AS last_updated
FROM products_history
GROUP BY product_id;

Cluster the history table on lookup key + time: CLUSTER BY (product_id, __START_AT). Accelerates both entity lookups and point-in-time scans. See performance.md#cluster-key-selection-by-layer.

Best Practices

1. Filter `__END_AT IS NULL` for "current" — never compare __START_AT against MAX(__START_AT) per entity. Slower and breaks under concurrent updates. 2. Inclusive-lower / exclusive-upper for point-in-time joins (__START_AT <= D AND (__END_AT > D OR __END_AT IS NULL)). 3. Materialize repeated filters. A dim_*_current MV is cheaper than re-filtering history on every downstream read. 4. High-precision `SEQUENCE BY`. Sub-second collisions cause non-deterministic ordering — use microsecond timestamps or STRUCT(ts, tiebreaker). 5. `TRACK HISTORY ON` only columns that need versions on wide tables (other columns get Type-1 in-place updates without creating new history rows).

Common Issues