Turn existing business rules into Databricks Lakeflow expectations
The short answer
To set up Databricks expectations on a Lakeflow pipeline for rules you already have, write each rule as one line in a Python dictionary: a name and a SQL condition that is true for a good row. Use three dictionaries to warn, drop or fail. Add the file to the pipeline as a source.
What you will have at the end
One Python file, orders_expectations.py, that you add as the source of a Lakeflow pipeline. It turns six business rules about an orders table into pipeline expectations (read 2026-11-05). Databricks describes an expectation as a SQL Boolean condition that is checked on every record passing through a pipeline, with one of three actions when a record fails: warn (the default, the record is written anyway), drop (the record is removed before it is written) or fail (the update stops).
The same file also writes a quarantine table. It holds every dropped row and names the rules that row broke, so a data steward can see what was removed and why. Lakeflow pipelines require the Premium plan (Databricks docs, read 2026-11-05).
Step 1. Write each business rule as a name and a condition
Take each rule from wherever it lives today (an ETL job, a spreadsheet, a data dictionary) and write it as a SQL condition that is true for a good row. Then decide what should happen when it is false. Put it in the dictionary for that action. The block below is the only part of the file you edit.
Python: the rules, in orders_expectations.py# ---- EDIT THIS BLOCK: your table and column names, and your business rules --------------------------
SOURCE_TABLE = "orders_raw" # the table the rules apply to
LOOKUP_TABLE = "customers_ref" # the table a key must exist in
SOURCE_KEY = "o_custkey" # the column in SOURCE_TABLE that must exist in LOOKUP_TABLE
LOOKUP_KEY = "c_custkey" # the matching column in LOOKUP_TABLE
# Each rule is: name -> SQL condition that is TRUE for a good row. The name is what the event log shows.
# A rule that stops the update. Use it only for a problem that makes every later number untrustworthy.
RULES_FAIL = {
"order_key_present": "o_orderkey IS NOT NULL",
}
# Rules that remove the row from the clean table. The same rules fill the quarantine table.
RULES_DROP = {
"price_positive": "o_totalprice > 0",
"status_valid": "o_orderstatus IN ('F', 'O', 'P')",
"order_date_present": "o_orderdate IS NOT NULL",
"customer_exists": "customer_found", # set by the join below: expectations cannot hold a subquery
}
# Rules that only count: the row is kept, the failures are recorded.
RULES_WARN = {
"priority_known": "o_orderpriority IN ('1-URGENT', '2-HIGH', '3-MEDIUM', '4-NOT SPECIFIED', '5-LOW')",
}
# ------------------------------------------------------------------------------------------------------Databricks documents the three Python decorators this file uses to apply a whole dictionary at once: expect_all, expect_all_or_drop and expect_all_or_fail. Grouping rules in a dictionary is Python only; SQL pipelines state each rule as its own CONSTRAINT clause (Databricks docs, read 2026-11-05).
Step 2. Apply the dictionaries to three datasets
The rest of the file does not change. One dataset reads the source and carries the fail rule. A second keeps the rows that pass the drop rules and counts the warn rule. A third is the quarantine: it keeps the rows that broke at least one drop rule, with a column listing which ones.
One rule needed a workaround. Databricks says an expectation cannot contain a subquery that refers to another table (Databricks docs, read 2026-11-05), so "the customer must exist" cannot be written as IN (SELECT ...). The first dataset instead joins the orders to the customer keys and adds a customer_found flag, and the rule reads that flag. The join uses equality, so a NULL key in the lookup table cannot hide a missing customer, which is the trap that a NOT IN subquery falls into.
def _passes(condition):
# A rule that evaluates to NULL is a failure, so the quarantine treats it as one.
return f"coalesce(({condition}), false)"
QUARANTINE_FILTER = "NOT (" + " AND ".join(_passes(c) for c in RULES_DROP.values()) + ")"
FAILED_RULES = "concat_ws(', ', " + ", ".join(
f"CASE WHEN NOT {_passes(c)} THEN '{n}' END" for n, c in RULES_DROP.items()) + ")"
@dp.materialized_view(comment="Source rows plus a customer_found flag; stops the update if a rule in RULES_FAIL breaks")
@dp.expect_all_or_fail(RULES_FAIL)
def orders_checked():
return spark.sql(f"""
SELECT s.*, l.{LOOKUP_KEY} IS NOT NULL AS customer_found
FROM {SOURCE_TABLE} s
LEFT JOIN (SELECT DISTINCT {LOOKUP_KEY} FROM {LOOKUP_TABLE}) l ON s.{SOURCE_KEY} = l.{LOOKUP_KEY}
""")
@dp.materialized_view(comment="Rows that pass every drop rule")
@dp.expect_all_or_drop(RULES_DROP)
@dp.expect_all(RULES_WARN)
def orders_clean():
return spark.read.table("orders_checked")
@dp.materialized_view(comment="Rows that broke at least one drop rule, with the rules they broke")
def orders_quarantine():
return (spark.read.table("orders_checked")
.filter(QUARANTINE_FILTER)
.selectExpr("*", f"{FAILED_RULES} AS failed_rules"))Databricks also documents a quarantine pattern (read 2026-11-05) that builds one table with an is_quarantined column. This file uses separate clean and quarantine tables built from the same dictionary, so a rule is written once.
To create the pipeline, use serverless compute and the Advanced product edition, which is the edition that includes expectations and which serverless pipelines require. Set the default catalog and schema to the ones that hold your two input tables, and add orders_expectations.py as the source file. Databricks says the default catalog and schema are where datasets are read from or written to when the code does not qualify them (Databricks docs, read 2026-11-05), so the bare table names in the file resolve to that schema.
Run 1. Clean data: every rule passes
The test input is 1,000 orders and the first 1,000 customers, copied from the TPC-H sample data that Azure Databricks provides. demo_data.sql builds both tables in a schema you can write to. We ran the pipeline once and read each expectation's counts from the pipeline's event log, which Databricks documents as the place to query passed_records, failed_records and dropped_records (Databricks docs, read 2026-11-05).
dataset | expectation | passed_records | failed_records ---------------------------------------------------------------------------------+--------------------+----------------+--------------- content_loop.databricks_lakeflow_expectations_from_business_rules.orders_checked | order_key_present | 1000 | 0 content_loop.databricks_lakeflow_expectations_from_business_rules.orders_clean | customer_exists | 1000 | 0 content_loop.databricks_lakeflow_expectations_from_business_rules.orders_clean | order_date_present | 1000 | 0 content_loop.databricks_lakeflow_expectations_from_business_rules.orders_clean | price_positive | 1000 | 0 content_loop.databricks_lakeflow_expectations_from_business_rules.orders_clean | priority_known | 1000 | 0 content_loop.databricks_lakeflow_expectations_from_business_rules.orders_clean | status_valid | 1000 | 0
Then we counted the rows in the three tables.
Output from our run: rows in, rows kept, rows quarantinedorders_raw | orders_clean | orders_quarantine -----------+--------------+------------------ 1000 | 1000 | 0
Run 2. Planted errors: five rows dropped, one kept with a warning
We added six rows with plant_errors.sql: a negative price, a status of X, an order for a customer that does not exist, an order with no date, an order with a priority the rules do not list, and one row that breaks two rules at once. Then we ran the pipeline again from a clean schema.
dataset | expectation | passed_records | failed_records ---------------------------------------------------------------------------------+--------------------+----------------+--------------- content_loop.databricks_lakeflow_expectations_from_business_rules.orders_checked | order_key_present | 1006 | 0 content_loop.databricks_lakeflow_expectations_from_business_rules.orders_clean | customer_exists | 1004 | 2 content_loop.databricks_lakeflow_expectations_from_business_rules.orders_clean | order_date_present | 1005 | 1 content_loop.databricks_lakeflow_expectations_from_business_rules.orders_clean | price_positive | 1005 | 1 content_loop.databricks_lakeflow_expectations_from_business_rules.orders_clean | priority_known | 1005 | 1 content_loop.databricks_lakeflow_expectations_from_business_rules.orders_clean | status_valid | 1004 | 2
Each drop rule failed the rows we planted: status_valid and customer_exists each failed two, because one planted row broke both. The warn rule priority_known failed one row. The flows table shows what that did to each dataset.
flow | status | output_rows | dropped_records ------------------------------------------------------------------------------------+-----------+-------------+---------------- content_loop.databricks_lakeflow_expectations_from_business_rules.orders_checked | COMPLETED | 1006 | 0 content_loop.databricks_lakeflow_expectations_from_business_rules.orders_quarantine | COMPLETED | 5 | 0 content_loop.databricks_lakeflow_expectations_from_business_rules.orders_clean | COMPLETED | 1001 | 5
The clean table wrote 1,001 of 1,006 rows and dropped 5. The one row that broke only the warn rule is among the 1,001: a warn rule keeps the row and records the failure. The quarantine table holds the five dropped rows, and its failed_rules column says why each one is there.
o_orderkey | failed_rules -----------+------------------------------ 900000001 | price_positive 900000002 | status_valid 900000003 | customer_exists 900000004 | order_date_present 900000006 | status_valid, customer_exists
Compare the first two runs. On clean data the quarantine table is empty. On the planted data it has one row for each planted row that broke a drop rule, and the row that broke two rules names both.
Run 3. The fail rule: one missing key stops the update
The fail rule is for a problem that makes everything after it untrustworthy. We added one order with no order key, using plant_null_key.sql, and ran the pipeline from a clean schema. The update failed.
Output from our run: the update stopped, and the error shows the recordflow | status | output_rows | dropped_records
------------------------------------------------------------------------------------+---------+-------------+----------------
content_loop.databricks_lakeflow_expectations_from_business_rules.orders_checked | FAILED | NULL | NULL
content_loop.databricks_lakeflow_expectations_from_business_rules.orders_clean | SKIPPED | NULL | NULL
content_loop.databricks_lakeflow_expectations_from_business_rules.orders_quarantine | SKIPPED | NULL | NULL
ERROR (pipeline): [EXPECTATION_VIOLATION.VERBOSITY_ALL] Flow 'content_loop.databricks_lakeflow_expectations_from_business_rules.orders_checked' failed to meet the expectation. Violated expectations: 'order_key_present'. Input data: '{}'. Output record: '{"o_orderkey":null,"o_custkey":1,"o_orderstatus":"F","o_totalprice":100.0,"o_orderdate":"1995-01-01","o_orderpriority":"1-URGENT","customer_found":true}'. Missing input data: false
The error names the broken expectation and prints the offending record, so the person who fixes the data can find it. The two datasets after the failing one were skipped, so neither wrote rows in this update. Databricks says that when a pipeline fails on an expectation violation, its code must be fixed to handle the invalid data before the pipeline is re-run (Databricks docs, read 2026-11-05). The same docs page says metrics are not recorded for a fail action. In our run the event log did show one failed record for order_key_present, so do not rely on a count being there.
Tested on
Tested on
- Platform
- Azure Databricks, Premium tier, serverless compute, Unity Catalog
- Pipeline
- Serverless Lakeflow pipeline, Advanced edition, one update per run
- Date
- 2026-11-05
- Dataset
samples.tpch: 1,000 orders and the first 1,000 customers, plus the planted rows- Code
- orders_expectations.py, demo_data.sql, plant_errors.sql, plant_null_key.sql
Use it on your own tables
To run this on your own data, open orders_expectations.py and change only the block between the line that says EDIT THIS BLOCK and the closing dashed line. Set SOURCE_TABLE and LOOKUP_TABLE to your two tables, written as catalog.schema.table unless a table sits in the pipeline's target schema. Set SOURCE_KEY and LOOKUP_KEY to the key column in each table. Then rewrite the conditions in the three rule dictionaries as SQL that is true for a good row, using your own column names, and keep the rule named customer_exists reading customer_found. Leave everything below the closing dashed line as it is.
We checked those sentences on a differently built table: other table and column names, a mixed-case column, a NULL key in the lookup table, one planted error for each drop rule and the warn rule, and one extra order with no customer at all. The update completed, the five planted rows that broke a drop rule were dropped, the row that broke the warn rule was kept, and the NULL lookup key did not hide the order for the account that does not exist.
Download: orders_expectations.py · Get the full guide (PDF)
Get the full guide as a PDF
We'll email you a link to it. Please use your work email.
Thanks, . We're sending the link to now. If it isn't there in a few minutes, check your spam or junk folder, or email info@datameaning.com.
Lakeflow expectations or Great Expectations?
They are different tools, and we ran only the first. Lakeflow expectations are clauses on the datasets of a Lakeflow pipeline and are checked on each record as the pipeline writes it (Databricks docs, read 2026-11-05). That is why they can drop a row before it is written. Great Expectations is a Python framework in which expectations are testable assertions grouped into suites, run through checkpoints that can trigger actions such as alerts (Great Expectations docs, read 2026-11-05). If your data is not produced by a Lakeflow pipeline, expectations in a pipeline cannot check it, and you need a check that runs elsewhere, such as Great Expectations. If it is, the pipeline can run the checks itself, though we did not test alerts. We did not run Great Expectations for this guide.
When this is not enough
- The rules only run when the pipeline updates. Rows that arrive in a table outside the pipeline are not checked. Our runs used one update each.
- Expectations measure and filter; they do not repair. A dropped row is gone from the clean table until someone fixes the source and the pipeline runs again. The quarantine table keeps the row and the rules it broke, but nothing sends it to its owner. We did not test alerts.
- A fail rule stops the datasets that read from it, not other pipelines or business processes. Databricks says that in a triggered pipeline a failed flow does not make other parallel flows fail, and that a validation table does not hold back downstream tables (Databricks docs, read 2026-11-05). If downstream work must not run, split the checks and the downstream work into separate pipelines.
- Rules that compare other tables or earlier loads. A condition cannot contain a subquery on another table, so the customer rule needed a join. A rule over many rows, such as duplicates or row counts, is a separate validation dataset. Databricks shows validation-table examples for row counts and primary key uniqueness.
- Deciding which rules are the business's rules. This file runs the rules you give it. Who approved each rule, and what a failure means for the business, is not decided by code.
Run your approved data quality rules natively in Databricks
Native Data Quality is a Data Meaning solution. We run your approved data quality rules natively in Snowflake or Databricks, on your own tables. See what it covers and whether it fits.