Pipelines with Lakeflow Spark Declarative Pipelines

Say what you want, let the platform run it

In lessons 1 and 2 we built layers with hand-written statements. That works for a demo and fails in production: you must order the steps, handle incremental reads, retry failures and record data quality yourself. Lakeflow Spark Declarative Pipelines move that work to the platform. You declare the datasets and the queries that define them, and the pipeline works out the order, the incremental processing and the recovery.

In this lesson we cover the three core objects (flows, streaming tables, materialised views), expectations for data quality, and how to structure a medallion pipeline. Orchestration of the pipeline as part of a larger workflow is lesson 4.

The concepts

The Databricks documentation describes Lakeflow Spark Declarative Pipelines as built on Apache Spark Declarative Pipelines, a declarative framework for batch and streaming pipelines, with added functionality on Databricks Runtime. You write them in SQL or Python. Three concepts carry most of the weight.

You can also create standalone streaming tables and materialised views in Databricks SQL or Python, outside a pipeline.

A note on names. This product was called Delta Live Tables. Older code uses import dlt and @dlt.table. The current Python interface imports the module as dp, and the documentation maps a streaming @dlt.table to @dp.table and a batch @dlt.table to @dp.materialized_view. When you read older blog posts and examples, translate them.

A medallion pipeline

The two listings below are schematic. They show the structure of a bronze, silver and gold pipeline. Verify the exact syntax in the current reference for your runtime before you copy them.

# Schematic Python pipeline. The import line is the documented one;
# check the reference for the exact decorator options in your version.
from pyspark import pipelines as dp
from pyspark.sql import functions as F
 
@dp.table                      # streaming table: reads new files incrementally
def orders_bronze():
    return (
        spark.readStream.format("cloudFiles")
             .option("cloudFiles.format", "json")
             .load("/Volumes/dev_sales/landing/orders/")
    )
 
@dp.table                      # streaming table: silver built from bronze, not from the source
def orders_silver():
    return (
        spark.readStream.table("orders_bronze")
             .select(F.col("order_id").cast("bigint"),
                     F.col("amount").cast("decimal(12,2)"),
                     F.col("order_ts").cast("timestamp"))
    )
 
@dp.materialized_view          # materialised view: aggregate for one business question
def orders_daily_gold():
    return (
        spark.read.table("orders_silver")
             .groupBy(F.to_date("order_ts").alias("order_date"))
             .agg(F.sum("amount").alias("revenue"))
    )

The documentation states two rules for these functions. A function that defines a dataset must return a Spark DataFrame, and it should not call actions such as collect(), count() or save() inside the definition.

In SQL the same pipeline reads as three statements with CREATE OR REFRESH STREAMING TABLE and CREATE OR REFRESH MATERIALIZED VIEW. We keep the Python listing as the single example here, because the exact SQL clauses for ingestion differ by source and version.

In a pipeline, Auto Loader needs no schema location and no checkpoint location. The pipeline manages both.

Expectations: data quality as part of the definition

An expectation is an optional clause in a dataset definition that checks every record against a SQL Boolean condition. Each expectation has a name that is unique within the dataset. You choose what happens when a record fails.

These are the examples from the documentation:

CONSTRAINT valid_customer_age EXPECT (age BETWEEN 0 AND 120)
CONSTRAINT valid_current_page EXPECT (current_page_id IS NOT NULL AND current_page_title IS NOT NULL) ON VIOLATION DROP ROW
CONSTRAINT valid_count EXPECT (count > 0) ON VIOLATION FAIL UPDATE

@dp.expect("valid_customer_age", "age BETWEEN 0 AND 120")
@dp.expect_or_drop("valid_current_page", "current_page_id IS NOT NULL AND current_page_title IS NOT NULL")
@dp.expect_or_fail("valid_count", "count > 0")

Python also offers expect_all, expect_all_or_drop and expect_all_or_fail that take a dictionary of names and constraints. Grouped actions are not available in SQL.

Limits to know: metrics are recorded for warn and drop, not for fail. A constraint cannot contain custom Python functions, calls to external services or subqueries. Constraints are checks on a record, so a rule that needs a lookup in another table belongs in the transformation.

Choosing the action

Decide per rule, with the data owner, and write the decision down.

• Fail for rules where bad data must never reach a consumer: a missing primary key, a negative quantity in a ledger.
• Drop for rows that are unusable but do not make the batch wrong, and only if the drop count is monitored. A silent drop is data loss.
• Warn for rules you are still learning. Watch the violation counts for a few weeks, then promote the rule to drop or fail.

This closes the loop from lesson 1: the silver query that quietly removed unusable rows becomes an expectation whose count you can see.

Structure and testing

• One pipeline per domain and environment. Dev, test, acceptance and production pipelines are deployed by the same bundle (lesson 5) with different targets.
• Keep transformations testable. Put business logic in plain functions that take and return DataFrames, and unit test them without running a pipeline.
• Treat expectations as tests that run on real data. Review them in pull requests like any other code.
• Iterate small. Iterate on a small sample in dev and promote through the environments.

Summary

Declare datasets, not steps. Use streaming tables for incremental layers and materialised views for aggregations. Put data quality into the definition with expectations, and decide per rule whether to warn, drop or fail.

Next lesson: jobs and orchestration around your pipelines.