Most data pipelines I've worked on have the same shape. Read a feed, clean it, join it to reference data, and write a few tables. And around that core sits a lot of supporting code: checking what's already loaded, deciding run order, logging row counts, and recording which rows failed validation and why.
Delta Live Tables (DLT) takes a different approach. You declare the tables you want and the rules the data must meet, and Databricks works out the dependencies, runs them in order, handles incremental processing, and records data quality results. This post builds a small DLT pipeline for insurance premium data, with the focus on expectations, the part I find most useful.
The pipeline
Three tables:
premium_raw: premium transactions as they arrive from the bordereaux feed, loaded with Auto Loader.premium: typed, validated transactions.premium_summary: monthly GWP by coverholder and year of account for reporting.
Step 1: The raw streaming table
A DLT pipeline is a notebook (Python or SQL) attached to a pipeline definition. In Python, each table is a function decorated with @dlt.table:
1: import dlt 2: from pyspark.sql import functions as F 3: 4: FEED_PATH = "abfss://feeds@mystorage.dfs.core.windows.net/premium/" 5: 6: @dlt.table(comment="Premium transactions as received from the bordereaux feed") 7: def premium_raw(): 8: return (spark.readStream 9: .format("cloudFiles") 10: .option("cloudFiles.format", "csv") 11: .option("header", True) 12: .load(FEED_PATH) 13: .withColumn("source_file", F.col("_metadata.file_path")) 14: .withColumn("loaded_at", F.current_timestamp()))
There's no checkpoint path and no schema location to manage. DLT handles both.
Step 2: Typed transactions with expectations
Expectations are rules attached to a table. Each one has a name, a SQL condition, and an action for rows that fail:
| Decorator | Rows that fail | Use it for |
|---|---|---|
@dlt.expect | Kept, and counted as failures | Things to monitor but not block |
@dlt.expect_or_drop | Dropped from the table, and counted | Rows that must not reach reporting |
@dlt.expect_or_fail | The update fails | Problems that mean the whole feed is wrong |
1: @dlt.table(comment="Typed and validated premium transactions") 2: @dlt.expect_or_fail("umr_present", "umr IS NOT NULL") 3: @dlt.expect_or_drop("valid_premium", "gross_premium IS NOT NULL") 4: @dlt.expect_or_drop("valid_inception", "inception_date IS NOT NULL") 5: @dlt.expect("expiry_after_inception", "expiry_date >= inception_date") 6: @dlt.expect("commission_in_range", "commission_pct BETWEEN 0 AND 50") 7: @dlt.expect("known_transaction_type", 8: "transaction_type IN ('NEW','RENEWAL','MTA','CANCELLATION')") 9: def premium(): 10: return (dlt.read_stream("premium_raw") 11: .select( 12: "umr", "section_no", "certificate_ref", 13: F.upper(F.trim("transaction_type")).alias("transaction_type"), 14: F.to_date("bordereau_month", "yyyy-MM-dd").alias("bordereau_month"), 15: F.to_date("inception_date", "dd/MM/yyyy").alias("inception_date"), 16: F.to_date("expiry_date", "dd/MM/yyyy").alias("expiry_date"), 17: F.col("original_ccy"), 18: F.regexp_replace("gross_premium", ",", "").cast("decimal(18,2)").alias("gross_premium"), 19: F.col("commission_pct").cast("decimal(9,4)").alias("commission_pct"), 20: "source_file", "loaded_at"))
The choice of action is a business decision as much as a technical one. A missing UMR means the feed itself is broken, so the update fails and someone looks at it. A missing premium on one row shouldn't stop everything else, so that row is dropped and counted. A commission percentage outside the usual range might be valid, so it's only monitored.
Several rules can be grouped with @dlt.expect_all_or_drop({...}), which takes a dictionary of names and conditions. That's handy when the rules live in a config file instead of in code.
Step 3: The reporting table
1: @dlt.table(comment="Monthly GWP by coverholder and year of account") 2: def premium_summary(): 3: sections = spark.read.table("prod.core.dim_section") 4: return (dlt.read("premium") 5: .join(sections, ["umr", "section_no"]) 6: .groupBy("coverholder_id", "year_of_account", "bordereau_month") 7: .agg(F.sum("gross_premium").alias("gwp"), 8: F.countDistinct("certificate_ref").alias("certificates")))
dlt.read reads a table from the same pipeline as a batch, which makes this a materialised view that DLT recomputes or updates as needed. The reference table comes from Unity Catalog with a normal spark.read.table.
The same thing in SQL
If your team prefers SQL, the whole pipeline can be written that way. Expectations become constraints:
1: CREATE OR REFRESH STREAMING TABLE premium ( 2: CONSTRAINT umr_present EXPECT (umr IS NOT NULL) ON VIOLATION FAIL UPDATE, 3: CONSTRAINT valid_premium EXPECT (gross_premium IS NOT NULL) ON VIOLATION DROP ROW, 4: CONSTRAINT expiry_after_inception EXPECT (expiry_date >= inception_date) 5: ) 6: COMMENT 'Typed and validated premium transactions' 7: AS SELECT 8: umr, section_no, certificate_ref, 9: upper(trim(transaction_type)) AS transaction_type, 10: to_date(inception_date, 'dd/MM/yyyy') AS inception_date, 11: to_date(expiry_date, 'dd/MM/yyyy') AS expiry_date, 12: CAST(regexp_replace(gross_premium, ',', '') AS DECIMAL(18,2)) AS gross_premium 13: FROM STREAM(LIVE.premium_raw);
Seeing the results
The pipeline UI shows the dependency graph, and each table shows how many rows passed and failed each expectation. The same numbers are in the pipeline's event log, which you can query to build a data quality dashboard over time:
1: SELECT timestamp, 2: details:flow_progress.data_quality.expectations AS expectations 3: FROM event_log(TABLE(prod.core.premium)) 4: WHERE event_type = 'flow_progress' 5: AND details:flow_progress.data_quality IS NOT NULL 6: ORDER BY timestamp DESC;
A chart of "rows dropped for invalid premium, per month" is a simple, useful way to show whether data quality from a feed is getting better or worse.
Keeping the dropped rows
One limitation: expect_or_drop discards failed rows. If you need to send them back to the provider, add a quarantine table with the opposite condition:
1: @dlt.table(comment="Premium rows that failed validation") 2: def premium_rejects(): 3: return (dlt.read_stream("premium_raw") 4: .withColumn("gross_premium_num", 5: F.regexp_replace("gross_premium", ",", "").cast("decimal(18,2)")) 6: .filter("gross_premium_num IS NULL OR inception_date IS NULL"))
When DLT fits, and when it doesn't
- Good fit: steady pipelines from feeds to reporting tables, where you want built-in data quality metrics and don't want to manage checkpoints and run order yourself.
- Less good: heavily procedural logic, complex reprocessing rules, or teams that already orchestrate everything with dbt or another tool. DLT is also Databricks-specific, so the code doesn't move to another platform easily.
Even if you don't adopt DLT for everything, the expectations model is worth borrowing. Naming each rule, deciding up front whether it warns, drops or fails, and tracking the counts over time is good practice in any pipeline.
