Partial snapshot replacement with REPLACE USING flows

Beta

This feature is in Beta.

A REPLACE USING flow keeps a target table in sync with a streaming source: it replaces all rows that match the specified key columns and leaves all other data unchanged.

A SEQUENCE BY column orders the updates so the result is correct even when updates arrive out of order. For each key, the highest sequence wins, and a lower-sequence row never overwrites a higher one already in the target. Rows that share the same key and the same sequence are appended rather than replaced.

How REPLACE USING works

Consider an events table that holds click and conversion events for two regions, sequenced by seq:

region_id	device_type	event_type	seq
1	iOS	click	1
1	Android	conversion	1
2	iOS	click	1
2	desktop	click	1

region_id	device_type	event_type	seq
1	iOS	click	1
1	Android	conversion	1
2	iOS	click	1
2	desktop	click	1

A REPLACE USING (region_id) SEQUENCE BY seq flow receives these updates for regions 1 and 3. Region 2 has no updates:

region_id	device_type	event_type	seq
1	iOS	click	2
1	Android	conversion	2
1	desktop	click	2
3	iOS	click	1
3	desktop	click	2

region_id	device_type	event_type	seq
1	iOS	click	2
1	Android	conversion	2
1	desktop	click	2
3	iOS	click	1
3	desktop	click	2

The target becomes:

region_id	device_type	event_type	seq	Outcome
1	iOS	click	2	Replaced, because seq 2 is greater than seq 1
1	Android	conversion	2	Replaced, because seq 2 is greater than seq 1
1	desktop	click	2	Replaced, because seq 2 is greater than seq 1
2	iOS	click	1	Untouched, because the key isn't present in this update
2	desktop	click	1	Untouched, because the key isn't present in this update
3	desktop	click	2	Added. The seq 1 row for region 3 isn't added, because only the highest sequence for a key is applied.

region_id	device_type	event_type	seq	Outcome
1	iOS	click	2	Replaced, because seq 2 is greater than seq 1
1	Android	conversion	2	Replaced, because seq 2 is greater than seq 1
1	desktop	click	2	Replaced, because seq 2 is greater than seq 1
2	iOS	click	1	Untouched, because the key isn't present in this update
2	desktop	click	1	Untouched, because the key isn't present in this update
3	desktop	click	2	Added. The seq 1 row for region 3 isn't added, because only the highest sequence for a key is applied.

Requirements

REPLACE USING flows have the following requirements:

• REPLACE USING flows run on Databricks Runtime 18.2 and above, on classic or serverless compute. Databricks recommends Unity Catalog.
• The source must be a streaming source. REPLACE USING rejects a non-streaming source.
• You must specify at least one key column and exactly one SEQUENCE BY column.

When to use REPLACE USING flows

Lakeflow pipelines offer three flows that overwrite existing rows. Choose based on what your source looks like and how it identifies the rows to replace:

• Use REPLACE USING when your source is a series of partial snapshots keyed by column. REPLACE USING overwrites only the data that has a match in the incoming data, leaving all other data untouched. It doesn't require a primary key.
• Use AUTO CDC when your source is a change data capture (CDC) feed with explicit insert, update, and delete operations, or you need slowly changing dimension (SCD) Type 2 history. AUTO CDC also requires a true primary key. See The AUTO CDC APIs: Simplify change data capture with pipelines.
• Use REPLACE WHERE when your source is a snapshot and you want to recompute and overwrite a range of the target table selected by a predicate, for example the last 7 days, as a batch operation. It doesn't require a primary key. See Batch processing with REPLACE WHERE flows.

Create a REPLACE USING flow

Define REPLACE USING flows in either SQL or Python.

SQL

Use the FLOW REPLACE USING clause inline with CREATE STREAMING TABLE:

SQL

CREATE STREAMING TABLE payments_current
FLOW REPLACE USING (payment_id) SEQUENCE BY payment_date BY NAME
SELECT payment_id, booking_id, status, payment_date
FROM STREAM(samples.wanderbricks.payments);

Alternatively, use the long-form CREATE FLOW syntax:

SQL

CREATE STREAMING TABLE payments_current;

CREATE FLOW payments_flow AS
INSERT INTO payments_current BY NAME
REPLACE USING (payment_id) SEQUENCE BY payment_date
SELECT payment_id, booking_id, status, payment_date
FROM STREAM(samples.wanderbricks.payments);

note

BY NAME is required in SQL. It matches columns by name rather than position.

Python

Declare the table and the flow together with @dp.table:

Python

from pyspark import pipelines as dp

@dp.table(name="payments_current", replace_using=["payment_id"], sequence_by="payment_date")
def payments_current():
  return spark.readStream.table("samples.wanderbricks.payments")

Alternatively, target an existing streaming table with @dp.replace_flow:

Python

from pyspark import pipelines as dp

dp.create_streaming_table("payments_current")

@dp.replace_flow(target="payments_current", replace_using=["payment_id"], sequence_by="payment_date")
def payments_flow():
  return spark.readStream.table("samples.wanderbricks.payments")

replace_using is a list of key columns. sequence_by is a column name or a Column expression, and is required whenever replace_using is set.

Sequencing and out-of-order data

The SEQUENCE BY column makes the result independent of the order in which updates arrive. A row is applied to a key only if its sequence is greater than the sequence already stored for that key, so a late or replayed row that is older than the current value is ignored. Keys not present in an update are left untouched.

Follow these practices so replacement behaves predictably:

Practice	Reason
Use a sequence that strictly increases per key version, such as a timestamp, version number, or log offset.	Two rows with the same key and the same sequence are both kept, which results in duplicate rows for that key.
Use a non-null sequence.	A null sequence can lead to undefined behavior.

Practice	Reason
Use a sequence that strictly increases per key version, such as a timestamp, version number, or log offset.	Two rows with the same key and the same sequence are both kept, which results in duplicate rows for that key.
Use a non-null sequence.	A null sequence can lead to undefined behavior.

Expectations

REPLACE USING flows support expectations. warn and fail behave as they do on other flows: warn keeps violating rows and records the violation, and fail stops the update. See Manage data quality with pipeline expectations.

A drop expectation treats a violating row as if the source never produced it. The dropped row doesn't replace, delete, or modify matching keys in the target table:

• Dropping happens before deduplication, so the flow keeps the latest valid version for the key.
• If every incoming row for a key is dropped, the key's existing rows are left untouched.
• Because a dropped row sets no sequence floor, a later valid update still lands even if its sequence is lower than the dropped row's.

Limitations

REPLACE USING flows have the following limitations:

• REPLACE USING supports a single flow per target table. Combining REPLACE USING with another flow type on the same target isn't supported.
• The target table must be created within the pipeline.
• The source must be a streaming source.
• You must specify at least one key column and a SEQUENCE BY column. Key columns can't be repeated, and each key column's type must be sortable. Atomic types, such as integers, strings, and dates, can be keys, while MAP and VARIANT can't.

Examples

The following examples read from samples.wanderbricks.booking_updates, a sample table of booking state changes that is available in every Unity Catalog-enabled workspace. Each booking appears once per change, so booking_id repeats with a new booking_update_id. See Wanderbricks dataset.

Example 1: Keep the latest record for each key

Keep only the current state of each booking. The flow keys on booking_id and sequences by booking_update_id, so the most recent update for a booking replaces its earlier ones. Use AUTO CDC instead when your source is a change feed with explicit insert, update, and delete operations.

SQL

SQL

CREATE OR REFRESH STREAMING TABLE bookings_current
FLOW REPLACE USING (booking_id) SEQUENCE BY booking_update_id BY NAME
SELECT booking_id, status, total_amount, booking_update_id
FROM STREAM(samples.wanderbricks.booking_updates);

Python

Python

from pyspark import pipelines as dp

@dp.table(
  name="bookings_current",
  replace_using=["booking_id"],
  sequence_by="booking_update_id"
)
def bookings_current():
  return spark.readStream.table("samples.wanderbricks.booking_updates")

This example sequences by booking_update_id rather than the updated_at timestamp because several updates to the same booking can share a timestamp. Rows that tie on the sequence are appended rather than replaced, which would leave more than one row for those bookings.

Example 2: Key on more than one column

When a record is identified by a combination of columns, list them all in REPLACE USING. Here each booking is identified by (property_id, booking_id), so the flow keeps the current state of every booking per property. If a key column can be null, REPLACE USING matches null to null rather than skipping the row.

SQL

SQL

CREATE OR REFRESH STREAMING TABLE bookings_by_property
FLOW REPLACE USING (property_id, booking_id) SEQUENCE BY booking_update_id BY NAME
SELECT property_id, booking_id, status, total_amount, booking_update_id
FROM STREAM(samples.wanderbricks.booking_updates);

Python

Python

from pyspark import pipelines as dp

@dp.table(
  name="bookings_by_property",
  replace_using=["property_id", "booking_id"],
  sequence_by="booking_update_id"
)
def bookings_by_property():
  return spark.readStream.table("samples.wanderbricks.booking_updates")

Example 3: Drop invalid records with an expectation

Add an expectation to keep bad rows out of the target. A dropped row is treated as if the source never produced it: it doesn't replace or delete the matching key, and the flow falls back to the latest valid row for that key. This flow drops updates that don't have a positive total_amount.

Python

from pyspark import pipelines as dp

@dp.table(
  name="bookings_validated",
  replace_using=["booking_id"],
  sequence_by="booking_update_id"
)
@dp.expect_or_drop("positive_amount", "total_amount > 0")
def bookings_validated():
  return spark.readStream.table("samples.wanderbricks.booking_updates")