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.
Stand 29.07.2026
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.
Why
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.
How
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 (the default when you provide NO schema)
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 default when you DO provide a schema)
the column is not added; with rescuedDataColumn set the value still lands in _rescued_data, without it the value is gone
a fixed contract, and only with rescuedDataColumn set
Read the defaults precisely, because the dependency is not the one people assume. The default follows from whether you supplied a schema — addNewColumns when you did not, none when you did — and not from whether you set cloudFiles.schemaLocation. cloudFiles.schemaHints is not a schema: hints correct inferred types, they do not switch the default. The reader above sets the mode explicitly for exactly this reason.
Keep rescuedDataColumn on whatever the mode: evolution handles new columns, the rescue column also catches type mismatches and case collisions — and under none it is the difference between an unread field and a lost one.
Doing it
Fix the Volume layout before the first filedt= subfolders, one document per file, timestamp and sequence in the name.
Take the catalog from pipeline configurationA /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 handInference 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 itselfThe retry is the lesson; explaining it does not land.
In review, ask where the checkpoint lives and what happens if it is deletedA 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 yoursThe 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
cloudFiles.maxFilesPerTrigger on the first backfillThe prefix is part of the name, the default is 1000 — so it does not size the cluster for two years.
Gotchas
Sharing a schema location between two streamsThe 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.
Writing maxFilesPerTrigger without the cloudFiles. prefixUnprefixed it is a file-source option, and cloudFiles ignores it without warning — so the throttle you added to the backfill does nothing, the first trigger takes the whole two-year folder, and the run either OOMs or holds an oversized cluster for hours. The option that exists is cloudFiles.maxFilesPerTrigger.
schemaEvolutionMode = none with no rescuedDataColumnNew fields never reach the table and nothing records that they arrived; the pipeline stays green and the failure surfaces when someone asks for a field that has 'been in the feed for months'. Usually nobody chose none — a schema was passed to the reader and none is the default in that case.
Inferring types with no hintsA 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.
Why
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.
How
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.
Doing it
Confirm what the feed contains before designing: physical deletes, a deletion flag, or neitherWith SAP it is usually a flag.
Sequence by a monotonic counterIf 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 flowsThe 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 identicalThat 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 revenueIt 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 columnOwning 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 timestampTwo 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 flagSAP 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 silverLines 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 correctlyIt 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.
Why
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.
How
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"],
)
So decide once, per warehouse, and write it down:
Path A — the flow owns the interval
silver.customer carries __START_AT/__END_AT, and every read that the rest of this handbook writes against _tech_valid_from goes through one projection.
Path B — you own the interval
Via apply_scd2() from src/dwh/common/scd2.py, which produces the four _tech_* columns the data-model route specifies — and which the gold as-of joins, the primary key (customer_sk, _tech_valid_from) and the gate below are all written against.
Path A is less code; path B is the one the rest of the handbook already assumes.
sqlPath A, the whole mapping — apply it wherever gold reads silver.customer
SELECT
customer_id, customer_name, street, postal_code, city, country, customer_segment,
__START_AT AS _tech_valid_from,
__END_AT AS _tech_valid_to, -- exclusive end, same convention
__END_AT IS NULL AS _tech_is_current
FROM IDENTIFIER(:catalog || '.silver.customer');
-- _tech_hash has no equivalent: on path A the change comparison happens inside
-- the flow and is never materialised. Any check that reads the hash is path B only.
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.
Doing it
Carry the extract's business date into bronze and order snapshots by itNever by _tech_loaded_at, which changes when you re-run.
Take the SCD2 clock from the extract datecurrent_timestamp() makes a backfill rewrite history with today's date.
Agree the tracked column list with the business once and write it into the conceptIt is rule 4, not a setting.
Quarantine an extract more than a few percent smaller than the previous oneDo not process it, do not drop it.
Pick path A or path B for the interval columns before the first dimension is builtPut the choice in the design document — it decides the column names in every as-of join in the warehouse.
Put the KNA1 quality rules inside next_kna1_snapshotThe snapshot flow takes no expectations.
Embedding it
Give the team yesterday's file with 40 % of the rows removed and let them run itNobody who has watched dim_customer close out half its customers ships without the count gate again.
Ask which columns are tracked and whyThen route the disagreement to the business instead of settling it yourself.
Have them run the same extract twice and assert no new version appearsIdempotency 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
One interval contract for the whole warehouseEither the flow's __START_AT/__END_AT everywhere, or _tech_valid_from/_tech_valid_to everywhere. Never one dimension each way.
Gotchas
Adding expect_all_or_drop to this flowExpectations are not supported with AUTO CDC FROM SNAPSHOT, so the rule is not the gate it looks like — and the reflex fix, moving it onto the target streaming table, does not rescue it either, because the restriction belongs to the flow and not to the decorator. The truncated extract sails through the check that was added to catch it.
Writing the gold as-of join as BETWEEN _tech_valid_from AND _tech_valid_toBoth contracts use an EXCLUSIVE end — the successor's start is the predecessor's end — so BETWEEN matches the boundary instant in two intervals and every as-of join returns a duplicate row per changed customer, on precisely the days something changed. It is >= AND (< OR IS NULL).
Assuming the flow writes the handbook's SCD2 columnsIt writes __START_AT and __END_AT; a gold build written against _tech_valid_from fails to resolve the column, someone adds a _tech_is_current column by hand to make it compile, and now two things claim to say which row is current and only one of them is maintained.
Hash-diffing against current silver instead of against the previous snapshotDeletions become invisible: a customer removed from SAP stays current forever, and the only symptom is a segment count that never goes down.
Hashing all columnsA 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 storageThis is the one pattern where bronze retention is a correctness requirement: silver cannot be rebuilt from snapshots you deleted.