On this page

• Replace all rows that match a grouping key
  • Replace, not merge
  • Before you start
  • Declare the replace write
  • Generate the pipeline
  • Read what Lakehouse Plumber wrote
  • What you just did
  • What’s next
    • Syntax and options

Replace all rows that match a grouping key

Some sources don’t emit row-level change events. Instead, whenever something about an entity changes, they resend the entity’s complete current set of rows. For example, when a subset of records in a source file are changed and the whole file is reingested. Or when an order-management system re-exports every line of an order the moment any line is added, repriced, or removed - you want the target table to hold, for each order, exactly the lines the source most recently sent, swapping the old set for the new one, and leaving orders that weren’t included in the latest partial snapshot alone.

That is a replace, not a merge, and it is the one thing AUTO CDC can’t do.

Replace, not merge

mode: replace generates a Databricks REPLACE USING flow. For every key value present in an incoming batch, it deletes the target rows for that key and writes the batch’s rows for that key in their place. The crucial part: replace_using is a grouping key, not a primary key. It does not have to be unique, one key can own many rows, and REPLACE USING keeps every row the source sent for that key; it never collapses them into one.

That is different from mode: cdc (AUTO CDC), which merges a stream of insert/update/delete events onto a unique primary key and resolves all the changes for a key into a single current row (or a versioned SCD Type 2 history). Same streaming-table target, fundamentally different operation:

Aspect	mode: replace (REPLACE USING)	mode: cdc (AUTO CDC)
Operation	Replace a key’s rows as a whole set	Merge change events onto a key
Key	Grouping key — need not be unique	Primary key — one logical row per key
Rows per key	Many, kept exactly as sent (appended, not combined)	One current row (SCD 1) or a versioned history (SCD 2)
Several source rows for one key	All are written to the target	Collapsed / merged into a single row
Source shape	Partial snapshots — all rows for each key the batch carries (changed or not)	Change deltas (explicit insert / update / delete)
Removing rows	Implicit — a row missing from a key’s new set is deleted; absent keys are untouched	Explicit only — needs a delete event (apply_as_deletes)

A worked example makes the difference concrete — watch the line that gets removed. Order 42 has three lines in the target. Upstream, one line’s quantity changes, one new line is added, and one line is deleted, so the source resends order 42 with its complete current set (a newer updated_at):

BEFORE — target rows for order 42
  order_id  product  qty
  42        widget   2
  42        gadget   5
  42        bolt     10

BATCH — the complete current set the source resends (updated_at = 2024-06-02)
  order_id  product  qty
  42        widget   3      # qty changed
  42        bolt     10     # unchanged
  42        cable    1      # new line
  # gadget is absent — it was deleted upstream

AFTER — target rows for order 42, via REPLACE USING (replace_using=[order_id])
  order_id  product  qty
  42        widget   3
  42        bolt     10
  42        cable    1
  # the whole set was replaced; gadget is gone — deleted with no delete event,
  # purely by being absent from the batch

Three things happened in one step: widget was updated, cable inserted, and gadget deleted, and the delete is the part an upsert can’t do. AUTO CDC applies only the rows a change feed carries: hand it the same batch and it would upsert widget and cable but never learn that gadget was removed, so the stale line would linger until the source emitted an explicit delete event (apply_as_deletes). Even with a unique per-line key, AUTO CDC can’t infer a delete from an absence. REPLACE USING can, because a partial snapshot states the whole truth for its keys — so a row that drops out of the set drops out of the table.

Reach for mode: replace when the source resends complete sets keyed by a grouping column; reach for mode: cdc when it emits per-row changes against a real primary key.

Note

REPLACE USING maintains only the current state of each key; the target always reflects the latest snapshot. It cannot keep a versioned SCD Type 2 history (validity windows, superseded rows preserved over time). When you need history, use mode: cdc with scd_type: 2.

Before you start

You need a Lakehouse Plumber project (see the Get Started course for lhp init) and a streaming source of partial snapshots. This guide reads a bronze order_line_updates table that another pipeline lands — each batch carries the full set of current lines for the order_id values it contains (whether or not they changed), plus an updated_at that orders competing batches for an order.

Two rules are worth knowing before you write it. The source must be a streaming read (readMode: stream); a batch read is rejected. And the target must be served by this single replace flow: it cannot share its table with any other write action, so Lakehouse Plumber always creates the table for you (create_table is forced true).

Declare the replace write

The flowgroup has two actions: a load that streams the snapshots into a view, and a write whose write_target is a streaming_table in mode: replace. Create pipelines/order_lines_current.yaml:

pipelines/order_lines_current.yaml

# Keep a current-state order-lines table from periodic partial snapshots.
# Action chain: load (delta, streaming read of order-line snapshots) -> write (streaming_table, mode: replace).
#
# Each order has many line rows. Whenever an order changes upstream, the source
# resends the COMPLETE current set of lines for that order_id. REPLACE USING
# swaps out every existing line for the order_ids in the batch and appends the
# incoming lines as-is — order_id is a grouping key, not a unique row key, so an
# order keeps as many line rows as the source sent. Nothing is merged.
pipeline: order_lines_current
flowgroup: order_lines_current

actions:
  # 1. Load: stream the bronze order-line snapshots another pipeline lands. Each
  #    batch carries the full set of current lines for the order_ids it contains.
  - name: load_order_line_updates
    type: load
    readMode: stream
    source:
      type: delta
      catalog: "${catalog}"
      schema: "${bronze_schema}"
      table: order_line_updates
    target: v_order_line_updates

  # 2. Write: replace each order's line rows as a whole set, keyed by order_id.
  - name: write_order_lines_current
    type: write
    source: v_order_line_updates
    write_target:
      type: streaming_table
      catalog: "${catalog}"
      schema: "${silver_schema}"
      table: order_lines_current
      mode: replace
      replace_config:
        replace_using: [order_id]
        sequence_by: updated_at

mode: replace is the switch; replace_config is the whole contract, and both fields map straight to the Databricks REPLACE USING parameters:

• replace_using — the grouping key columns. Every row in a batch that carries a given key replaces that key’s rows in the target; keys not present in the batch are left untouched. Here a single order_id, but it need not be unique — an order owns many line rows and they are all kept.

• sequence_by — a single column that orders competing batches for a key, so a late-arriving stale batch never overwrites a newer one; the highest value wins. Unlike cdc_config, which also accepts a list, REPLACE USING takes exactly one sequence column.

The load reads with readMode: stream, so each new batch of snapshots flows through continuously. The ${...} tokens resolve per environment from substitutions/dev.yaml, so the same flowgroup targets dev, staging, and prod unchanged.

Generate the pipeline

Validate first, then generate:

$ lhp validate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ order_lines_current  0 files
✓ validate (0.29s)
1 validated · 0.3s

$ lhp generate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ order_lines_current  1 file
✓ generate (0.29s)
✓ format (0.01s)
✓ monitoring (0.00s)
1 pipeline generated · 1 file · 0.3s

validate resolves the tokens and checks the replace_config — that replace_using is a non-empty list and sequence_by is a single column — and that the source is a streaming read, before you commit to generating.

Read what Lakehouse Plumber wrote

Open generated/dev/order_lines_current/order_lines_current.py. This is the entire output — nothing is hidden behind a runtime:

generated/dev/order_lines_current/order_lines_current.py

# Generated by LakehousePlumber
# Pipeline: order_lines_current
# FlowGroup: order_lines_current

from pyspark import pipelines as dp

# Pipeline Configuration
PIPELINE_ID = "order_lines_current"
FLOWGROUP_ID = "order_lines_current"


# ============================================================================
# SOURCE VIEWS
# ============================================================================


@dp.temporary_view()
def v_order_line_updates():
    """Delta source: dev_catalog.bronze.order_line_updates"""
    df = spark.readStream.table("dev_catalog.bronze.order_line_updates")

    return df


# ============================================================================
# TARGET TABLES
# ============================================================================

# Create the streaming table for the replace-using flow
dp.create_streaming_table(
    name="dev_catalog.silver.order_lines_current",
    comment="Streaming table: order_lines_current",
)


# Replace-using flow: f_order_lines_current
@dp.replace_flow(
    target="dev_catalog.silver.order_lines_current",
    name="f_order_lines_current",
    replace_using=["order_id"],
    sequence_by="updated_at",
    comment="Replace-using flow to dev_catalog.silver.order_lines_current",
)
def f_order_lines_current():
    """Replace-using flow to dev_catalog.silver.order_lines_current"""
    # Streaming flow (REPLACE USING requires a streaming source)
    df = spark.readStream.table("v_order_line_updates")

    return df

There is no MERGE and no foreachBatch — and, unlike AUTO CDC, no apply-changes call either. LHP wrote two things: a dp.create_streaming_table that declares the target, and one @dp.replace_flow that replaces each key’s rows into it. Both fields of your replace_config came through as arguments — replace_using=["order_id"] and sequence_by="updated_at" — and the ${...} tokens resolved into the three-part name dev_catalog.silver.order_lines_current. The flow reads spark.readStream, because REPLACE USING requires a streaming source.

Note

@dp.replace_flow is Databricks’ REPLACE USING flow. Lakehouse Plumber decides how the flow is wired and declared; Databricks performs the per-key replacement at run time. Because the target is served by this single flow, LHP always emits the dp.create_streaming_table alongside it — you cannot add a second flow to this table.

What you just did

A short replace_config block compiled to a dp.create_streaming_table plus a @dp.replace_flow, and zero lines of set-replacement logic came from you: not the delete-the-old-set-then-insert, not the grouping by key, not the sequencing that keeps a stale batch from clobbering fresh data. You declared which key groups the rows and which column orders batches, and LHP owns the rest.

What’s next

• Merge change events onto a primary key instead. When the source emits per-row inserts/updates/deletes against a unique key and you want one current row (or SCD history) per key, use mode: cdc (AUTO CDC) — covered in its own guide.

• See the field reference. replace_using and sequence_by, the forced table creation, and the single-flow and streaming-source constraints are all listed in the write action reference.

Syntax and options

See Replace rows in a streaming table for all supported settings and defaults.