Lakeflow Spark Declarative Pipelines Development

databricks-core

Decision Tree

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

• "Create a table" without specifying type → ask whether the source is streaming or batch
• Materialized View from streaming source is an error → use a Streaming Table instead, or switch to a batch read
• Streaming Table from batch source is an error → use a Materialized View instead, or switch to a streaming read
• Aggregation over streaming table → use a Materialized View with batch read (

  spark.read.table

  /

  SELECT FROM

  without

  STREAM

  ), NOT a Streaming Table. This is the correct pattern for Gold layer aggregation.
• Aggregation over batch/historical data → use a Materialized View, not a Streaming Table. MVs recompute or incrementally refresh aggregates to stay correct; STs are append-only and don't recompute when source data changes.
• Preprocessing before Auto CDC → use a Temporary View to filter/transform the source before feeding into the CDC flow. SQL: the CDC flow reads from the view via

  STREAM(view_name)

  . Python: use

  spark.readStream.table("view_name")

  .
• Intermediate logic → default to Temporary View → Use a Temporary View for intermediate/preprocessing logic, even when reused by multiple downstream tables. Only consider a Private MV/ST (

  private=True

  /

  CREATE PRIVATE ...

  ) when the computation is expensive and materializing once would save significant reprocessing.
• View vs Temporary View → Persistent Views publish to Unity Catalog (SQL only), Temporary Views are pipeline-private
• Union of streams → use multiple Append Flows. Do NOT present UNION as an alternative — it is an anti-pattern for streaming sources.
• Changing dataset type → cannot change ST→MV or MV→ST without manually dropping the existing table first. Full refresh does NOT help. Rename the new dataset instead.
• SQL

  OR REFRESH

  → Prefer

  CREATE OR REFRESH

  over bare

  CREATE

  for SQL dataset definitions. Both work identically, but

  OR REFRESH

  is the idiomatic convention. For PRIVATE datasets:

  CREATE OR REFRESH PRIVATE STREAMING TABLE

  /

  CREATE OR REFRESH PRIVATE MATERIALIZED VIEW

  .
• Kafka/Event Hubs sink serialization → The

  value

  column is mandatory. Use

  to_json(struct(*)) AS value

  to serialize the entire row as JSON. Read the sink skill for details.
• Multi-column sequencing in Auto CDC → SQL:

  SEQUENCE BY STRUCT(col1, col2)

  . Python:

  sequence_by=struct("col1", "col2")

  . Read the auto-cdc skill for details.
• Auto CDC supports TRUNCATE (SCD Type 1 only) → SQL:

  APPLY AS TRUNCATE WHEN condition

  . Python:

  apply_as_truncates=expr("condition")

  . Do NOT say 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, explicitly clarify this and suggest switching to Python.
• MV incremental refresh → Materialized Views on serverless pipelines support automatic incremental refresh for aggregations. Mention the serverless requirement when discussing incremental refresh.
• Recommend ONE clear approach → Present a single recommended approach. Do NOT present anti-patterns or significantly inferior alternatives — it confuses users. Only mention alternatives if they are genuinely viable for different trade-offs.

Publishing Modes

• 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.

Comprehensive API Reference

• Auto Loader → also read the streaming-table skill (Auto Loader produces a streaming DataFrame, so the target is a streaming table) and look up format-specific options for the file format being loaded
• Auto CDC → also read the streaming-table skill (Auto CDC always targets a streaming table)
• Sinks → also read the streaming-table skill (sinks use streaming append flows)
• Expectations → also read the corresponding dataset definition skill to ensure constraints are correctly placed

Dataset Definition APIs

@dp.table()

@dlt.table()

CREATE OR REFRESH STREAMING TABLE

CREATE STREAMING LIVE TABLE

@dp.materialized_view()

@dlt.table()

CREATE OR REFRESH MATERIALIZED VIEW

CREATE LIVE TABLE

@dp.temporary_view()

@dlt.view()

@dp.view()

CREATE TEMPORARY VIEW

CREATE TEMPORARY LIVE VIEW

CREATE VIEW

dp.create_streaming_table()

dlt.create_streaming_table()

CREATE OR REFRESH STREAMING TABLE

Flow and Sink APIs

@dp.append_flow()

@dlt.append_flow()

CREATE FLOW ... INSERT INTO

@dp.append_flow(once=True)

@dlt.append_flow(once=True)

CREATE FLOW ... INSERT INTO ... ONCE

dp.create_sink()

dlt.create_sink()

@dp.foreach_batch_sink()

CDC APIs

dp.create_auto_cdc_flow()

dlt.apply_changes()

dp.apply_changes()

AUTO CDC INTO ... FROM STREAM

APPLY CHANGES INTO ... FROM STREAM

dp.create_auto_cdc_from_snapshot_flow()

dlt.apply_changes_from_snapshot()

Data Quality APIs

@dp.expect()

@dlt.expect()

CONSTRAINT ... EXPECT (...)

@dp.expect_or_drop()

@dlt.expect_or_drop()

CONSTRAINT ... EXPECT (...) ON VIOLATION DROP ROW

@dp.expect_or_fail()

@dlt.expect_or_fail()

CONSTRAINT ... EXPECT (...) ON VIOLATION FAIL UPDATE

@dp.expect_all({})

@dlt.expect_all({})

CONSTRAINT

@dp.expect_all_or_drop({})

@dlt.expect_all_or_drop({})

DROP ROW

@dp.expect_all_or_fail({})

@dlt.expect_all_or_fail({})

FAIL UPDATE

Reading Data APIs

spark.read.table("name")

dp.read("name")

dlt.read("name")

SELECT ... FROM name

SELECT ... FROM LIVE.name

spark.readStream.table("name")

dp.read_stream("name")

dlt.read_stream("name")

SELECT ... FROM STREAM name

SELECT ... FROM STREAM LIVE.name

spark.readStream.format("cloudFiles")

STREAM read_files(...)

spark.readStream.format("kafka")

STREAM read_kafka(...)

spark.readStream.format("kinesis")

STREAM read_kinesis(...)

spark.readStream.format("pubsub")

STREAM read_pubsub(...)

spark.readStream.format("pulsar")

STREAM read_pulsar(...)

spark.readStream.format("kafka")

STREAM read_kafka(...)

spark.read.format("postgresql")

spark.read[Stream].format("custom")

spark.read.format("json"|"csv"|...).load()

read_files(...)

.option("skipChangeCommits", "true")

read_stream("name", skipChangeCommits => true)

Table/Schema Feature APIs

cluster_by=[...]

CLUSTER BY (col1, col2)

cluster_by_auto=True

CLUSTER BY AUTO

partition_cols=[...]

PARTITIONED BY (col1, col2)

table_properties={...}

TBLPROPERTIES (...)

schema="col1 TYPE, ..."

(col1 TYPE, ...) AS

schema="..., col TYPE GENERATED ALWAYS AS (expr)"

col TYPE GENERATED ALWAYS AS (expr)

row_filter="ROW FILTER fn ON (col)"

WITH ROW FILTER fn ON (col)

schema="..., col TYPE MASK fn USING COLUMNS (col2)"

col TYPE MASK fn USING COLUMNS (col2)

private=True

CREATE PRIVATE ...

Import / Module APIs

from pyspark import pipelines as dp

import dlt

dp

dlt

spark.read.table()

spark.readStream.table()

dp.read()

dp.read_stream()

dlt.read()

dlt.read_stream()

spark.*

LIVE.

CREATE LIVE TABLE

CREATE LIVE VIEW

CREATE STREAMING TABLE

CREATE MATERIALIZED VIEW

CREATE TEMPORARY VIEW

Language-specific guides

Scaffolding a New Pipeline Project

databricks bundle init

<project_name>/

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

• project_name

  : letters, numbers, underscores only

• language

  :

  python

  or

  sql

  . Ask the user which they prefer:
  • SQL: Recommended for straightforward transformations (filters, joins, aggregations)
  • Python: Recommended for complex logic (custom UDFs, ML, advanced processing)

CLAUDE.md

AGENTS.md

# Declarative Automation Bundles Project

This project uses Declarative Automation Bundles (formerly Databricks Asset Bundles) for deployment.

## Prerequisites

Install the Databricks CLI (>= v0.288.0) if not already installed:
- macOS: `brew tap databricks/tap && brew install databricks`
- Linux: `curl -fsSL https://raw.githubusercontent.com/databricks/setup-cli/main/install.sh | sh`
- Windows: `winget install Databricks.DatabricksCLI`

Verify: `databricks -v`

## For AI Agents

Read the `databricks-core` skill for CLI basics, authentication, and deployment workflow.
Read the `databricks-pipelines` skill for pipeline-specific guidance.

If skills are not available, install them: `databricks experimental aitools skills install`

Pipeline Structure

• Follow the medallion architecture pattern (Bronze → Silver → Gold) unless the user specifies otherwise
• Use the convention of 1 dataset per file, named after the dataset
• Place transformation files in a

  src/

  or

  transformations/

  folder

my-pipeline-project/
├── databricks.yml                        # Bundle configuration
├── resources/
│   ├── my_pipeline.pipeline.yml          # Pipeline definition
│   └── my_pipeline_job.job.yml           # Scheduling job (optional)
└── src/
    ├── my_table.py (or .sql)             # One dataset per file
    ├── another_table.py (or .sql)
    └── ...

Scheduling Pipelines

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}

Running Pipelines

databricks bundle deploy

• Selective refresh is preferred when you only need to run one table. For selective refresh it is important that dependencies are already materialized.
• Full refresh is the most expensive and dangerous option, and can lead to data loss, so it should be used only when really necessary. Always suggest this as a follow-up that the user explicitly needs to select.

Development Workflow

1. Validate:

   databricks bundle validate --profile <profile>

2. Deploy:

   databricks bundle deploy -t dev --profile <profile>

3. Run pipeline:

   databricks bundle run <pipeline_name> -t dev --profile <profile>

4. Check status:

   databricks pipelines get --pipeline-id <id> --profile <profile>

Pipeline API Reference

• Write Spark Declarative Pipelines — Core syntax and rules (Python, SQL)
• Streaming Tables — Continuous data stream processing (Python, SQL)
• Materialized Views — Physically stored query results with incremental refresh (Python, SQL)
• Views — Reusable query logic published to Unity Catalog (SQL)
• Temporary Views — Pipeline-private views (Python, SQL)
• Auto Loader — Incrementally ingest files from cloud storage (Python, SQL)
• Auto CDC — Process Change Data Capture feeds, SCD Type 1 & 2 (Python, SQL)
• Expectations — Define and enforce data quality constraints (Python, SQL)
• Sinks — Write to Kafka, Event Hubs, external Delta tables (Python)
• ForEachBatch Sinks — Custom streaming sink with per-batch Python logic (Python)