This route decides how each source reaches bronze, and where the quality gate lives. Pick the pattern from the shape of the source and ingestion stays reviewable; pick it per source and you get six bespoke pipelines and six people who each understand one. The second decision — that validation does not live in the pipeline that publishes — is the one nobody misses until a wrong number arrives through a chain of green jobs.
Stand 29.07.2026
One decision table
There are four source shapes. The shape is a property of the source, not a choice you get to make.
The question that selects the pattern is not which system is this. It is can I tell what changed, and can I tell what was deleted? Connector, format and vendor all follow from those two answers.
Source shape → pattern → what it costs to get wrong
Source shape
Pattern
Lands in
Cost of the wrong choice
Files in a Volume, schema drifts
Auto Loader streaming table, schema location per stream
bronze.shop_order
a folder listing that double-counts any file landing twice
Change feed with a sequence (CDC)
AUTO CDC flow, sequence_by the change counter
bronze.sap_vbak, bronze.sap_vbap
a MERGE keyed on a timestamp: same-second changes apply in arbitrary order
Full extract, no change timestamp
append every extract, diff consecutive snapshots
bronze.sap_kna1
no history at all, or every unchanged row versioned daily
Dated reference file, insert-only
Auto Loader, unique on (currency, rate_date)
bronze.fx_rate
one duplicated rate row fans out the join and doubles amount_eur
API with a paging cursor
land the raw response as files, then pattern one
a Volume, then bronze
a parse bug you cannot re-run — the API will not return that page again
Why here
The failure this prevents is pattern proliferation: six sources, six notebooks, each with its own idea of restart, deduplication and lateness. Nothing is broken and nothing is reviewable — the author of source three is the only person who can fix source three.
The subtler failure is choosing a pattern one shape too weak. The webshop feed starts as just read the folder each night, which holds until a file lands late, lands twice, or a run dies half-written. Each is a silent wrong number, and the fix is a backfill rather than an option change.
Doing it
Answer both questions in the ticket first: can I tell what changed, can I tell what was deleted.
Land raw before parsing for anything you cannot re-request.
Give each stream its own schema location and checkpoint; never share a path.
Add the three technical columns in the framework, so a new source cannot forget them.
Embedding it
Hand the team the next new source with only this table and no instructions. Which pattern they pick, and why, is the whole assessment.
When someone proposes a bespoke ingestion, ask which shape it is and what breaks in that one. Usually the answer is 'nothing — it is how the last project did it'.
Have them write the row for the next source, cost column included. Writing that column is where the thinking happens.
Defaults
Four patterns, each selected by an observable property of the source.
Raw lands before it is parsed; parsing belongs downstream of a durable copy.
Bronze is append-only and carries the three technical columns.
One stream, one schema location, one checkpoint.
Gotchas
Choosing CDC because the source system offers CDC, when what arrives is a nightly full dump. sequence_by is then identical for every row in the file, and last-writer-wins picks an arbitrary version of each customer.
Reading a Volume with spark.read 'because there are only a few files'. It works for a year; then the folder holds 400,000 files, the nightly listing takes six hours, and you migrate under pressure.
Parsing during ingestion. A field the shop team changed on Tuesday is unrecoverable for Monday, because bronze holds your interpretation of the file rather than the file.
Assuming a full extract lets you skip history. It is the only shape that reveals deletions — and the only one where a truncated file is indistinguishable from a mass delete.
The three patterns this warehouse needs
The webshop drops files. SAP sends a change feed for orders and a full dump for customers. The ECB rates file is the Auto Loader pattern again with the volume turned down — one small CSV a day, unique on (currency, rate_date) — so it earns a row in the table above and no section of its own.
All three run as Lakeflow pipelines, and the naming moved twice while most code on the internet did not. Delta Live Tables is now Lakeflow pipelines, running Spark Declarative Pipelines. Existing DLT code keeps working — but the import and the decorators changed, and the decorator change is not cosmetic.
pythonThe import, and the decorator split
# old — still runs
import dlt
# current
from pyspark import pipelines as dp
# @dp.table -> a STREAMING TABLE
# @dp.materialized_view -> a materialized view
# @dp.temporary_view -> the former @dlt.view
Under DLT, @table produced whichever object the query implied; it now always declares a streaming table. A snippet copied from a 2024 post therefore runs, builds the wrong object, and announces itself as a cost line rather than an error.
The webshop writes one JSON document per order into a Volume and changes its shape on its own release cadence. The layout below is the only part of this interface we control, so it is fixed before the first file lands.
textVolume layout — landing, schema and checkpoint state are siblings, never nested
/Volumes/<catalog>/bronze/landing/shop_order/
dt=2026-07-28/
shop_order_2026-07-28T02-15-04Z_0001.json
shop_order_2026-07-28T02-15-04Z_0002.json
dt=2026-07-29/
...
/Volumes/<catalog>/bronze/_schemas/shop_order/ <- cloudFiles.schemaLocation
/Volumes/<catalog>/bronze/_checkpoints/shop_order/ <- stream checkpoint
<catalog> is dwh_dev | dwh_test | dwh_prod — never a literal in code
pythonsrc/dwh/pipelines/bronze_shop_order.py
from pyspark import pipelines as dp
from pyspark.sql import functions as F
CATALOG = spark.conf.get("dwh.catalog")
LANDING = f"/Volumes/{CATALOG}/bronze/landing/shop_order"
SCHEMA_LOC = f"/Volumes/{CATALOG}/bronze/_schemas/shop_order"
BATCH_ID = spark.conf.get("dwh.batch_id", "manual")
@dp.table(
name="shop_order",
comment="One row per webshop order document. Raw; technical columns only.",
)
def bronze_shop_order():
return (
spark.readStream.format("cloudFiles")
.option("cloudFiles.format", "json")
.option("cloudFiles.schemaLocation", SCHEMA_LOC)
.option("cloudFiles.schemaEvolutionMode", "addNewColumns")
.option("cloudFiles.inferColumnTypes", "true")
.option(
"cloudFiles.schemaHints",
"order_id STRING, order_ts TIMESTAMP, total_amount DECIMAL(18,2)",
)
.option("rescuedDataColumn", "_rescued_data")
.load(LANDING)
.select(
"*",
F.current_timestamp().alias("_tech_loaded_at"),
F.col("_metadata.file_path").alias("_tech_source_file"),
F.lit(BATCH_ID).alias("_tech_batch_id"),
)
)
Inside a pipeline the schema location may be omitted — the pipeline manages that state. It is explicit here because the same reader runs from a plain job in CI, where it is mandatory, and because knowing where the state lives separates reprocessing on purpose from reprocessing by accident.
What each evolution mode does with a column it has never seen
schemaEvolutionMode
New column arrives
Use for
addNewColumns (default with a schema location)
fails once, records the column, succeeds on retry
bronze, where you want the column and can absorb a retry
rescue
never fails; the value lands in _rescued_data as JSON
a source adding fields weekly that nothing may block
failOnNewColumns
fails, and stays failed until someone changes the code
an interface contract you want broken loudly
none
the column is ignored, silently, forever
almost never — this is how you lose data with a green pipeline
Keep rescuedDataColumn on even with addNewColumns: evolution handles new columns, the rescue column also catches type mismatches and case collisions.
Why here
The alternative is written in an afternoon — list the folder, read what is new, remember the high-water mark — and is wrong in three ways nobody notices for months. A file that lands late is never read. A file that lands twice is read twice. A run that dies after writing but before updating the mark re-reads what it committed. The checkpoint answers all three.
The second reason is arithmetic: at a few thousand files, listing is free; at several hundred thousand it costs more than reading the data, and file-event notification is not something to discover at 03:00 mid-backfill.
Doing it
Fix the Volume layout before the first file: dt= subfolders, one document per file, timestamp and sequence in the name.
Take the catalog from pipeline configuration. A /Volumes/dwh_prod path left in a dev pipeline reads production files and writes them into a dev table, and nothing errors.
Set schemaHints for every column downstream code names by hand. Inference is a sample, and samples lie.
Keep _rescued_data, and alert on it being non-null rather than on it existing.
Measure listing time weekly; move to file-event notification before file count becomes a problem.
Embedding it
Have them add a field to a sample file in dev and watch the stream fail once and recover by itself. The retry is the lesson; explaining it does not land.
In review, ask where the checkpoint lives and what happens if it is deleted. A pipeline whose owner cannot answer has an owner in name only.
Make the shop team's schema change a ticket on their board rather than a firefight on yours. The coaching here is cross-team.
Defaults
One stream, one schema location, one checkpoint — never shared, never reused.
addNewColumns in bronze; rescue only where blocking is unacceptable; never none.
Land the raw document; parse in silver.
maxFilesPerTrigger on the first backfill, so it does not size the cluster for two years.
Gotchas
Sharing a schema location between two streams. The schemas merge, each stream starts seeing the other's columns, and unpicking it means a full re-ingest of both. It looks tidy when you write it.
Deleting the checkpoint to 'reprocess'. Bronze is append-only, so every file is read again and every row exists twice. Nothing errors; the discovery is a stakeholder remarking that revenue looks unusually strong.
schemaEvolutionMode = none. New fields are dropped silently and the pipeline stays green; the failure surfaces when someone asks for a field that has 'been in the feed for months'.
Inferring types with no hints. A first batch where total_amount is always absent produces no such column at all, and the file that finally carries it restarts the stream mid-backfill.
SAP sends change records for order headers (VBAK) and items (VBAP). Bronze appends them verbatim, SAP field names intact. Silver applies them in order — which is what AUTO CDC does, the API previously spelled APPLY CHANGES INTO and dlt.apply_changes.
pythonsrc/dwh/pipelines/silver_sales_order.py — rename, then apply
from pyspark import pipelines as dp
from pyspark.sql import functions as F
@dp.temporary_view(name="vbak_typed")
def vbak_typed():
"""SAP names in, canonical names out. Renaming only — no logic."""
return spark.readStream.table("bronze.sap_vbak").selectExpr(
"VBELN AS order_id",
"KUNNR AS customer_id",
"ERDAT AS order_date",
"LOEKZ AS deletion_flag",
"change_type",
"change_seq",
)
dp.create_streaming_table(
name="sales_order",
comment="One row per order, current state (SCD1). Applied from SAP VBAK CDC.",
)
dp.create_auto_cdc_flow(
target="sales_order",
source="vbak_typed",
keys=["order_id"],
sequence_by=F.col("change_seq"),
apply_as_deletes=F.expr("change_type = 'D' OR deletion_flag = 'X'"),
except_column_list=["change_type", "change_seq", "deletion_flag"],
stored_as_scd_type=1,
)
# The item side is the same shape. Its source view, vbap_typed, is the one that
# carries logic (rename + currency conversion) and is shown in full under
# "The transform is a function" below. VBAP has no LOEKZ, so an item is deleted
# by a change record only — the deletion flag is a header concern.
dp.create_streaming_table(name="sales_order_line", comment="One row per order line (SCD1).")
dp.create_auto_cdc_flow(
target="sales_order_line",
source="vbap_typed",
keys=["order_id", "line_number"],
sequence_by=F.col("change_seq"),
apply_as_deletes=F.expr("change_type = 'D'"),
except_column_list=["change_type", "change_seq"],
stored_as_scd_type=1,
)
sequence_by alone decides which of two changes to the same order wins, so it must increase strictly per key — the source's change counter or log sequence number. If the feed offers only a timestamp, sequence by a struct of (timestamp, counter), or make the source team add one.
AUTO CDC needs the PRO pipeline edition; expectations need ADVANCED, which is the default and covers both. Downgrading to CORE fails with an explanatory error rather than skipping the CDC silently — a good failure arriving at the worst moment, which is deployment.
Why here
The hand-rolled MERGE reviews well and is wrong at exactly two moments. When a batch holds two changes to the same key, MERGE fails outright on multiple source matches. When a delete arrives before the insert it deletes, the merge applies them in file order and leaves a row that should not exist.
The second reason is optionality: moving silver.sales_order to SCD2 later is stored_as_scd_type=2 plus a backfill, not a rewrite of merge logic eleven tables depend on.
Doing it
Confirm what the feed contains before designing: physical deletes, a deletion flag, or neither. With SAP it is usually a flag.
Sequence by a monotonic counter. If the feed has none, that is a written requirement on the source team.
Strip every control column with except_column_list so change_type never reaches silver.
Keep header and line as two independent flows. The join belongs in gold, where the unknown member exists.
Embedding it
Have the team replay yesterday's change file twice and assert the target is identical. That one test converts CDC from folklore into a property.
Ask what happens to an order deleted in SAP that is already in last month's revenue. It is a business decision, and the team should take it to the business.
Let them own the conversation with the SAP team about the sequence column. Owning the interface is the real handover.
Defaults
sequence_by a strictly increasing per-key value; never a bare timestamp.
apply_as_deletes reads the source's real deletion signal, flag or record type.
SCD1 in silver for orders; history belongs to the gold dimensions.
ADVANCED edition, so CDC and expectations need no second decision.
Gotchas
sequence_by on a second-granularity timestamp. Two changes in the same second apply in whatever order the engine chose that run, and re-running picks a different winner — non-determinism no test catches unless the test replays.
Missing the deletion flag. SAP marks an order deleted rather than removing it, so apply_as_deletes reading only change_type = 'D' leaves deleted orders alive in silver and inside revenue forever. Rule 1 covers only CANCELLED.
Joining VBAK to VBAP in silver. Lines arrive without their header often enough to matter: the inner join drops them, the outer join yields rows with no customer, and neither is an error.
Assuming a full refresh replays CDC correctly. It replays whatever bronze still holds — if bronze was vacuumed or truncated, the rebuilt silver is missing changes and looks entirely healthy.
#Snapshot comparison — bronze.sap_kna1, the hard case
What
KNA1 arrives as a full extract once a day with no change timestamp. Nothing in the file says what changed: the change is the difference between today's file and yesterday's, and it has to be computed rather than read.
So bronze keeps every extract in full — one row per customer per extract — and silver.customer is the SCD2 built by comparing consecutive snapshots. This source carries a fourth technical column, _tech_extract_date, the extract's business date from the filename.
pythonsrc/dwh/pipelines/silver_customer.py
from pyspark import pipelines as dp
from pyspark.sql import functions as F
dp.create_streaming_table(
name="customer",
comment="One row per customer per validity interval (SCD2), from snapshot comparison.",
)
def next_kna1_snapshot(latest_version):
"""(DataFrame, version) for the next unprocessed extract, or None when caught up."""
days = [
r._tech_extract_date
for r in spark.read.table("bronze.sap_kna1")
.select("_tech_extract_date")
.distinct()
.orderBy("_tech_extract_date")
.collect()
]
pending = [d for d in days if latest_version is None or d > latest_version]
if not pending:
return None
day = pending[0]
df = (
spark.read.table("bronze.sap_kna1")
.filter(F.col("_tech_extract_date") == day)
.selectExpr( # SAP names in, canonical names out
"KUNNR AS customer_id",
"NAME1 AS customer_name",
"STRAS AS street",
"PSTLZ AS postal_code",
"ORT01 AS city",
"LAND1 AS country",
"KUKLA AS customer_segment",
)
)
return df, day
dp.create_auto_cdc_from_snapshot_flow(
target="customer",
source=next_kna1_snapshot,
keys=["customer_id"],
stored_as_scd_type=2,
track_history_column_list=["customer_segment"],
)
track_history_column_list is where business rule 4 lives: a change of customer_segment opens a new version, an address change updates in place. Without it every corrected postcode becomes an interval, and the three segment moves that mattered sit under nine hundred that did not.
Absence is the signal only a full extract carries: a customer_id present yesterday and missing today was deleted in SAP. The snapshot flow applies that as a close-out; a hash-diff against current silver never sees it, because it only looks at rows that are present.
Why here
The naive version — overwrite silver.customer from today's file — discards every question beginning as of. Which segment was this customer in when the order was placed becomes unanswerable, and it is asked in month four by whoever owns the sales targets.
The other naive version — hash every column, version on any difference — is expensive and wrong. It records address corrections as business history, so the dimension grows fast while carrying less signal than before.
Doing it
Carry the extract's business date into bronze and order snapshots by it — never by _tech_loaded_at, which changes when you re-run.
Take the SCD2 clock from the extract date. current_timestamp() makes a backfill rewrite history with today's date.
Agree the tracked column list with the business once and write it into the concept. It is rule 4, not a setting.
Quarantine an extract more than a few percent smaller than the previous one — do not process it, do not drop it.
Embedding it
Give the team yesterday's file with 40 % of the rows removed and let them run it. Nobody who has watched dim_customer close out half its customers ships without the count gate again.
Ask which columns are tracked and why, then route the disagreement to the business instead of settling it yourself.
Have them run the same extract twice and assert no new version appears. Idempotency on a snapshot pattern is rarely tested.
Defaults
Bronze keeps every extract in full — the only record of what the source said that day.
Order snapshots by the extract's business date, never by load time.
track_history_column_list encodes the business rule; everything else updates in place.
Deletions close the interval; the row is never removed.
Gotchas
Hash-diffing against current silver instead of against the previous snapshot. Deletions become invisible: a customer removed from SAP stays current forever, and the only symptom is a segment count that never goes down.
Hashing all columns. A postcode correction across 3,000 rows creates 3,000 new intervals; the dimension grows without carrying information and every as-of join gets slower for nothing.
Taking _tech_valid_from from current_timestamp(). Re-running a week of extracts on Friday stamps them all Friday, intervals overlap, and every as-of query returns duplicates. The clock is an input, not an ambient fact.
Vacuuming bronze.sap_kna1 to save storage. This is the one pattern where bronze retention is a correctness requirement: silver cannot be rebuilt from snapshots you deleted.
Two structural decisions that determine whether ingestion can be tested at all, and whether bad data can reach a dashboard. Neither is about ingestion technology, and both are lost by default.
#The transform is a function; the expectations are a dict
What
A decorated function cannot be unit-tested: importing the module registers a dataset, and calling the function needs a pipeline context. The fix is not a test framework, it is a two-file split — the logic in a module that imports nothing from the pipeline API, and a three-line wrapper that does.
pythonsrc/dwh/pipelines/transforms/silver_sales_order.py — pure, importable, no pipeline API
from pyspark.sql import DataFrame, functions as F
EXPECTATIONS = {
"line_key_present": "order_id IS NOT NULL AND line_number IS NOT NULL",
"known_currency": "currency IN ('EUR', 'CHF')",
"rate_resolved": "fx_rate IS NOT NULL",
"amount_positive": "amount_eur > 0 OR change_type = 'D'",
"order_not_ancient": "order_ts >= TIMESTAMP'2020-01-01 00:00:00'",
}
def to_silver_sales_order(lines: DataFrame, rates: DataFrame) -> DataFrame:
"""SAP item fields in, canonical names out, valued in EUR at the order date.
lines: bronze.sap_vbap change records.
rates: silver.fx_rate — (currency, rate_date, fx_rate), one row per day.
Business rule 2: amount_eur = ROUND(quantity * unit_price * fx_rate, 2).
"""
return (
lines.selectExpr(
"VBELN AS order_id",
"POSNR AS line_number",
"MATNR AS product_id",
"KWMENG AS quantity",
"NETWR / NULLIF(KWMENG, 0) AS unit_price",
"WAERK AS currency",
"CAST(ERDAT AS TIMESTAMP) AS order_ts",
"change_type",
"change_seq",
)
.withColumn("rate_date", F.to_date("order_ts"))
.join(rates, on=["currency", "rate_date"], how="left")
.withColumn(
"amount_eur",
F.round(F.col("quantity") * F.col("unit_price") * F.col("fx_rate"), 2),
)
.drop("rate_date")
)
pythonsrc/dwh/pipelines/silver_sales_order.py — the wrapper; same module as the AUTO CDC flows above
from pyspark import pipelines as dp
from dwh.pipelines.transforms.silver_sales_order import EXPECTATIONS, to_silver_sales_order
@dp.temporary_view(name="vbap_typed")
@dp.expect_all_or_drop(EXPECTATIONS)
def vbap_typed():
return to_silver_sales_order(
spark.readStream.table("bronze.sap_vbap"),
spark.read.table("silver.fx_rate"),
)
This is the vbap_typed view the AUTO CDC flow above consumes, and the placement is deliberate: expectations cannot be attached to an AUTO CDC target, only to the flow's source. That is also where you want them — a row dropped here never becomes a version in silver.sales_order_line, whereas a row dropped after the apply would leave a gap in the change sequence.
rate_resolved earns its place: the rate join is a LEFT join, so a missing ECB rate produces fx_rate = NULL, then amount_eur = NULL, then a revenue total that is quietly short by one currency for one day. Nothing fails. amount_positive allows deletes through because a delete record carries no amount; rule 5 — positive amounts on non-cancelled lines — is completed in gold, where status exists.
The test imports the same dict. A rule is written once and asserted in two places, which is the only version of the rule is enforced that survives a refactor.
Three actions, three consequences
Decorator
Violating row
Update
Use for
expect_all
kept
succeeds
anything you want measured before you dare enforce it
expect_all_or_drop
dropped, counted in the event log
succeeds
rows unusable downstream and safe to lose
expect_all_or_fail
written nowhere
fails immediately
invariants where publishing nothing beats publishing something wrong
Expectations require the ADVANCED pipeline edition. CORE is streaming ingest only, PRO adds CDC, ADVANCED adds expectations — and ADVANCED is the default, so this only bites the team that changed the edition to save money.
sqlWhat actually happened — the event log, not the run status
SELECT
e.timestamp,
x.name AS expectation,
x.dataset,
x.passed_records,
x.failed_records
FROM event_log(TABLE(dwh_prod.silver.sales_order_line)) AS e
LATERAL VIEW explode(
from_json(
e.details:flow_progress.data_quality.expectations,
'array<struct<name: string, dataset: string,
passed_records: bigint, failed_records: bigint>>'
)
) AS x
WHERE e.event_type = 'flow_progress'
AND e.timestamp > current_timestamp() - INTERVAL 1 DAY
ORDER BY x.failed_records DESC;
Why here
Without the split, the only way to test a transform is to run a pipeline: three to eight minutes per attempt, on shared infrastructure, with a cluster start in the middle. What happens next is universal — people stop testing and start looking at the output table.
Without the shared dict, each rule exists twice, once in the pipeline and once in the test, and they drift inside a sprint. The test then proves that the test's copy of the rule is satisfied, which is a fact about the test.
Doing it
Put every transform in transforms/ with no pipeline import. Mocking does not make a decorated module testable.
Keep EXPECTATIONS beside the function producing the rows it constrains, and import it in both places.
Start every new rule at warn; promote to drop or fail after watching failed-records for a week.
Read the event log per run rather than the run status — a drop expectation is a success at job level.
Embedding it
Have the team convert one pipeline file to the two-file split and time both feedback loops themselves. The number does the persuading.
In review, ask where the rule is written. If the answer is two places, the review just found a bug that has not happened yet.
Run the over-strict exercise: set a rule that drops 5 % of valid rows and see how long anyone takes to notice from job status alone.
Defaults
transforms/ imports nothing from the pipeline API. Ever.
EXPECTATIONS is a plain dict, imported by the pipeline and by the test.
New rules land as warn; promotion is a decision with evidence behind it.
Alerts read the event log; the run status is not a quality signal.
Gotchas
expect_all_or_drop on a rule that is slightly too strict. Rows disappear, the pipeline is green, the number is quietly low — the most expensive expectation failure, because the system reports success while losing data.
Testing the pipeline module instead of the transform. Importing it registers datasets and the test needs a pipeline context; the usual response is to mock the decorator, which tests the mock.
Rules copied into the test file. They drift, always in the same direction: the test relaxes, because relaxing is what makes it pass.
Reading only the run status. A drop expectation that removed 400,000 rows and a clean run are indistinguishable from the jobs UI.
The tempting design is one pipeline: bronze, silver, a few validation tables, gold. It reads as one unit of work and it gates nothing. Inside a single pipeline, whether gold refreshes depends on the dependency graph, not the quality outcome — and a validation table that nothing reads is a table that nothing waits for.
Forcing the edge is worse. Make gold read the validation table so it 'depends' on it, and lineage now claims a fact table is derived from a row count, while every full refresh recomputes the check as part of publishing the thing it was meant to guard.
textTwo pipelines, one gate between them
dwh_ingest bronze.* -> silver.* expectations live here
|
gate one SQL task; raise_error on breach, counts in the message
|
dwh_publish silver.* -> gold.* does not start if the gate failed
sqltests/sql/gate_silver_to_gold.sql — rule 7 with a tolerance, because the unknown member is legal
-- owner: data-platform rota
-- tolerance: 0.5 %, agreed with sales ops 06.2026
WITH recent AS (
SELECT o.order_id,
c.customer_id IS NULL AS unresolved
FROM IDENTIFIER(:catalog || '.silver.sales_order') o
LEFT JOIN IDENTIFIER(:catalog || '.silver.customer') c
ON c.customer_id = o.customer_id
AND c._tech_is_current
WHERE o.order_date >= current_date() - INTERVAL 1 DAY
)
SELECT CASE
WHEN COUNT(*) = 0 THEN 'ok - no orders in window'
WHEN SUM(IF(unresolved, 1, 0)) / COUNT(*) <= 0.005 THEN 'ok'
ELSE raise_error(CONCAT(
'gate silver->gold: ',
CAST(SUM(IF(unresolved, 1, 0)) AS STRING), ' of ',
CAST(COUNT(*) AS STRING),
' orders resolve to no current customer'))
END AS gate
FROM recent;
yamlresources/jobs/dwh_daily.yml — the gate is a task, never a table
The gate exists to answer one question at 03:00: is it better to publish yesterday's gold, or today's suspect gold? In a single pipeline that question is never asked — the answer was baked in when the DAG was drawn, and it is always publish.
Splitting also separates two failure classes needing different people. An ingest failure is a source problem; a publish failure is a modelling problem. One pipeline gives both the same alert, and whoever is on call must diagnose which kind it is before acting.
Doing it
Two pipelines, one job: dwh_ingest ends at silver, dwh_publish starts at silver.
Write gates as SQL files in the repo, one assertion per file, and let a breach call raise_error so the task fails with both counts in the message.
Parameterise the catalog. A gate file with dwh_prod baked in passes the test deployment by reading production.
Give every gate a tolerance and a named owner in a comment at the top. A gate with neither gets disabled the third time it fires.
Decide per gate whether failing means stop or publish-and-alert, and put the decision in the runbook.
Embedding it
Ask the team what happens to gold when a silver expectation drops 10 % of rows. If the answer is 'the pipeline is green', they have found the gap themselves.
Have them write the first gate including the tolerance, and take that tolerance to the business for a number rather than inventing one.
Break the gate deliberately during the game day and check whether anyone can tell, from the alert alone, which pipeline to open.
Defaults
Two pipelines with a job between them; the gate is a task, never a table.
A breach raises; the message carries both counts, so the alert already says how bad it is.
Tolerances are business numbers, agreed and written down.
One alert per chain, routed to a rota — see the monitoring route.
Gotchas
A validation table inside the publishing pipeline. It computes, it goes green, and gold publishes regardless, because nothing downstream reads it. The team believes there is a gate for months; there has never been one.
expect_all_or_fail used as the gate. One malformed row from one source stops the entire update including the eleven tables that were fine — and at 03:00 nobody consciously chose to trade a wrong number for no number.
Forcing a dependency edge so gold 'waits' for validation. Lineage then shows fct_order_line derived from a quality metric, and the next person to read the DAG is misled about where the numbers come from.
A gate with no tolerance. It fires on three rows, someone re-runs the job with the check commented out, and the commented line is still there in December.
The generic platform material lives in the official docs and is not restated here. These are the pages worth having open while you work through this route.