How to create Delta CDC tables in Lakehouse: step by step

João Barros 07 de September de 2026 6 min read

We will implement a simple Change Data Capture (CDC) flow using Delta tables in Lakehouse to capture inserts, updates and deletes from a transactional source. This is useful to keep efficient synchronization between systems and enable historical analysis without replicating the entire database. With a well-designed CDC flow you can reduce the volume of data moved: for example, instead of reprocessing 100,000 records per hour, you only apply the 1,200 changes that occurred in that period.

Prerequisites

Account with access to Microsoft Fabric and a configured Lakehouse.
Permissions to create Delta tables and run notebooks with PySpark.
Data source (CSV or stream) with a unique identifier and an operation field (op: I/U/D) or a change log.

Additionally, it is recommended to have volume estimates: for example, knowing you expect on average 500–5,000 CDC events per hour helps define micro-batches and cluster configuration. Ensuring you have retention and cleanup policies (VACUUM) aligned with auditing needs is also important — for example, a minimum retention of 7 days for Delta.

Step 1: Prepare the target Delta table (base type)

Create a Delta table that will act as the materialized store with the current state of records. We will use a simple schema: id, valor, updated_at. This table will keep the latest state per id. For a real scenario, the table may have thousands to millions of rows; in this initial example you can start with 10,000 records.

from pyspark.sql.types import StructType, StructField, IntegerType, StringType, TimestampType

schema = StructType([
    StructField("id", IntegerType(), False),
    StructField("valor", StringType(), True),
    StructField("updated_at", TimestampType(), True)
])

# Path do Lakehouse onde guardar a tabela
target_path = "/lakehouse/yourcatalog/yourdb/delta_target"

# Criar DataFrame vazio e gravar como Delta
empty_df = spark.createDataFrame([], schema)
empty_df.write.format("delta").mode("overwrite").save(target_path)

# Criar tabela no catálogo (opcional)
spark.sql(f"CREATE TABLE IF NOT EXISTS yourdb.delta_target USING DELTA LOCATION '{target_path}'")

Tip: if you expect many records, consider partitioning by a suitable column (e.g.: ano_mês) or using Z-order to optimize reads by id.

Step 2: Initial ingestion (full load)

Load the initial state from the source into the Delta table. This ensures the target table starts with a consistent snapshot. In production, a full load can take minutes to hours (e.g.: 1 million records), so do this off hours or with throttling.

# Exemplo a partir de um CSV de origem
source_df = spark.read.format("csv").option("header", True).schema(schema).load("/data/initial_snapshot.csv")

# Gravar no target (substitui conteúdo inicial)
source_df.write.format("delta").mode("overwrite").save(target_path)

After the full load validate: for example, confirm there are 10,000 records with spark.read.format("delta").load(target_path).count() or run sample checksums to ensure integrity.

Step 3: Structure the CDC feed

Assume a feed with columns: id, valor, op, event_time. op = 'I'|'U'|'D'. Prepare a DataFrame with the changes to apply. In a 1-hour window you may have, for example, 300 updates, 150 inserts and 50 deletes — these numbers help define batch sizes and required memory.

cdc_schema = StructType([
    StructField("id", IntegerType(), False),
    StructField("valor", StringType(), True),
    StructField("op", StringType(), False),
    StructField("event_time", TimestampType(), True)
])

cdc_df = spark.read.format("csv").option("header", True).schema(cdc_schema).load("/data/cdc_feed.csv")

Validate the ordering and accuracy of timestamps. If the feed arrives unordered, use a deduplication strategy based on event_time.

Step 4: Apply CDC with MERGE (upsert/delete)

Use MERGE INTO in Delta to apply inserts, updates and deletes transactionally. Filter by event order if necessary (e.g.: last event per id). MERGE guarantees atomicity — meaning either all changes in the batch succeed, or none are applied.

from delta.tables import DeltaTable

delta_table = DeltaTable.forPath(spark, target_path)

# Optional: se existirem múltiplos eventos por id, pega o último por event_time
from pyspark.sql import Window
from pyspark.sql.functions import row_number

w = Window.partitionBy("id").orderBy(cdc_df["event_time"].desc())
cdc_latest = cdc_df.withColumn("rn", row_number().over(w)).filter("rn = 1").drop("rn")

# Executar MERGE
(delta_table.alias("t")
  .merge(cdc_latest.alias("s"), "t.id = s.id")
  .whenMatchedUpdate(
      condition = "s.op != 'D'",
      set = {"valor": "s.valor", "updated_at": "s.event_time"}
  )
  .whenMatchedDelete(condition = "s.op = 'D'")
  .whenNotMatchedInsert(values = {"id": "s.id", "valor": "s.valor", "updated_at": "s.event_time"})
  .execute())

Practical example: in a batch with 500 deduplicated events, you expect to apply ~350 updates, 120 inserts and 30 deletes. After the MERGE, validate the row counts and some random records to confirm behavior.

Step 5: Automate and handle common errors

Put this flow in a scheduled notebook or pipeline. Handle common errors: duplicate data in the feed, missing op field, timestamp conflicts. Implement logs and metrics (num applied, num rejected) for monitoring. If you use Microsoft Fabric, schedule the notebook with triggers and capture failures for reprocessing.

# Exemplo simples de validação antes do MERGE
invalid_ops = cdc_df.filter("op NOT IN ('I','U','D')").count()
if invalid_ops > 0:
    raise ValueError(f"Encontradas {invalid_ops} operações inválidas no feed CDC")

Other practices: use checkpoints for structured streaming, limit batch size to 100k events, and set VACUUM with minimum retention (e.g.: 7 days) to avoid premature removal of history needed for replays.

Verify the result

Confirm that records are correct by querying the Delta table and comparing with the expected feed. Check inserts, updates and deletes and the updated_at field. Verification examples: count differences between expected snapshot and table (by id), check most recent timestamps and confirm there are no rows marked as deleted when they should not be.

# Consultar a tabela Delta
spark.read.format("delta").load(target_path).show()

# Comparar contagens
print("Total na tabela:", spark.read.format("delta").load(target_path).count())

Conclusion

You have implemented a basic CDC flow with Delta tables in Lakehouse, using MERGE to ensure atomicity. Next steps: integrate a stream (structured streaming) to reduce latency, store lineage metadata and implement automated tests and alerts. Practical tip: start with small CDC batches (e.g.: 1k–5k events) and always validate event order before scaling to production.