A Complete Associate and Professional Exam Guide on CDC and CDF

Change Data Capture questions may appear simple, often combine several related concepts:

DataWithRohit · 2026-07-28 02:42 · 2 claps · 28.9 min read

#databricks #spark #apache-spark #spark-sql #databricks-certification

Wiki topics: 🔒 · Cybersecurity 🔧 · Data Engineering

A Complete Associate and Professional Exam Guide on CDC and CDF

Change Data Capture questions may appear simple, often combine several related concepts:

Change Data Capture, or CDC
Delta Change Data Feed, or CDF
Delta Lake MERGE INTO
Lakeflow AUTO CDC APIs
Streaming tables
SCD Type 1 and Type 2
Late-arriving and duplicate records

Terminology note: Delta Live Tables, or DLT, was renamed Spark Declarative Pipelines on Lakeflow. Current documentation generally uses “Lakeflow pipelines.” The AUTO CDC APIs also replaced the older APPLY CHANGES APIs. Older terminology is included because it may still appear in courses and certification questions.

The 60-Second Mental Model

Change Data Capture

CDC is the overall process of detecting inserts, updates, and deletes in a source system and delivering those changes to a target.

Change Data Feed

CDF is a Delta Lake feature that generates row-level change information between table versions.

MERGE INTO

MERGE INTO is a Delta Lake command used to manually apply inserts, updates, and deletes from a source dataset to a target Delta table.

AUTO CDC

AUTO CDC is a Lakeflow pipelines API that automatically processes ordered CDC events and supports SCD Type 1 and Type 2 targets.

AUTO CDC FROM SNAPSHOT

AUTO CDC FROM SNAPSHOT compares complete source snapshots and determines which rows were inserted, updated, or deleted. It is useful when a source does not provide a native CDC feed.

The easiest way to remember the relationship is:

CDC is the process. CDF generates change events. MERGE or AUTO CDC applies those events to a target.

Change Data Capture Fundamentals

Change Data Capture, or CDC, is the process of identifying changes made to data in a source system and delivering those changes to a target.

The captured changes can include:

New source records that must be inserted into the target
Existing source records that were updated
Source records that were deleted and must also be deleted from the target

A CDC feed usually contains row data and metadata.

The metadata commonly identifies:

The business key
Whether the operation is an INSERT, UPDATE, or DELETE
A sequence number, version, or timestamp
The logical order in which changes occurred

A CDC feed may be received through:

A continuous data stream
Database transaction logs
Kafka or another message bus
JSON or other files in cloud storage
Delta Lake Change Data Feed

CDC Example: France, Canada, USA, and India

Assume the current target table contains:

France with a vaccination rate of 0.60
Canada with a vaccination rate of 0.71

The incoming CDC feed contains:

France updated to 0.74 at 7:00 AM
France updated to 0.75 at 8:00 AM
Canada marked for deletion
USA inserted with a vaccination rate of 0.50
India inserted with a vaccination rate of 0.66

France has two updates. Therefore, the target must apply only the most recent update, which is 0.75.

Canada must be deleted.

USA and India are new records and must be inserted.

The final target contains:

France: 0.75
USA: 0.50
India: 0.66

Canada is no longer present because its delete event was applied.

A delete event does not necessarily need to contain every original column. The business key and DELETE metadata may be sufficient to identify and remove the target row.

Exam Tip: If a question describes capturing only inserts, updates, and deletes rather than repeatedly copying the entire source table, the answer is usually CDC.

Processing CDC with MERGE INTO

Delta Lake supports the MERGE INTO command for applying inserts, updates, and deletes from a source dataset to a target Delta table.

# Create dummy target and source tables
target_data = [
    (1, "Alice", 10, 100),
    (2, "Bob", 20, 200),
    (3, "Charlie", 30, 300)
]
target_df = spark.createDataFrame(target_data, ["key_field", "name", "sequence_field", "value"])
target_df.write.format("delta").mode("overwrite").saveAsTable("target_table")

source_data = [
    (1, "DELETE", 101, None, "Alice_Updated"),      # Delete Alice (sequence higher)
    (2, "UPDATE", 201, 250, "Bob_Updated"),         # Update Bob (sequence higher)
    (4, "INSERT", 1, 400, "David"),                 # Insert new row
    (3, "DELETE", 299, None, "Charlie_Updated"),    # Delete Charlie (sequence lower, should not delete)
    (5, "DELETE", 1, None, "Eve")                   # Delete for non-existing key, should not insert
]
source_df = spark.createDataFrame(source_data, ["key_field", "operation_field", "sequence_field", "value", "name"])
source_df.write.format("delta").mode("overwrite").saveAsTable("latest_source_updates")

display(spark.table("target_table"))
display(spark.table("latest_source_updates"))

MERGE INTO target_table AS t
USING latest_source_updates AS s
ON t.key_field = s.key_field
WHEN MATCHED
  AND s.operation_field = 'DELETE'
  THEN DELETE
WHEN MATCHED
  AND s.operation_field <> 'DELETE'
  AND t.sequence_field < s.sequence_field
  THEN UPDATE SET *
WHEN NOT MATCHED
  AND s.operation_field <> 'DELETE'
  THEN INSERT *

display(spark.table("target_table"))

key_field   name         sequence_field  value
2           Bob_Updated  201             250
4           David        1               400

The logic works as follows:

A matched DELETE event removes the target row.
A matched event with a newer sequence updates the target.
A non-matched INSERT or UPDATE event inserts a new row.
A DELETE event should not accidentally be inserted when its key does not exist in the target.

Syntax for insert-only merge

MERGE INTO target_table
USING soruce_table
ON merge_condition
WHEN NOT MATCHED
INSERT *

# Insert-only MERGE: only insert rows from source that do not exist in target based on key_field
spark.sql("""
MERGE INTO target_table AS t
USING latest_source_updates AS s
ON t.key_field = s.key_field
WHEN NOT MATCHED THEN
  INSERT *
""")

display(spark.table("target_table"))

key_field   name            sequence_field   value
2           Bob_Updated     201              250
4           David           1                400
1           Alice_Updated   101              null
3           Charlie_Updated 299              null
5           Eve             1                null

Specify the NOT MATCHED clause, which inserts a row when a source row does not match any target row based on the merge_condition (merge keys). Records that have the same keys as an existing record in the table will be simply ignored. key_field 2 and 4 were already existing, hence were ignored and rest were added.

Use stateful streaming deduplication with a suitable watermark and an insert-only MERGE against the target.

MERGE INTO allows to merge a set of updates, insertions, and deletions based on a source table into a target Delta table. With MERGE INTO, you can avoid inserting the duplicate records when writing into Delta tables.

To meet the requirements of retaining 30 days of historical data while preventing duplicates during nightly loads, the data engineer should use the MERGE INTO command.

The MERGE INTO statement (often referred to as an “upsert”) allows you to join a source data set with a target table and perform multiple actions based on whether a match is found.

Deduplication: You can define a unique key (e.g., transaction_id). If the ID already exists in the target table, you can choose to update the record or do nothing; if it doesn’t exist, you insert it.
Retention: Since MERGE only modifies or adds specific rows, the rest of your historical data (up to your 30-day limit) remains untouched.

The Critical MERGE Limitation

A MERGE operation can fail when multiple source rows match the same target row and attempt to modify it.

The operation is ambiguous because Delta Lake cannot determine which source row should update the target.

In the France example, both France events match the same target row. Sending both directly to MERGE can generate an exception.

The rule to remember is: Only one modifying source row should match a given target row.

Exam Tip:Deduplicate or preprocess the source before MERGE.

Runtime-specific behavior

In Databricks Runtime 16.0 and above, MERGE evaluates both the ON condition and the WHEN MATCHED conditions when determining duplicate matches.

In Databricks Runtime 15.4 LTS and below, MERGE considers only the ON condition.

Multiple matching rows may be allowed for an unconditional delete because deleting the target row multiple times is not ambiguous.

Keeping Only the Latest CDC Event

A ranking window can be used to keep the latest event for each business key.

from pyspark.sql import functions as F
from pyspark.sql.window import Window

window_spec = (
    Window
    .partitionBy("country_id")
    .orderBy(F.col("sequence").desc())
)
latest_updates = (
    cdc_feed
    .withColumn("rank", F.rank().over(window_spec))
    .filter(F.col("rank") == 1)
    .drop("rank")
)

The window:

Partitions records by the business key
Orders each key by the sequence column in descending order
Assigns rank 1 to the newest event
Keeps only the newest event before MERGE

Exam Trap: RANK can Preserve Ties. RANK assigns the same rank to tied rows. If two events have the same timestamp, both may receive rank 1. This can recreate the multiple-source-row MERGE problem.

For deterministic deduplication, use ROW_NUMBER with an additional tie-breaker.

window_spec = (
    Window
    .partitionBy("country_id")
    .orderBy(
        F.col("sequence").desc(),
        F.col("operation_number").desc()
    )
)

latest_updates = (
    cdc_feed
    .withColumn(
        "row_number",
        F.row_number().over(window_spec)
    )
    .filter(F.col("row_number") == 1)
    .drop("row_number")
)

# This will throw an exception because non-time-based window operations are not supported on streaming DataFrames.
ranked_df = (spark.readStream
                   .table("bronze")
                   .filter("topic = 'customers'")
                   .select(F.from_json(F.col("value").cast("string"), schema).alias("v"))
                   .select("v.*")
                   .filter(F.col("row_status").isin(["insert", "update"]))
                   .withColumn("rank", F.rank().over(window))
                   .filter("rank == 1")
                   .drop("rank")
             )

(ranked_df.writeStream
            .option("checkpointLocation", f"{bookstore.checkpoint_path}/ranked")
            .trigger(availableNow=True)
            .format("console")
            .start()
)

exception:
[NON_TIME_WINDOW_NOT_SUPPORTED_IN_STREAMING] Window function is not supported in RANK(ROW_TIME#11451) (as column `rank`) on streaming DataFrames/Datasets.
Structured Streaming only supports time-window aggregation using the WINDOW function. (window specification: (PARTITION BY CUSTOMER_ID ORDER BY ROW_TIME DESC NULLS LAST ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)) SQLSTATE: 42KDE

To avoid this we can use forEach logic.

from pyspark.sql.window import Window

def batch_upsert(microBatchDF, batchId):
    window = Window.partitionBy("customer_id").orderBy(F.col("row_time").desc())

    (microBatchDF.filter(F.col("row_status").isin(["insert", "update"]))
                 .withColumn("rank", F.rank().over(window))
                 .filter("rank == 1")
                 .drop("rank")
                 .createOrReplaceTempView("ranked_updates"))

    query = """
        MERGE INTO customers_silver c
        USING ranked_updates r
        ON c.customer_id=r.customer_id
            WHEN MATCHED AND c.row_time < r.row_time
              THEN UPDATE SET *
            WHEN NOT MATCHED
              THEN INSERT *
    """

    microBatchDF.sparkSession.sql(query)

query = (spark.readStream
                  .table("bronze")
                  .filter("topic = 'customers'")
                  .select(F.from_json(F.col("value").cast("string"), schema).alias("v"))
                  .select("v.*")
                  .join(F.broadcast(df_country_lookup), F.col("country_code") == F.col("code") , "inner")
               .writeStream
                  .foreachBatch(batch_upsert)
                  .option("checkpointLocation", f"{bookstore.checkpoint_path}/customers_silver")
                  .trigger(availableNow=True)
                  .start()
          )

query.awaitTermination()

Exam Tip:

If a question simply asks how to keep the latest record, a rank or row-number window may be acceptable.
If ties or strict uniqueness are mentioned, ROW_NUMBER with a deterministic tie-breaker is safer.

CDC and CDF Are Not the Same

CDC is a general data-engineering process.

CDF, is a Databricks table feature that exposes row-level changes between table versions.

CDF records:

The row data involved in the change
Whether the row was inserted, updated, or deleted
The table version containing the change
The commit timestamp

CDF is frequently used to propagate incremental changes through a multi-hop architecture:

Bronze → Silver → Gold → Downstream applications

Change Data Feed Example

Assume version 1 of a Delta table contains:

France: 0.70
Canada: 0.65
India: 0.60

At version 2:

France changes from 0.70 to 0.75
USA is inserted with a value of 0.50

CDF produces:

France 0.70 with change type update_preimage
France 0.75 with change type update_postimage
USA 0.50 with change type insert

An update normally produces two records:

update_preimage

Contains the values before the update.

update_postimage

Contains the values after the update.

Canada and India are not included because they did not change.

At version 3, Canada is deleted.

CDF then records:

Canada 0.65 with change type delete
Commit version 3
The corresponding commit timestamp

CDF Metadata Columns

The three metadata columns to memorize are:

_change_type
_commit_version
_commit_timestamp

The possible _change_type values are:

insert
update_preimage
update_postimage
delete

Exam trap: The correct column is **_commit_timestamp, not simply timestamp. The table version column is `_commit_version`**.

Querying Change Data Feed

Query from a starting version

SELECT *
FROM table_changes(
  'catalog.schema.table_name',
  start_version
);

This returns changes from the specified version through the latest available version.

Query a limited version range

SELECT *
FROM table_changes(
  'catalog.schema.table_name',
  start_version,
  end_version
);

Query using timestamps

SELECT *
FROM table_changes(
  'catalog.schema.table_name',
  start_timestamp,
  end_timestamp
);

For batch CDF queries:

The starting version or timestamp is required.
The ending boundary is optional.

Reading CDF as a Stream

cdf_stream = (
    spark.readStream
    .option("readChangeFeed", "true")
    .table("catalog.schema.source_table")
)

When a streaming CDF query starts without an explicit starting version:

The current table snapshot is returned as INSERT events.
Future changes are returned as CDF events.

A checkpoint tracks which table versions have already been processed.

If a downstream target already contains all source changes through version 75, a restarted stream can use startingVersion = 76 with a new checkpoint to avoid replaying the existing snapshot.

Enabling Change Data Feed

Legacy CDF: The Rule Most Exam Questions Expect Legacy Delta CDF must be explicitly enabled.

Enable CDF on a new table

CREATE TABLE my_table (
  id INT,
  name STRING
)
TBLPROPERTIES (
  delta.enableChangeDataFeed = true
);

Enable CDF on an existing table

ALTER TABLE my_table
SET TBLPROPERTIES (
  delta.enableChangeDataFeed = true
);

Enable CDF by default for new tables in a Spark session

spark.conf.set(
    "spark.databricks.delta.properties.defaults.enableChangeDataFeed",
    "true"
)

This configuration applies only to newly created tables. It does not create CDF history for old versions of existing tables.

Current Platform Update: Automatic CDF

Databricks also documents automatic change data feed as a Public Preview feature.

Automatic CDF computes changes at read time using row tracking or row lineage rather than materializing all change records during writes.

Its documented requirements include:

Databricks Runtime 18 or above
Supported Unity Catalog tables
Row tracking for eligible Delta tables
Row lineage for eligible Apache Iceberg v3 tables

Automatic CDF does not require the legacy delta.enableChangeDataFeed property when all requirements are satisfied.

Exam strategy

If the question uses classic Delta CDF wording or asks how to enable CDF, choose:

delta.enableChangeDataFeed = true

If the question explicitly mentions:

Automatic CDF
Databricks Runtime 18+
Unity Catalog
Row tracking or Iceberg v3 row lineage

then legacy per-table enablement may not be required.

Do not mix legacy and automatic CDF on the same table.

CDF Retention and VACUUM

CDF is not a permanent audit archive. CDF data follows the table’s retention behavior.

VACUUM can delete:

Unreferenced table data files
Change-data files
History required to query older CDF versions

Transaction-log retention can also determine which table versions remain accessible.

Therefore:

CDF history is temporary.
Querying an expired starting version can fail.
Permanent history should be copied incrementally to a separate audit table.
Retention may need to be increased if downstream consumers can fall behind.

Exam trap: VACUUM does not preserve CDF indefinitely.

When to Use CDF

Use CDF when:

Delta changes include updates and deletes.
Only a small fraction of rows changes in each batch.
An external source provides data in CDC format.
Incremental changes must be sent to a downstream application.
A downstream process must distinguish inserts, updates, and deletes.
You want incremental Bronze-to-Silver or Silver-to-Gold processing.

When Not to Use CDF

CDF may not be necessary when:

The source is strictly append-only.
The downstream target needs only new rows.
Most rows are updated during every batch.
The table is completely overwritten during every batch.
The source uses destructive full-table loads.
The requirement is to discover and ingest data from outside the lakehouse.

For external data ingestion, use an appropriate ingestion mechanism such as:

Auto Loader
Kafka
Lakeflow Connect
Another supported connector

Memory rule: Updates or deletes plus a small changed fraction usually means CDF Append-only or full-table replacement usually does not.

Streaming Deduplication and Late-Arriving Duplicates

Suppose a pipeline reads order events and uses **dropDuplicates**.

A senior data engineer says this is not enough to guarantee unique records in the target when duplicate events can arrive late.

The engineer is correct because two checks are required:

Deduplicate the incoming stream.
Prevent rows already stored in the target from being inserted again.

Deduplicate the Incoming Stream

deduplicated_orders = (
    incoming_orders
    .withWatermark("event_time", "2 hours")
    .dropDuplicatesWithinWatermark(["order_id"])
)

Structured Streaming exactly-once processing does not automatically remove duplicate business events from the source.

The application must define:

The duplicate key
The event-time column
The acceptable late-arrival duration

Important distinction

Calling **dropDuplicates on a streaming DataFrame maintains state across micro-batches.**

Calling **batch_df.dropDuplicates inside `foreachBatch` deduplicates only the current micro-batch.**

A watermark limits how long state is retained. A duplicate arriving later than the retained watermark state is not guaranteed to be removed.

AUTO CDC in Lakeflow Pipelines

AUTO CDC simplifies CDC processing in Lakeflow pipelines. It replaces much of the custom MERGE logic required to handle:

Inserts
Updates
Deletes
Late-arriving events
Out-of-order events
SCD Type 1
SCD Type 2

Current and Older API Names

Current API:

AUTO CDC INTO
create_auto_cdc_flow
AUTO CDC FROM SNAPSHOT
create_auto_cdc_from_snapshot_flow

Older API:

APPLY CHANGES INTO
apply_changes
APPLY CHANGES FROM SNAPSHOT
apply_changes_from_snapshot

The older APIs may still work, but Databricks recommends the AUTO CDC names.

AUTO CDC SQL Syntax

CREATE OR REFRESH STREAMING TABLE users_target;

CREATE FLOW users_cdc_flow AS
AUTO CDC INTO users_target
FROM stream(users_raw)
KEYS (user_id)
APPLY AS DELETE
WHEN operation = 'DELETE'
SEQUENCE BY updated_timestamp
COLUMNS * EXCEPT (
  operation,
  updated_timestamp
)
STORED AS SCD TYPE 1;

The clauses mean:

Target streaming table

The target must first be declared as a streaming table.

FROM STREAM

The source must provide streaming semantics.

KEYS

Identifies the business key used to match CDC events to target records.

APPLY AS DELETE

Identifies which source events should be treated as deletes.

SEQUENCE BY

Specifies the logical order of events.

COLUMNS

Controls which source columns are stored in the target.

STORED AS

Selects SCD Type 1 or SCD Type 2 behavior.

AUTO CDC Python Syntax

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

dp.create_streaming_table("users_target")
dp.create_auto_cdc_flow(
    target="users_target",
    source="users_raw",
    keys=["user_id"],
    sequence_by=col("updated_timestamp"),
    apply_as_deletes=expr(
        "operation = 'DELETE'"
    ),
    except_column_list=[
        "operation",
        "updated_timestamp"
    ],
    stored_as_scd_type=1
)

The equivalent core syntax from older course material may appear as:

create_auto_cdc_flow(
    target="target_table",
    source="cdc_source_table",
    keys=["key_field"],
    sequence_by=col("operation_date"),
    apply_as_deletes=expr(
        "operation_type = 'DELETE'"
    ),
    except_column_list=[
        "operation_type",
        "operation_date"
    ],
    stored_as_scd_type=1
)

In a current Lakeflow Python pipeline, the function is normally called through the pipelines module:

dp.create_auto_cdc_flow(...)

Older APPLY CHANGES Syntax

SQL

APPLY CHANGES INTO target_table
FROM stream(cdc_source_table)
KEYS (key_field)
APPLY AS DELETE
WHEN operation_type = 'DELETE'
SEQUENCE BY operation_date
COLUMNS * EXCEPT (
  operation_type,
  operation_date
)
STORED AS SCD TYPE 1;

Python

apply_changes(
    target="target_table",
    source="cdc_source_table",
    keys=["key_field"],
    sequence_by=col("operation_date"),
    apply_as_deletes=expr(
        "operation_type = 'DELETE'"
    ),
    except_column_list=[
        "operation_type",
        "operation_date"
    ]
)

SEQUENCE BY and Out-of-Order Events

SEQUENCE BY specifies the logical order of CDC events.

The order in which events arrive is not always the order in which they occurred.

For example:

An update with sequence 6 arrives.
An older update with sequence 5 arrives afterward.
The sequence-5 event must not overwrite the newer sequence-6 state.

AUTO CDC uses SEQUENCE BY to handle this automatically.

The sequencing field must:

Be sortable
Represent the logical order
Not contain null sequencing values

Sequencing with Multiple Columns

If timestamps tie, use STRUCT.

SQL:

SEQUENCE BY STRUCT(
  operation_timestamp,
  operation_number
)

Python:

from pyspark.sql.functions import struct
sequence_by=struct(
    "operation_timestamp",
    "operation_number"
)

from pyspark import pipelines as dp
from pyspark.sql.functions import struct, col, expr, to_timestamp

# Dummy source data - create as materialized view first
@dp.materialized_view()
def cdc_source_data():
    df = spark.createDataFrame([
        # Newer event arrives first
        (1, "UPDATE", "2026-07-24 10:00:00", 6, "Alice_New"),
        # Older event arrives later
        (1, "UPDATE", "2026-07-24 09:59:00", 5, "Alice_Old"),
        # Tie on timestamp, use operation_number as tie-breaker
        (2, "UPDATE", "2026-07-24 11:00:00", 2, "Bob_Second"),
        (2, "UPDATE", "2026-07-24 11:00:00", 1, "Bob_First"),
        # Delete event
        (3, "DELETE", "2026-07-24 12:00:00", 1, None)
    ], ["user_id", "operation", "operation_timestamp", "operation_number", "name"])

    # Cast to match the target schema: INT and TIMESTAMP types
    return df.select(
        col("user_id").cast("int"),
        col("operation"),
        to_timestamp(col("operation_timestamp")).alias("operation_timestamp"),
        col("operation_number").cast("int"),
        col("name")
    )

# Streaming view reading from the batch source
@dp.temporary_view()
def cdc_events():
    return spark.readStream.table("cdc_source_data")

# Target streaming table for CDC
dp.create_streaming_table(
    name="users_cdc",
    schema="user_id INT, name STRING, operation_timestamp TIMESTAMP, operation_number INT, operation STRING"
)

# Auto CDC flow using SEQUENCE BY STRUCT for logical ordering and tie-breaking
dp.create_auto_cdc_flow(
    target="users_cdc",
    source="cdc_events",
    keys=["user_id"],
    sequence_by=struct(
        col("operation_timestamp"),
        col("operation_number")
    ),
    apply_as_deletes=expr("operation = 'DELETE'")
)

input table= users_cdc

user_id operation operation_timestamp           operation_number name
1       UPDATE    2026-07-24T10:00:00.000+00:00 6                Alice_New
1       UPDATE    2026-07-24T09:59:00.000+00:00 5                Alice_Old
2       UPDATE    2026-07-24T11:00:00.000+00:00 2                Bob_Second
2       UPDATE    2026-07-24T11:00:00.000+00:00 1                Bob_First
3       DELETE    2026-07-24T12:00:00.000+00:00 1                null

target table= cdc_events
user_id name       operation_timestamp           operation_number operation
1       Alice_New  2026-07-24T10:00:00.000+00:00 6                UPDATE
2       Bob_Second 2026-07-24T11:00:00.000+00:00 2                UPDATE

The fields are compared in order:

operation_timestamp
operation_number as a tie-breaker

Exam tip: Out-of-order events plus a timestamp and tie-breaker usually means SEQUENCE BY STRUCT.

Slowly Changing Dimensions (SCD) in Databricks

Slowly Changing Dimensions define how a table handles attribute values that change over time. The selected SCD type determines whether changes are rejected, overwritten, or preserved as history.

SCD Type 0 — No Changes

Existing dimension records never change and are treated as static. New business keys may be appended, but existing values are not updated.

Exam Tip: Static or fixed attributes, such as an original registration date.

SCD Type 1 — Overwrite

Updates the current record
Does not preserve previous values
Represents only the latest state
Is appropriate for current inventory or current customer information

Memory rule: SCD Type 1 means latest state only. The old value is overwritten by the new value, so only the latest state is retained. No historical versions are stored.

For example, if Product A’s price changes from $10 to $50, the existing row is updated to $50 and the $10 value is no longer available in the dimension table.

Exam Tip: Latest value required; history is not needed.

SCD Type 2 — Preserve Full History

Preserves historical values
Creates multiple versions of a business key
Uses validity intervals
Supports complete or selected-column history

AUTO CDC SCD Type 2 targets use:

__START_AT
__END_AT

When explicitly declaring the target schema, these columns must use the same data type as the SEQUENCE BY field.

Memory rule: SCD Type 2 means preserve history.

A new row is inserted for every change, while the previous row is marked as historical. This preserves both the old and new versions.

SCD Type 2 commonly uses:

A current-record flag
An effective start date
An effective end date

Example:

from pyspark.sql.functions import col, lit, current_timestamp
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, TimestampType
from datetime import datetime

# Step 1: Define schema explicitly to handle NULL values
schema = StructType([
    StructField("customer_id", IntegerType(), False),
    StructField("name", StringType(), False),
    StructField("city", StringType(), False),
    StructField("status", StringType(), False),
    StructField("__START_AT", TimestampType(), False),
    StructField("__END_AT", TimestampType(), True)  # Nullable for current records
])

# Create initial customer data (Jan 1, 2024)
initial_data = [
    (1, "Alice", "Seattle", "Gold", datetime(2024, 1, 1), None),
    (2, "Bob", "Portland", "Silver", datetime(2024, 1, 1), None)
]

df_initial = spark.createDataFrame(initial_data, schema)

# Save as Delta table
df_initial.write.format("delta").mode("overwrite").saveAsTable("scd2_customers")

print("✓ Created initial customer table with SCD Type 2 structure")

# View the initial state
print("=== INITIAL STATE (Jan 1, 2024) ===")
print("Current customers:")
display(spark.table("scd2_customers").orderBy("customer_id", "__START_AT"))

customer_id name  city     status __START_AT                     __END_AT
1           Alice Seattle  Gold   2024-01-01T00:00:00.000+00:00  -
2           Bob   Portland Silver 2024-01-01T00:00:00.000+00:00  -

from delta.tables import DeltaTable

# Change 1: Alice moves from Seattle to Austin on Feb 15, 2024
change_date = datetime(2024, 2, 15)

# Close the old row by setting __END_AT
spark.sql(f"""
    UPDATE scd2_customers
    SET __END_AT = '{change_date}'
    WHERE customer_id = 1 AND __END_AT IS NULL
""")

# Insert new row with updated city
new_row = [
    (1, "Alice", "Austin", "Gold", change_date, None)
]
df_new = spark.createDataFrame(new_row, schema)
df_new.write.format("delta").mode("append").saveAsTable("scd2_customers")

print("✓ Applied Change 1: Alice moved to Austin")

# View the state after first change
print("=== AFTER CHANGE 1 (Feb 15, 2024) ===")
print("Notice: Alice now has 2 rows - one historical (Seattle) and one current (Austin)")
print()
display(spark.table("scd2_customers").orderBy("customer_id", "__START_AT"))

customer_id name   city     status  __START_AT                     __END_AT
1           Alice  Seattle  Gold    2024-01-01T00:00:00.000+00:00  2024-02-15T00:00:00.000+00:00
1           Alice  Austin   Gold    2024-02-15T00:00:00.000+00:00  -
2           Bob    Portland Silver  2024-01-01T00:00:00.000+00:00  -

# Change 2: Alice upgrades from Gold to Platinum on Mar 20, 2024
change_date2 = datetime(2024, 3, 20)

# Close the current row
spark.sql(f"""
    UPDATE scd2_customers
    SET __END_AT = '{change_date2}'
    WHERE customer_id = 1 AND __END_AT IS NULL
""")

# Insert new row with updated status
new_row2 = [
    (1, "Alice", "Austin", "Platinum", change_date2, None)
]
df_new2 = spark.createDataFrame(new_row2, schema)
df_new2.write.format("delta").mode("append").saveAsTable("scd2_customers")

print("✓ Applied Change 2: Alice upgraded to Platinum")

# View the complete history
print("=== COMPLETE HISTORY (All versions) ===")
print("Alice's journey: Seattle/Gold → Austin/Gold → Austin/Platinum")
print()
display(spark.table("scd2_customers").orderBy("customer_id", "__START_AT"))

customer_id  name   city     status    __START_AT                     __END_AT
1            Alice  Seattle  Gold      2024-01-01T00:00:00.000+00:00  2024-02-15T00:00:00.000+00:00
1            Alice  Austin   Gold      2024-02-15T00:00:00.000+00:00  2024-03-20T00:00:00.000+00:00
1            Alice  Austin   Platinum  2024-03-20T00:00:00.000+00:00  -
2            Bob    Portland Silver    2024-01-01T00:00:00.000+00:00  -

# Query current records only (most recent version of each customer)
print("=== QUERY: CURRENT RECORDS ONLY ===")
print("Filter: WHERE __END_AT IS NULL")
print()

current_records = spark.sql("""
    SELECT customer_id, name, city, status, __START_AT, __END_AT
    FROM scd2_customers
    WHERE __END_AT IS NULL
    ORDER BY customer_id
""")

display(current_records)

# Point-in-time query: What did the data look like on Feb 20, 2024?
print("=== QUERY: POINT-IN-TIME (Feb 20, 2024) ===")
print("This is AFTER Alice moved to Austin but BEFORE she upgraded to Platinum")
print()

point_in_time = spark.sql("""
    SELECT customer_id, name, city, status, __START_AT, __END_AT
    FROM scd2_customers
    WHERE __START_AT <= '2024-02-20'
      AND (__END_AT > '2024-02-20' OR __END_AT IS NULL)
    ORDER BY customer_id
""")

print("Expected: Alice in Austin with Gold status")
display(point_in_time)

# Another point-in-time query: What did the data look like on Jan 15, 2024?
print("=== QUERY: POINT-IN-TIME (Jan 15, 2024) ===")
print("This is BEFORE any changes were made")
print()

early_point = spark.sql("""
    SELECT customer_id, name, city, status, __START_AT, __END_AT
    FROM scd2_customers
    WHERE __START_AT <= '2024-01-15'
      AND (__END_AT > '2024-01-15' OR __END_AT IS NULL)
    ORDER BY customer_id
""")

print("Expected: Alice in Seattle with Gold status")
display(early_point)

In Databricks AUTO CDC, SCD Type 2 automatically uses __START_AT and __END_AT. A record with **__END_AT IS NULL represents the current version**, so a separate current flag can be derived rather than physically stored.

SCD Type 2 vs. Delta Time Travel

Delta Time Travel should not be treated as a replacement for SCD Type 2. Time Travel provides access to earlier table versions only while those versions remain within the configured retention period. VACUUM can remove the historical files.

SCD Type 2 stores business history directly in the table and is therefore the appropriate choice for long-term historical reporting.

Exam Tips:

Type 0: No updates to existing records.
Type 1: Overwrite the old value; no history.
Type 2: Insert a new version and preserve full history.
Latest state only: Choose Type 1.
Historical reporting: Choose Type 2.
Current Type 2 row: __END_AT IS NULL.
Delta Time Travel: Not a long-term SCD Type 2 replacement.
AUTO CDC: Supports SCD Type 1 and Type 2, not Type 0 as a STORED AS option.

Internal AUTO CDC Objects

Consider this flow:

CREATE OR REFRESH STREAMING TABLE users_target;

CREATE FLOW cdc_flow AS
AUTO CDC INTO users_target
FROM stream(users_raw)
KEYS (user_id)
APPLY AS DELETE
WHEN operation = 'DELETE'
SEQUENCE BY updated_timestamp
COLUMNS *;

The metastore may show:

A view named users_target
An internal table named __apply_changes_storage_users_target

These objects are used for internal CDC processing.

The backing table stores information such as:

Sequence values
Version metadata
Tombstone markers
Information required to reconcile late events
State required to handle out-of-order events

The users_target view presents the clean, queryable state.

It filters internal implementation records, including tombstones for deleted rows.

The internal table is not:

A temporary write cache
A query index
An audit-only table
The recommended user-facing dataset

The backing table and view support internal CDC processing with SEQUENCE BY, tombstones, and version information required to handle out-of-order data.

The older phrase apply_changes may still appear in the internal table name even when AUTO CDC syntax is used.

IGNORE NULL UPDATES

By default, an incoming null value overwrites the existing target value with null.

When **IGNORE NULL UPDATES is used, an incoming null means: Keep the existing target value unchanged.**

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr, to_timestamp

# Dummy source data - create as materialized view first
@dp.materialized_view()
def user_updates_source():
    df = spark.createDataFrame([
        (1, "UPDATE", "2026-07-25 10:00:00", "Alice_New", "Seattle"),
        (1, "UPDATE", "2026-07-25 11:00:00", None, "Austin"),  # name is NULL, city changes
        (2, "UPDATE", "2026-07-25 12:00:00", "Bob_New", None), # city is NULL, name changes
        (3, "DELETE", "2026-07-25 13:00:00", None, None)       # delete event
    ], ["user_id", "operation", "update_timestamp", "name", "city"])

    # Cast to match the target schema: INT and TIMESTAMP types
    return df.select(
        col("user_id").cast("int"),
        col("operation"),
        to_timestamp(col("update_timestamp")).alias("update_timestamp"),
        col("name"),
        col("city")
    )

# Streaming view reading from the batch source
@dp.temporary_view()
def user_updates():
    return spark.readStream.table("user_updates_source")

# Target streaming table
dp.create_streaming_table(
    name="users_ignore_null_demo",
    schema="user_id INT, name STRING, city STRING, update_timestamp TIMESTAMP, operation STRING"
)

# Auto CDC flow with ignore_null_updates
dp.create_auto_cdc_flow(
    target="users_ignore_null_demo",
    source="user_updates",
    keys=["user_id"],
    sequence_by=col("update_timestamp"),
    apply_as_deletes=expr("operation = 'DELETE'"),
    ignore_null_updates=True
)

user_id operation update_timestamp               name       city
1       UPDATE    2026-07-25T10:00:00.000+00:00  Alice_New  Seattle
1       UPDATE    2026-07-25T11:00:00.000+00:00  null       Austin
2       UPDATE    2026-07-25T12:00:00.000+00:00  Bob_New    null
3       DELETE    2026-07-25T13:00:00.000+00:00  null       null

user_id  name       city     update_timestamp               operation
1        Alice_New  Austin   2026-07-25T11:00:00.000+00:00  UPDATE
2        Bob_New    null     2026-07-25T12:00:00.000+00:00  UPDATE

Exam trap:

IGNORE NULL UPDATES cannot by itself distinguish between:

A column omitted because it did not change
A column intentionally changed to null

Additional update-column metadata may be needed when explicit null assignments must be supported.

AUTO CDC FROM SNAPSHOT

Use AUTO CDC when the source provides a CDC feed. Use AUTO CDC FROM SNAPSHOT when the source provides only complete snapshots.

AUTO CDC FROM SNAPSHOT:

Compares consecutive snapshots
Determines inserted rows
Determines updated rows
Determines deleted rows
Supports SCD Type 1 and Type 2
Is supported only through the Python pipeline interface
Can use snapshots from Delta tables, cloud files, or JDBC
Requires snapshots to be processed in a defined order

Conceptual example:

from pyspark import pipelines as dp
from pyspark.sql.functions import col, to_date

# Example: Auto CDC from snapshot comparing daily customer snapshots
# Creating sample snapshot data to demonstrate the concept

@dp.materialized_view()
def customer_snapshots():
    # Sample snapshot showing customer state at a single point in time
    # Each customer_id appears only once (latest snapshot: Mar 1, 2024)
    df = spark.createDataFrame([
        (1, "Alice", "Austin", "Platinum", "2024-03-01"),
        (2, "Bob", "Portland", "Silver", "2024-03-01"),
        (3, "Carol", "Denver", "Gold", "2024-03-01")
    ], ["customer_id", "name", "city", "status", "snapshot_date"])

    # Cast to match the target schema: INT and DATE types
    return df.select(
        col("customer_id").cast("int"),
        col("name"),
        col("city"),
        col("status"),
        to_date(col("snapshot_date")).alias("snapshot_date")
    )

# Create the target table for CDC results
# Note: SCD Type 2 adds __START_AT and __END_AT columns automatically
dp.create_streaming_table(
    name="customer_cdc_snapshot",
    schema="customer_id INT, name STRING, city STRING, status STRING, snapshot_date DATE, __START_AT TIMESTAMP, __END_AT TIMESTAMP"
)

# Apply Auto CDC from snapshot
dp.create_auto_cdc_from_snapshot_flow(
    target="customer_cdc_snapshot",
    source="customer_snapshots",
    keys=["customer_id"],  # Primary key(s)
    stored_as_scd_type=2  # Use 1 for SCD Type 1, 2 for SCD Type 2
)

Exam Trap: AUTO CDC processes a real CDC feed and supports SQL or Python. AUTO CDC FROM SNAPSHOT compares snapshots and is Python only.

High-Frequency Exam Traps

Trap 1: CDC and CDF are interchangeable

They are related but different. CDC is the general process. CDF is a table feature that generates row-level changes.

Trap 2: Exactly-once means duplicates disappear

False.

Exactly-once protects processing and sink commits. It does not remove duplicate business events from the source.

Trap 3: MERGE automatically chooses the newest source row

False. Multiple modifying source rows can create an ambiguous MERGE. Deduplicate before merging.

Trap 4: RANK always returns one record

False. Ties can produce multiple rank-1 rows.

Use ROW_NUMBER with a deterministic tie-breaker when uniqueness is required.

Trap 5: CDF returns the entire table after every change

False. CDF normally returns only changed rows. The initial streaming snapshot is the important exception.

Trap 6: One UPDATE creates one CDF row

Usually false. An update normally produces:

update_preimage
update_postimage

Trap 7: CDF is permanent audit history

False. CDF follows table and log retention. Archive it if permanent history is required.

Associate Exam Scenario Questions

Question 1

A pipeline copies only new, updated, and deleted customer rows from a source database to a Delta target. Which pattern is being used?

A. Full refresh B. Change Data Capture C. Time travel D. Compaction

Answer: B — Change Data Capture

CDC identifies inserts, updates, and deletes without copying the complete source every time.

Question 2

A Delta table has CDF enabled. One record is updated and one new record is inserted in version 10. How are the changes normally represented?

A. One update and one insert B. One update_preimage, one update_postimage, and one insert C. The entire table as inserts D. Only the updated values

Answer: B An update normally generates both preimage and postimage records.

Question 3

Which SQL query returns CDF changes starting with version 20?

A. SELECT * FROM VERSION AS OF 20 B. **SELECT * FROM table_changes('orders', 20)* C. `SELECT FROM STREAM orders VERSION 20 D.DESCRIBE HISTORY orders LIMIT 20`

Answer: B table_changes accepts a table name and starting version or timestamp.

Question 4

Which table property enables legacy Delta CDF?

A. delta.enableCDC = true B. delta.readChangeFeed = true C. **delta.enableChangeDataFeed = true** D. pipelines.enableChangeDataFeed = true

Answer: C

**readChangeFeed is a reader option. **delta.enableChangeDataFeed enables legacy CDF on the table.

Question 5

A team expects to query every CDF event forever, even after aggressive VACUUM operations. Which statement is correct?

A. CDF is an unlimited audit archive. B. VACUUM deletes only checkpoints. C. CDF follows retention, and old changes can become unavailable. D. CDF is stored outside the table.

Answer: C Permanent change history should be copied to a separate audit table.

Question 6

Which columns identify the CDF operation, table version, and commit time?

A. operation, version, timestamp B. _change_type, _commit_version, _commit_timestamp C. change, delta_version, event_time D. _operation, _version, _time

Answer: B Memorize the three underscore-prefixed CDF metadata columns.

Professional Exam Scenario Questions

Question 1

A micro-batch contains two UPDATE events for customer_id 100. Both match the same target row in a MERGE. What should the engineer do?

A. Let Delta select one randomly. B. Sort the DataFrame and merge both rows. C. Deduplicate to one deterministic latest row before MERGE. D. Replace MERGE with INSERT OVERWRITE.

Answer: C Multiple modifying source rows matching the same target row are ambiguous.

Question 2

Two events for the same key have the same timestamp. Filtering RANK = 1 still returns both. What is the safest correction?

A. Remove the partition key. B. Use ROW_NUMBER with a deterministic tie-breaker. C. Use COUNT instead of RANK. D. Reverse the MERGE aliases.

Answer: B

RANK preserves ties. ROW_NUMBER selects one winner when the ordering is deterministic.

Question 3

A stream can receive duplicate order events in later micro-batches. The target must remain unique by order_id. Which design is best?

A. Run dropDuplicates only inside the current foreachBatch call. B. Insert every record and run VACUUM. C. Use watermark-based streaming deduplication and an insert-only MERGE. D. Repartition the stream by order_id.

Answer: C This handles late duplicates within the watermark horizon and prevents reinserting keys already stored in the target.

Question 4

Which current Python API replaces apply_changes?

A. merge_changes B. **create_auto_cdc_flow** C. readChangeFeed D. create_streaming_merge

Answer: B create_auto_cdc_flow replaces apply_changes.

Question 5

An AUTO CDC target exposes users_target and __apply_changes_storage_users_target. What is the backing table used for?

A. Temporary caching B. Query indexing C. Internal sequence, version, and tombstone state D. A duplicate user-facing table

Answer: C It maintains state required for out-of-order CDC processing.

Question 6

CDC events contain event_timestamp and event_number. Timestamps can tie. Which clause establishes the correct event order?

A. KEYS (event_timestamp, event_number) B. SEQUENCE BY STRUCT(event_timestamp, event_number) C. COLUMNS event_timestamp ORDER BY event_number D. APPLY AS UPDATE WHEN event_number > 0

Answer: B STRUCT uses the second field as a tie-breaker.

Question 7

A target must retain every historical customer-address version and its validity interval. Which storage mode should be used?

A. SCD Type 0 B. SCD Type 1 C. SCD Type 2 D. Append-only mode

Answer: C SCD Type 2 preserves historical versions using __START_AT and __END_AT.

Question 8

A database produces a complete customer snapshot every night but does not provide a transaction-log change feed. Which API should be used to infer inserts, updates, and deletes?

A. AUTO CDC FROM SNAPSHOT B. table_changes C. OPTIMIZE D. skipChangeCommits

Answer: A AUTO CDC FROM SNAPSHOT compares ordered snapshots and is Python only.

Question 9

Source update events use null for columns that did not change. Existing target values must be preserved. Which option should be used?

A. IGNORE NULL UPDATES B. APPLY AS TRUNCATE C. WHEN NOT MATCHED BY SOURCE DELETE D. VACUUM RETAIN 0 HOURS

Answer: A Without this option, incoming nulls overwrite target values by default.

Question 10

An AUTO CDC flow must treat a special source event as a complete target-table truncate. Which statement is correct?

A. APPLY AS TRUNCATE is supported only for SCD Type 1. B. APPLY AS TRUNCATE is supported only for SCD Type 2. C. Every DELETE becomes a TRUNCATE. D. CDF must be disabled.

Answer: A SCD Type 2 does not support APPLY AS TRUNCATE.

Question 11

A downstream pipeline must process changes emitted by an AUTO CDC target. What is the most robust approach?

A. Read the target’s Change Data Feed. B. Run SELECT DISTINCT during every full refresh. C. Query the internal backing table. D. Ignore update and delete commits.

Answer: A CDF explicitly exposes inserts, updates, and deletes for downstream processing.

Question 12

A data engineer is configuring the following CDC data processing using AUTO CDC APIs in Lakeflow Declarative Pipelines:

CREATE FLOW cdc_flow AS AUTO CDC INTO silver_transactions
FROM stream(bronze_transactions)
KEYS (transaction_id)
_________________________
COLUMNS *

The engineer wants to define the processing order of source records and handle late-arriving data using a composite key of multiple columns, ordering by transaction_timestamp first, and in case of ties, by version_number.

Answer: SEQUENCE BY STRUCT (transaction_timestamp, version_number)

SEQUENCE BY is used to define the processing order for CDC streams, and using STRUCT allows specifying a composite key of multiple columns. This ensures that records are ordered first by transaction_timestamp and, in case of ties, by version_number, which also allows handling late-arriving data correctly.

Question 13

Given the following query on the Delta table ‘customers’ on which Change Data Feed is enabled:

spark.read
.option("readChangeFeed", "true")
.option("startingVersion", 0)
.table ("customers")
.filter (col("_change_type").isin(["update_postimage"]))
.write
.mode("append")
.table("customers_updates")

Which statement describes the result of query each time it is executed ?

Answer: The entire history of updated records will be appended to the target table at each execution, which leads to duplicate entries.

Reading table’s changes, captured by CDF, using spark.read means that you are reading them as a static source. So, each time you run the query, all table’s changes (starting from the specified startingVersion) will be read.

The query in the question then appends the data to the target table at each execution since it’s using the ‘append’ writing mode.

Question 14

Given the following query on the Delta table customers on which Change Data Feed is enabled:

spark.read
.option("readChangeFeed", "true")
.option("startingVersion", 0)
.table("customers")
.filter(col("_change_type").isin(["update_postimage"]))
.write
.mode("overwrite")
.table("customers_updates")

Which statement describes the results of this query each time it is executed?

Answer: The entire history of updated records will overwrite the target table at each execution.

The query in the question then writes the data in mode “overwrite” to the target table, which completely overwrites the table at each execution.

Question 15

A data engineer is noticing that a large UC-managed Delta table (≈750GB) has become slow when applying intensive CDC feeds.

Which of the following actions should the data engineer take to improve the performance?

Answer: Enable deletion vectors on the table and apply liquid clustering using the primary keys.

Since Change Data Capture (CDC) involves processing updates and deletions, to improve the performance of a large Delta table experiencing slow CDC feeds, the data engineer should enable deletion vectors on the table and apply liquid clustering using the primary keys.

Enabling deletion vectors allows Delta to efficiently track and manage rows that are deleted or updated without requiring full rewrites of the underlying files, which significantly reduces the overhead for CDC operations. Applying liquid clustering on the CDC merging keys organizes the data physically based on these keys, ensuring that related records are colocated and minimizing the amount of data scanned during updates and deletions. Together, these optimizations help maintain high ingestion and query performance, reduce latency for CDC workloads, and make the table more manageable at scale.

Question 16

A data engineer is using a foreachBatch logic to upsert data in a target Delta table. The function to be called at each new microbatch processing is displayed below with a blank:

def upsert_data(microBatchDF, batch_id):
microBatchDF.createOrReplaceTempView("sales_microbatch")

sql_query = """
MERGE INTO sales_silver a
USING sales_microbatch b
ON a.item_id=b.item_id AND a.item_timestamp=b.item_timestamp
WHEN NOT MATCHED THEN INSERT *
"""
________________

Which option correctly fills in the blank to execute the sql query in the function on a cluster with recent Databricks Runtime above 10.5 ?

Answer: microBatchDF.sparkSession.sql(sql_query)

Usually, we use spark.sql() function to run SQL queries. However, in this particular case, the spark session cannot be accessed from within the microbatch process. Instead, we can access the local spark session from the microbatch dataframe.

Question 17

A data engineer has the following logic to handle duplicates in Spark Structured Streaming:

(spark.readStream
.table("bronze")
.filter("topic = 'orders'")
.select(F.from_json(F.col("value").cast("string"), schema).alias("v"))
.select("v.*")
.withWatermark("order_timestamp", "30 seconds")
.dropDuplicates(["order_id", "order_timestamp"]))

However, they notice that this logic is not sufficient to prevent duplicates for events that arrive later than the watermark threshold.

Which of the following code snippets can the data engineer include in a foreachBatch function to completely handle streaming duplicates?

Answer:

MERGE INTO orders_silver a
USING microbatch b
ON a.order_id=b.order_id AND a.order_timestamp=b.order_timestamp
WHEN NOT MATCHED THEN INSERT *

In Spark Structured Streaming, dropDuplicates with a watermark only removes duplicates that arrive within the defined event-time threshold, for example, within 30 seconds of order_timestamp. However, any records arriving later than that threshold are considered “too late” and are not deduplicated by Spark’s in-memory state. To ensure complete deduplication (including very late-arriving data), the foreachBatch sink can use an idempotent write pattern with Delta Lake’s MERGE operation.

The MERGE INTO statement compares each micro-batch of incoming data (microbatch) with the target Delta table (orders_silver) based on unique keys (order_id and order_timestamp). It only inserts rows that do not already exist in the target table, preventing duplicates even across micro-batches or late arrivals. This combination of in-stream deduplication (for near-real-time performance) and MERGE-based deduplication (for completeness and correctness) provides an end-to-end reliable way to handle duplicates in streaming pipelines.

spark.readStream.table("orders_raw")
.dropDuplicates(["order_id", "order_timestamp"])
.writeStream
.option("checkpointLocation", "dbfs:/checkpoints")
.table("orders_unique")

Question 18

A data engineering team at a supply chain company uses Lakeflow Declarative Pipelines to manage inventory data. The team maintains a streaming table, inventory_updates, with Change Data Feed (CDF) enabled. The table captures real-time changes to product inventory levels, with columns: product_id, quantity, and update_timestamp.

The team needs to incrementally propagate all inventory changes from the inventory_updates table to downstream layers.

Which implementation approach correctly satisfies this requirement?

Answer:

(spark.readStream
.option("readChangeFeed", "true")
.table("inventory_updates")
)

Because the inventory_updates table contains updates and deletes, it breaks the append-only requirement of standard streaming tables. This means that we cannot directly stream from the base table or just skip change commits. Instead, since Change Data Feed (CDF) is enabled, the correct approach is to use spark.readStream to consume all inventory changes — including inserts, updates, and deletes — from the CDF output and apply them downstream using AUTO CDC APIs (previously known as APPLY CHANGES APIs).

Remember, to read the change data feed from a table, you need to set the option readChangeFeed to true when configuring a stream read from the table.

https://docs.databricks.com/aws/en/delta/delta-change-data-feed

https://docs.databricks.com/aws/en/ldp/cdc

Last-Minute Cheat Sheet

Definitions

CDC captures inserts, updates, and deletes from a source.
CDF exposes row-level table changes.
MERGE manually applies changes to a Delta target.
AUTO CDC applies ordered CDC events in Lakeflow pipelines.
AUTO CDC FROM SNAPSHOT compares complete snapshots and is Python only.

CDF columns

_change_type
_commit_version
_commit_timestamp

CDF change types

insert
update_preimage
update_postimage
delete

Enable legacy CDF

TBLPROPERTIES (
  delta.enableChangeDataFeed = true
)

Query CDF

SELECT *
FROM table_changes(
  'table_name',
  start_version,
  end_version
);

CDF retention

CDF follows retention.
VACUUM can remove change history.
CDF is not a permanent audit archive.

MERGE

Multiple modifying source rows for one target can fail.
Deduplicate the source before MERGE.
Use ROW_NUMBER with a tie-breaker when ties are possible.
Insert-only MERGE checks against the target, not within an undeduplicated source batch.

Streaming deduplication

Exactly-once does not equal business-event deduplication.
Use a watermark with dropDuplicatesWithinWatermark.
Use insert-only MERGE to avoid reinserting existing target keys.

AUTO CDC

Replaces APPLY CHANGES.
The target must be a streaming table.
KEYS identifies the record.
SEQUENCE BY orders events.
APPLY AS DELETE handles deletes.
SCD Type 1 stores the latest state.
SCD Type 2 preserves history.
Use SEQUENCE BY STRUCT for multiple ordering fields.
The backing table stores tombstones, sequence state, and versions.

Pipeline objects

Streaming table: persistent incremental target
AUTO CDC target: streaming table that applies updates and deletes correctly
Materialized view: stored derived result maintained from changing inputs
Temporary view: nonpersistent intermediate transformation

Final Memory Rules

CDC is the process; CDF is the feed.
One update normally creates two CDF records: preimage and postimage.
Resolve multiple source updates for one key before MERGE.
RANK can tie; ROW_NUMBER selects one deterministic winner.
SEQUENCE BY handles late and out-of-order events.
AUTO CDC is the new name; APPLY CHANGES is the old name.
SCD Type 1 stores the latest state; SCD Type 2 preserves history.
CDF follows retention and is not permanent history.

Official Databricks References

메타데이터

post_id: b756ea3cfafd
slug: a-complete-associate-and-professional-exam-guide-on-cdc-and-cdf-b756ea3cfafd
url: https://medium.com/@rohit299pradhan/a-complete-associate-and-professional-exam-guide-on-cdc-and-cdf-b756ea3cfafd
canonical_url: https://medium.com/@rohit299pradhan/a-complete-associate-and-professional-exam-guide-on-cdc-and-cdf-b756ea3cfafd
author_url: https://medium.com/@rohit299pradhan
status: ok
fetched_at: 2026-08-11 04:08:35