Develop Lakeflow Spark Declarative Pipelines (formerly Delta Live Tables) on Databricks. Use when building batch or streaming data pipelines with Python or SQL. Invoke BEFORE starting implementation.

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. 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](references/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. Error / symptom Cause / fix Rejection of CREATE OR REPLACE STREAMING TABLE / MATERIALIZED VIEW CREATE OR REPLACE is standard SQL, NOT SDP. Use CREATE OR REFRESH STREAMING TABLE / CREATE OR REFRESH MATERIALIZED VIEW . CLI errors on databricks fs ls /Volumes/... The dbfs: prefix is required even for UC Volume paths: databricks fs ls dbfs:/Volumes/<catalog /<schema /<volume /<path . DELTA CLUSTERING COLUMNS DATATYPE NOT SUPPORTED at first write A CLUSTER BY column is BOOLEAN / ARRAY / MAP / STRUCT / BINARY. SDP doesn't pre validate — verify with DESCRIBE before submitting. Cluster keys must be numeric / string / date / timestamp. Full type rules in [references/performance.md](references/performance.md cluster key data types). Cannot create streaming table from batch query In a streaming table query you wrote FROM read files(...) (batch). Use FROM STREAM read files(...) so Auto Loader kicks in. Column not found at ingest time schemaHints don't match the actual file schema. DESCRIBE a sample file and align the hints. Streaming reads fail with parser error Use FROM STREAM read files(...) for file ingestion and FROM stream(table) (or FROM STREAM table name — legacy DLT, prefer function form) for table to table streams. Don't mix. Pipeline stuck INITIALIZING for serverless Normal — first run takes a few minutes for cold start. Don't kill it. Materialized View doesn't incrementally refresh Automatic incremental refresh for aggregations requires serverless + Delta row tracking on the source ( delta.enableRowTracking = true ). Without both, falls back to full recompute. Mention the serverless requirement when the user asks about incremental refresh. SCD2 query returns nothing / "column not found" on START AT Lakeflow uses START AT / END AT (double underscore). Current rows: WHERE END AT IS NULL . error.exceptions[0].message missing from your events output Your jq is reading .message (which is just "Update X is FAILED"). Read error.exceptions[0].message for the real cause — see [2 rapid iteration with cli.md](references/2 rapid iteration with cli.md step 4 start an update and poll that update). 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](references/streaming table python.md) / [streaming table sql.md](references/streaming table sql.md). Expectations attach to a dataset → also read the dataset definition file (streaming table / materialized view / temporary view). Dataset Definition APIs Feature Description Python SQL Skill (Py) Skill (SQL) Streaming Table Continuous incremental processing, exactly once, append only. @dp.table() returning streaming DF CREATE OR REFRESH STREAMING TABLE [streaming table python](references/streaming table python.md) [streaming table sql](references/streaming table sql.md) Materialized View Physically stored query result, incrementally refreshed. @dp.materialized view() CREATE OR REFRESH MATERIALIZED VIEW [materialized view python](references/materialized view python.md) [materialized view sql](references/materialized view sql.md) Temporary View Pipeline private, not persisted to Unity Catalog. @dp.temporary view() CREATE TEMPORARY VIEW [temporary view python](references/temporary view python.md) [temporary view sql](references/temporary view sql.md) Persistent View (UC) Published to UC; query runs on access (no storage). N/A — SQL only CREATE VIEW — [view sql](references/view sql.md) Streaming Table (explicit) Empty target, populated by separate flows (Append Flow, AUTO CDC). dp.create streaming table() CREATE OR REFRESH STREAMING TABLE (no AS) [streaming table python](references/streaming table python.md) [streaming table sql](references/streaming table sql.md) Flow and Sink APIs Feature Description Python SQL Skill (Py) Skill (SQL) Append Flow Fan in: multiple sources → one streaming table. Use instead of UNION. @dp.append flow() CREATE FLOW ... INSERT INTO [streaming table python](references/streaming table python.md) [streaming table sql](references/streaming table sql.md) Backfill Flow One time historical load + ongoing live stream into same table. @dp.append flow(once=True) CREATE FLOW ... INSERT INTO ... ONCE [streaming table python](references/streaming table python.md) [streaming table sql](references/streaming table sql.md) Sink (Delta/Kafka/EH/custom) Write streaming output to external Delta / Kafka / Event Hubs. dp.create sink() N/A — Python only [sink python](references/sink python.md) — ForEachBatch Sink Custom per batch Python logic (merge/upsert, multi destination). Public Preview. @dp.foreach batch sink() N/A — Python only [foreach batch sink python](references/foreach batch sink python.md) — RTM update flow Real Time Mode: route a flow to a sink with sub second latency. Public Preview. @dp.update flow(target=...) N/A — Python only [real time mode](references/real time mode.md) — CDC APIs Feature Description Python SQL Skill (Py) Skill (SQL) Auto CDC (streaming source) SCD Type 1 (overwrite) or Type 2 (history) from a CDC feed. dp.create auto cdc flow() AUTO CDC INTO ... FROM STREAM [auto cdc python](references/auto cdc python.md) [auto cdc sql](references/auto cdc sql.md) Auto CDC (periodic snapshot) Compare consecutive full snapshots to detect changes. dp.create auto cdc from snapshot flow() N/A — Python only [auto cdc python](references/auto cdc python.md) — For querying SCD Type 2 history tables ( START AT / END AT , point in time, joining facts with historical dimensions), see [scd 2 querying.md](references/scd 2 querying.md). Data Quality APIs Feature Description Python SQL Skill (Py) Skill (SQL) Expect (warn) Log violations, keep all rows. @dp.expect() CONSTRAINT ... EXPECT (...) [expectations python](references/expectations python.md) [expectations sql](references/expectations sql.md) Expect or drop Drop violating rows. @dp.expect or drop() CONSTRAINT ... EXPECT (

databricks-pipelines