How do I do CDC or SCD Type 2 in SDP?

Summary

Open-source Spark 4.2.0 added Auto CDC to Spark Declarative Pipelines (SDP). It applies a change feed to a streaming table with create_auto_cdc_flow, but only as SCD Type 1, which keeps the current row for each key. For SCD Type 2 history, write a Delta Lake MERGE yourself in open-source Spark, or use Lakeflow Pipelines on Databricks, where AUTO CDC supports SCD Type 2.

How do I do CDC or SCD Type 2 in SDP?

CDC means applying a feed of row-level changes (inserts, updates and deletes) to a target table. The slowly changing dimension (SCD) type says how the target stores them. The Apache Spark 4.2.0 release notes list "Auto CDC (declarative SCD Type 1) in Spark Declarative Pipelines," and the create_auto_cdc_flow docs say of stored_as_scd_type: "Only 1 (or "1") is supported." In Spark 4.2.0, CDC is declarative but Type 1 only, and Type 2 needs another route.

Goal	Open-source Apache Spark	Lakeflow Pipelines on Databricks
Current row per key (SCD Type 1)	dp.create_auto_cdc_flow with stored_as_scd_type=1 (Spark 4.2.0)	AUTO CDC with stored_as_scd_type = 1, or AUTO CDC INTO in SQL
Full history per key (SCD Type 2)	A Delta Lake MERGE pattern that you write	AUTO CDC with stored_as_scd_type = "2", or STORED AS SCD TYPE 2 in SQL
History columns	You define them, for example current, effectiveDate and endDate	__START_AT and __END_AT

How do I apply a change feed with Auto CDC in open-source SDP?

Create the target streaming table first; the docstring says it "must have already been created using the create_streaming_table function." Spark 4.2.0 also requires a target "backed by a connector that supports MERGE," and refuses to start the flow otherwise.

from pyspark import pipelines as dp

@dp.table
def customer_changes():
    return spark.readStream.table("raw_customer_changes")

dp.create_streaming_table("customers")

dp.create_auto_cdc_flow(
    target="customers",
    source="customer_changes",
    keys=["customer_id"],
    sequence_by="change_version",
    apply_as_deletes="operation = 'DELETE'",
    except_column_list=["operation", "change_version"],
    stored_as_scd_type=1,
)

keys identify a row, sequence_by orders the change events, and apply_as_deletes is a boolean expression that marks delete events.

How do I keep SCD Type 2 history in open-source Spark?

The Delta Lake docs describe SCD Type 2 as an operation that "maintains history of all changes made to each key in a dimensional table," and express it with MERGE. Stage the updates twice. Rows with a NULL mergeKey cannot match the merge condition, so they fall through to the insert clause as the new current version. Rows keyed by customerId match the current row and expire it. Delta Lake's compatibility table lists Spark 4.0.x and 4.1.x for its 4.3.x line (as of October 2026), so run this job on a Spark version your Delta Lake release supports.

customersTable = ...  # DeltaTable: customerId, address, current, effectiveDate, endDate
updates = updatesDF.alias("updates")  # updatesDF: customerId, address, effectiveDate

newAddressesToInsert = updates \
  .join(customersTable.toDF().alias("customers"), "customerid") \
  .where("customers.current = true AND updates.address <> customers.address")

stagedUpdates = (
  newAddressesToInsert.selectExpr("NULL as mergeKey", "updates.*")
  .union(updates.selectExpr("updates.customerId as mergeKey", "*")))

customersTable.alias("customers").merge(
  stagedUpdates.alias("staged_updates"), "customers.customerId = mergeKey") \
.whenMatchedUpdate(
  condition = "customers.current = true AND customers.address <> staged_updates.address",
  set = {"current": "false", "endDate": "staged_updates.effectiveDate"}) \
.whenNotMatchedInsert(values = {
  "customerid": "staged_updates.customerId", "address": "staged_updates.address",
  "current": "true", "effectiveDate": "staged_updates.effectiveDate", "endDate": "null"}) \
.execute()

To run a merge on a stream, call it from foreachBatch, as the Delta Lake docs do for upserts.

What errors does Auto CDC raise in Spark 4.2.0?

Mistake	What Spark 4.2.0 reports	Fix
stored_as_scd_type=2	PySparkTypeError (NOT_EXPECTED_TYPE) from the Python API; the engine's AUTOCDC_SCD2_NOT_SUPPORTED says AutoCDC flows "do not currently support SCD Type 2 transformations"	Use the MERGE pattern above, or Lakeflow Pipelines
Target without MERGE support	AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE	Use a table format whose connector supports MERGE
col("sourceTable.keyId") as a key	AUTOCDC_NON_COLUMN_IDENTIFIER: arguments "must reference unqualified column names"	Use keys=["keyId"]
Changing keys between runs	AUTOCDC_INVALID_STATE (KEY_SCHEMA_DRIFT): "To change keys, perform a full refresh of the target table."	Run spark-pipelines run --full-refresh customers
NULL in a key or sequence column	AUTOCDC_MICROBATCH_VALIDATION (NULL_KEY, NULL_SEQUENCE)	Filter or repair those rows upstream

How Databricks approaches this

Lakeflow Pipelines covers both types with one API. The AUTO CDC APIs "automate the complexity of computing slowly changing dimensions (SCD) Type 1 and Type 2 from either a CDC feed or database snapshots." Type 2 targets get __START_AT and __END_AT columns, and "Records with __END_AT = null are currently active." The pipeline must use serverless Lakeflow Pipelines or the Pro or Advanced edition. Databricks recommends AUTO CDC "to ensure that out of order records are handled correctly when processing CDC feeds." It replaces the older APPLY CHANGES APIs, which are still available.

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

# "users" is a streaming dataset of change events defined elsewhere in the pipeline
dp.create_streaming_table("users_history")
dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2")

FAQs

What is the difference between CDC and SCD?

CDC is the feed of changes from a source. The SCD type is how the target keeps them. Per the Databricks docs, "Use SCD Type 1 to update records directly. History is not retained for updated records." Type 2 keeps a row for each version of a record.

Can I read changes straight from a Delta table?

Yes. Set the table property with ALTER TABLE myDeltaTable SET TBLPROPERTIES (delta.enableChangeDataFeed = true), then read with the readChangeFeed option. Rows carry _change_type (insert, update_preimage, update_postimage or delete), _commit_version and _commit_timestamp. Only changes made after you enable the feed are recorded.