One set of transforms, two paths

Most streaming systems grow a second code path the first time something goes wrong. The stream misses a day, someone writes a batch job to “fix yesterday”, and from then on the two paths drift: the backfill fixes yesterday and quietly reintroduces last month’s bug. LakeFlow was built so that this cannot happen.

What the pipeline does

PostgreSQL runs with logical WAL and REPLICA IDENTITY FULL, so Debezium emits complete before/after images for every insert, update and delete. Kafka (KRaft) carries one topic per table. Spark Structured Streaming appends the raw envelope to an immutable Bronze layer — checkpointed, partitioned by table and ingest date, exactly-once at the sink.

Silver is derived inside foreachBatch: latest-per-key on the WAL commit order (ts_ms, then lsn), deletes honoured, and the customers table versioned as SCD Type 2. Rows that fail declarative rules land in a quarantine table with the names of the rules they failed. Gold marts — daily revenue, product ranks, customer lifetime value — are Spark SQL.

The one decision that holds it together

Every Silver and Gold transform is a plain function from DataFrame to DataFrame. The stream calls them once per micro-batch. The Airflow backfill calls the same functions over a range of Bronze partitions. Nothing in the transforms knows which path invoked it, so the seven pytest tests exercise the logic once and cover both.

def to_silver(bronze: DataFrame) -> DataFrame:
    w = Window.partitionBy("id").orderBy(F.col("ts_ms").desc(), F.col("lsn").desc())
    latest = bronze.withColumn("rn", F.row_number().over(w)).filter("rn = 1")
    return latest  # deletes carry op == "d" and are applied by upsert()

# streaming path — one micro-batch at a time
(bronze_stream.writeStream
    .foreachBatch(lambda df, _: upsert(to_silver(df)))
    .option("checkpointLocation", ckpt)
    .start())

# backfill path — the same function over a partition range (Airflow)
upsert(to_silver(spark.read.parquet(bronze_path).where(ingest_day.between(start, end))))

Why replay is safe

Upserts are keyed by the business key and ordered by commit position, so replaying the same offsets — or re-running a backfill over the same days — produces identical tables. If a micro-batch fails halfway, the checkpoint replays it and the upsert absorbs the duplicates. Idempotency is a property of the design, not a retry policy.

Measured on 2 vCPUs: 510,663 CDC events replayed end-to-end into Gold in 29.5 s (~17.3K events/s), with 5,239 rows quarantined — each with the rule that rejected it.

The throughput is a by-product. The thing that holds is the discipline: one transform set, explicit schemas, a test for every transform, and metrics for every run.