Processing · Chapter 15

Pipelines

Databricks offers a declarative pipeline framework, previously called Delta Live Tables and now part of Lakeflow, in which you describe the tables you want and the transformations that produce them, and the platform works out the execution graph, the incremental processing, the retries, and the data quality enforcement.

Declarative by default #

The argument for the declarative framework is not that it is easier to write. It is that the things teams reliably get wrong, such as processing order, partial failure recovery, and the difference between a full refresh and an incremental run, are handled by the framework rather than by whoever wrote the notebook.

The honest counterargument is that the framework constrains you. Highly custom logic, unusual state handling, and integrations needing fine control over execution can be awkward inside it. We run those as ordinary jobs, and we treat it as an exception that needs stating rather than a default.

Expectations are the feature that earns its keep #

Data quality expectations declared alongside the transformation are the part of the framework we would miss most, because they turn quality from a separate testing exercise into a property of the pipeline and produce metrics you can alert on.

Three enforcement modes exist and the choice matters. Warning records the violation and lets the row through, which suits signals you want to watch. Dropping removes the row, which suits a row that is genuinely meaningless. Failing stops the pipeline, which suits the case where continuing would publish something wrong.

@dlt.table(name="loan_application")
@dlt.expect_or_drop("valid_application_id", "application_id IS NOT NULL")
@dlt.expect("plausible_amount", "amount_requested BETWEEN 500 AND 250000")
@dlt.expect_or_fail("known_channel", "channel IN ('web', 'dealer', 'broker', 'branch')")
def loan_application():
    return dlt.read_stream("bronze_applications").transform(conform_application)

The dlt module name in that example is the long-standing spelling and still works. The same declarations are the basis of Spark Declarative Pipelines, which Databricks has contributed to open source Apache Spark, so newer examples in the documentation may spell the imports differently while meaning exactly the same graph.

Expectations that drop rows should drop them into a quarantine table rather than into nothing, so that "how much are we losing and why" stays answerable. The pattern we use is a warning expectation plus two derived tables, one filtered to the rows that pass and one to the rows that fail with the failing condition recorded.

Streaming tables and materialised views #

The framework distinguishes streaming tables, which process each input record once and suit append-heavy sources, from materialised views, which recompute a result and can do so incrementally when the query shape allows it.

The practical rule is that ingestion and bronze to silver movement are streaming tables, while aggregations, joins across dimensions, and anything gold-facing are materialised views. Getting this backwards produces either unnecessary recomputation or a stream that cannot express the logic you need.

The part worth knowing is that a materialised view is not always a full recomputation. When the query shape allows it, the platform maintains the result incrementally from the changes in its inputs, which is why a gold aggregate over a large silver table can refresh in seconds. Non-deterministic functions, certain window expressions, and sources without a usable change feed defeat this, and the refresh quietly becomes a full recompute instead. If a materialised view that should be cheap is expensive, that is the first thing to check, and the event log records which path each refresh took, which we come back to in detail under refreshes and backfills later in this chapter.

The competitors are not weak here, and Cloud Composer and Step Functions in particular are good at coordinating heterogeneous work across many services. The distinction is that they orchestrate steps while the declarative framework maintains tables. Orchestrating steps leaves the ordering, the incrementalisation, and the recovery semantics yours to design on every pipeline, whereas maintaining tables means you declare the result and those properties come from the framework. We keep both concepts in the estate rather than picking one, which is the subject of Orchestration.

Beyond the basics #

The features below are the ones we find in use on perhaps one estate in five, and each of them removes a category of hand-written machinery.

AUTO CDC handles changes properly, including out-of-order ones. Given a change feed, the apply-changes flow produces either a current-state table or a full slowly changing dimension of type 2, ordered by a sequence column you nominate, with deletes applied and late-arriving records placed correctly. Out-of-order delivery is what justifies it, because hand-written merges get that wrong in a way that is very hard to notice, since the table looks plausible and a handful of rows carry a stale value.

Append flows let several sources feed one table. When six regional feeds belong in one silver table, an append flow per source writing into a single streaming table avoids both a union of six readers and six near-identical tables, and each flow keeps its own checkpoint, so one region's backfill does not disturb the others.

Pipelines can be generated from configuration. Because the definitions are ordinary Python, a loop over a configuration table can emit fifty table definitions that differ only in source path and key columns. We use this for wide, repetitive ingestion estates and we are careful with it, because a generated graph is harder to read than an explicit one and a reviewer still has to see what changed.

The event log is a table, so treat it as one. Every run publishes structured events covering expectation pass and fail counts, row counts, refresh types, cluster resizes, and lineage. Pointing a dashboard and a few alerts at it is how quality metrics stop being something you only look at during an incident. Our default set is rejected rows per expectation over time, an alert on any expectation whose failure rate moves sharply, and an alert on a materialised view that switched to full recomputation.

Protect bronze from a full refresh. Setting pipelines.reset.allowed to false on a bronze table means a full refresh of the pipeline will not discard and reload it, which is the guard rail against the accident described below, where a refresh reprocesses from a source that no longer holds the history the table did.

Refresh selectively. A refresh can target named tables rather than the whole graph, so recovering one broken gold aggregate does not mean rebuilding the estate beneath it.

Publish where the tables belong. A single pipeline can publish to multiple schemas and catalogs, so the layer boundaries in The Medallion Architecture do not force one pipeline per schema. We still split pipelines along ownership lines rather than technical ones, because a pipeline is also a unit of failure and of on-call responsibility.

Serverless changes the cost conversation. Serverless pipelines remove cluster startup from every run and bill for the work rather than the uptime, which matters most for small hourly pipelines that spend a meaningful share of each run waiting for compute to arrive.

Development, testing, and the thing people skip #

Pipelines are code and they deserve tests. The framework's development mode, which keeps the cluster alive between runs and does not retry on failure, makes iteration bearable and is not what should run in production.

For testing, we separate the transformation logic from the pipeline declaration so that the logic is an ordinary function taking a DataFrame and returning one, testable with small fixtures and no cluster. The tests then run in seconds in CI rather than in minutes against real infrastructure.

Beyond unit tests, we run each pipeline in staging against a small, deliberately nasty fixture dataset containing the null keys, duplicate rows, out-of-range values, and unicode that production will eventually deliver. How that fits into a release process is covered in CI/CD.

Refreshes, backfills, and surgical rewrites #

Every table in a pipeline is brought up to date one of two ways, and the difference is operational rather than cosmetic. An incremental refresh is the ordinary path, where a streaming table processes the records that arrived since its checkpoint and a materialised view updates only the part of its result the changed inputs touch. A full refresh of a streaming table is a different animal, because it truncates the table, removes the checkpoint data, and restarts the stream with a new checkpoint, so what you get back is not the table you had but whatever the source can still supply today. Databricks names the cases that genuinely require one, including a change of source table, type, or location, a change to stateful logic such as an aggregation, a join, or a deduplication key, a schema or clustering change, and a corrupted checkpoint or an expired change log. Everything outside that list is usually somebody reaching for the largest lever in the room because a number looked wrong.

Materialised views recompute on their own terms, and as noted above the platform maintains them incrementally when the query shape allows, choosing by cost model between a row-based update, a partition overwrite, and a plain append. The shapes that defeat incremental maintenance are specific rather than mysterious, including recursive common table expressions, non-deterministic functions used outside a WHERE clause, and aggregations over FLOAT or DOUBLE columns, which are better cast to DECIMAL. The source matters as much as the query, since incremental maintenance covers Delta tables, Unity Catalog managed Iceberg tables, materialised views, and streaming tables rather than arbitrary external ones. You need not guess which path a definition will take, because EXPLAIN CREATE MATERIALIZED VIEW run against the query reports incremental update eligibility along with the specific reason when the answer is no. It confirms structural eligibility rather than promising what a given run will do, so we read it alongside the event log, where the planning information events record what actually happened as values such as ROW_BASED, FULL_RECOMPUTE, or NO_OP.

EXPLAIN CREATE MATERIALIZED VIEW
  SELECT region, order_date, sum(margin) AS margin
  FROM silver_orders GROUP BY region, order_date;

INSERT INTO TABLE gold_daily_margin
REPLACE WHERE region = 'emea'
  AND order_date BETWEEN DATE'2026-03-01' AND DATE'2026-03-03'
SELECT region, order_date, sum(margin) AS margin
FROM silver_orders
WHERE region = 'emea'
  AND order_date BETWEEN DATE'2026-03-01' AND DATE'2026-03-03'
GROUP BY region, order_date;

When a bounded slice of a table is wrong, say one region across three days after a bad upstream deployment, the choice gets presented as rebuilding everything or living with the damage, and neither is necessary. For a Delta table your own code writes, the middle ground is a selective overwrite, where INSERT INTO ... REPLACE WHERE replaces exactly the rows matching a predicate in a single atomic commit. It is a sharp instrument in both senses, and the documentation is blunt about the risk, which is that a single row landing outside the intended predicate takes its whole partition with it, so we run the select alone and reconcile the row counts before the write goes near production. For tables the framework owns we do not reach past it to patch rows by hand, because the next refresh reconciles the table back to its definition and the edit disappears. There the equivalent surgery is refreshing the named tables rather than the whole graph, with the correction expressed in the definition so that it survives.

Backfill discipline is what keeps all of this from becoming the warning above, where a full refresh reprocesses from a source that has aged out data the target still held. Before any full refresh in production we establish what the source retention actually is and compare it against the oldest row the target is meant to contain, which is why pipelines.reset.allowed set to false on bronze is a default rather than an option. Where history has to be recovered we bring it in as a separate flow into the same table while the live flow keeps running, as Real-Time and Change Data Capture describes, so the backfill is an addition rather than a rebuild and no window exists in which the table sits empty. Knowing what can be reconstructed and from where belongs to Resilience and Recovery, and the refresh outcomes worth alerting on to Observability.

Where pipelines end and orchestration begins #

A pipeline handles the dependencies inside its own graph. It does not handle the dependency between your pipeline and a file that has to arrive, a model that has to be retrained, or an external system that has to be notified. That is Orchestration, and keeping the two concerns separate is what keeps both of them comprehensible.

Table of Contents