Change Data Capture (CDC): Sync Data Without Full Reloads

Quickstart

If you’re setting up CDC (or debugging one that’s flaky), start here. These steps are the highest-impact decisions that determine whether your pipeline behaves like a reliable stream… or like a nightly lottery.

Overview

Full reloads are the default when teams start: “We’ll rebuild the table every night.” That works until it doesn’t: tables get bigger, SLAs get tighter, and downstream systems want near-real-time updates. CDC replaces reloads with a continuous flow of committed changes.

Core concepts

CDC can look complex because people mix together three different problems: capturing changes, transporting changes, and applying changes. If you separate them, the design becomes much clearer.

Ordering, duplicates, and “exactly once”

CDC delivery is typically at-least-once (messages can be retried). “Exactly once” is achievable only with extra constraints: deterministic ordering, transactional writes to the sink, and a checkpoint that advances atomically with the sink update.

Deletes: tombstones vs soft deletes

CDC pipelines often “look correct” until the first real deletion. Choose an approach up front:

• Tombstone: emit an event indicating the key is deleted and remove it downstream
• Soft delete: keep the row but set deleted=true (useful for auditing or GDPR workflows)
• Type 2 history: keep versions with valid_from/valid_to (best for analytics history)

Step-by-step

This guide is tool-agnostic: the steps are the same whether you use a managed service, a connector framework, or a custom pipeline. The goal is a pipeline you can restart, replay, and reason about.

Step 1 — Define the contract (the “truth” you want downstream)

Step 2 — Backfill a snapshot (get the baseline into the sink)

CDC streams changes, not entire state. You still need an initial baseline. Keep it boring: bulk-load the current state into the sink, record the cutover point, and don’t mix “random snapshots” with “live changes.”

Step 3 — Start streaming changes (log-based example)

In log-based CDC, a connector reads committed changes from the database log and emits structured change events to a transport (often a message broker). Even if you use a managed tool, you should understand the moving parts: connection, checkpoint storage, and how schema changes are represented.

{
  "name": "cdc-postgres-orders",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres.internal",
    "database.port": "5432",
    "database.user": "cdc_user",
    "database.password": "${CDC_PASSWORD}",
    "database.dbname": "appdb",
    "database.server.name": "appdb",
    "slot.name": "debezium_orders_slot",
    "publication.autocreate.mode": "filtered",
    "schema.include.list": "public",
    "table.include.list": "public.orders,public.order_items",

    "plugin.name": "pgoutput",
    "snapshot.mode": "initial",
    "tombstones.on.delete": "true",

    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter.schemas.enable": "false",

    "topic.prefix": "cdc",
    "heartbeat.interval.ms": "10000",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "cdc.dlq"
  }
}

Step 4 — Apply changes to the sink (upsert + “only accept newer”)

The sink is where CDC correctness is won. You want a deterministic rule: “For a given primary key, apply the newest change and ignore duplicates/older arrivals.” A common pattern is to store the last-applied position per row (or per key) and only update when the incoming position is newer.

-- Example sink table (current state) stores the last applied position.
-- The position can be a log sequence number (LSN), an offset, or a strictly increasing event_id.
--
-- target.orders_current(order_id PK, status, total_cents, updated_at, source_pos)

INSERT INTO target.orders_current (order_id, status, total_cents, updated_at, source_pos)
SELECT
  s.order_id,
  s.status,
  s.total_cents,
  s.updated_at,
  s.source_pos
FROM staging.orders_cdc AS s
WHERE s.op IN ('c', 'u')
ON CONFLICT (order_id) DO UPDATE
SET
  status      = EXCLUDED.status,
  total_cents = EXCLUDED.total_cents,
  updated_at  = EXCLUDED.updated_at,
  source_pos  = EXCLUDED.source_pos
WHERE EXCLUDED.source_pos > target.orders_current.source_pos;

-- Handle deletes explicitly (either hard-delete or soft-delete).
DELETE FROM target.orders_current t
USING staging.orders_cdc s
WHERE s.op = 'd'
  AND s.order_id = t.order_id
  AND s.source_pos > t.source_pos;

Step 5 — Make consumers restart-safe (checkpointing + idempotency)

You should be able to stop your pipeline mid-stream, restart it, and end up in the same state. That requires: (1) storing progress (checkpoint/offset), and (2) making reprocessing safe (idempotent writes).

import json
import psycopg2
from psycopg2.extras import execute_values

# Strategy:
# - Each CDC event has an (stream_id, position) tuple.
# - We write data changes and the checkpoint in ONE transaction.
# - If we crash and retry, the unique constraint prevents double-advancing incorrectly.

DDL = """
CREATE TABLE IF NOT EXISTS cdc_checkpoints (
  stream_id TEXT PRIMARY KEY,
  last_pos  TEXT NOT NULL,
  updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Optional: track processed positions to dedupe (useful if events can repeat with same pos).
CREATE TABLE IF NOT EXISTS cdc_processed (
  stream_id TEXT NOT NULL,
  pos       TEXT NOT NULL,
  PRIMARY KEY (stream_id, pos)
);
"""

def apply_batch(conn, stream_id: str, events: list[dict]) -> None:
  # events: [{"pos": "...", "op": "c|u|d", "key": {"order_id": 1}, "after": {...}}]
  with conn:
    with conn.cursor() as cur:
      # 1) Deduplicate positions (safe if batch repeats)
      execute_values(
        cur,
        "INSERT INTO cdc_processed(stream_id, pos) VALUES %s ON CONFLICT DO NOTHING",
        [(stream_id, e["pos"]) for e in events],
      )

      # 2) Apply changes (idempotent upsert/delete).
      # In real pipelines, route per table/topic and validate schema.
      for e in events:
        order_id = e["key"]["order_id"]
        pos = e["pos"]
        if e["op"] in ("c", "u"):
          after = e["after"]
          cur.execute(
            """
            INSERT INTO target.orders_current(order_id, status, total_cents, updated_at, source_pos)
            VALUES (%s, %s, %s, %s, %s)
            ON CONFLICT (order_id) DO UPDATE
            SET status = EXCLUDED.status,
                total_cents = EXCLUDED.total_cents,
                updated_at = EXCLUDED.updated_at,
                source_pos = EXCLUDED.source_pos
            WHERE EXCLUDED.source_pos > target.orders_current.source_pos
            """,
            (order_id, after["status"], after["total_cents"], after["updated_at"], pos),
          )
        elif e["op"] == "d":
          cur.execute(
            """
            DELETE FROM target.orders_current
            WHERE order_id = %s AND source_pos <= %s
            """,
            (order_id, pos),
          )

      # 3) Advance checkpoint to the max position in this batch.
      last_pos = max(e["pos"] for e in events)
      cur.execute(
        """
        INSERT INTO cdc_checkpoints(stream_id, last_pos)
        VALUES (%s, %s)
        ON CONFLICT (stream_id) DO UPDATE
        SET last_pos = EXCLUDED.last_pos, updated_at = now()
        """,
        (stream_id, last_pos),
      )

if __name__ == "__main__":
  conn = psycopg2.connect("dbname=warehouse user=etl password=secret host=127.0.0.1")
  with conn:
    with conn.cursor() as cur:
      cur.execute(DDL)

  # Demo batch (pretend it came from a broker/connector)
  demo = [
    {"pos": "00000016/000000A0", "op": "u", "key": {"order_id": 42},
     "after": {"status": "paid", "total_cents": 1299, "updated_at": "2026-01-09T12:00:00Z"}}
  ]
  apply_batch(conn, "orders_stream", demo)

Step 6 — Plan for schema changes (because they will happen)

CDC pipelines don’t fail because of “bad SQL.” They fail because schema changes are treated as surprises. Decide your rules:

• Additive changes (safe): new nullable columns, new optional fields (usually safe to ignore at first)
• Breaking changes: rename columns, change types, split/merge tables (requires versioning and coordinated rollout)
• Event contract: do you ship full-row state (“after”) or a minimal patch? Full-row tends to be easier downstream.

Step 7 — Operate CDC like a production system

Common mistakes

CDC bugs are rarely loud. They’re usually subtle: one missed update here, one duplicate there, and eventually your numbers don’t match. Here are the most common causes—plus the straightforward fixes.

FAQ

Is CDC the same as incremental ETL?

No. Incremental ETL usually means “process new/changed rows on a schedule,” often using timestamps. CDC is specifically about capturing committed changes as a stream (often from the database log) with checkpoints, replay, and delete semantics.

Do I need Kafka to do CDC?

No. Kafka is a common transport because it’s durable and replayable, but CDC can work with managed streams, queues, cloud migration services, or even direct apply-to-sink connectors. The core requirements are durability, ordering rules, and checkpointing—not a specific product.

Can CDC be “exactly once”?

Sometimes, but don’t assume it. Most CDC systems deliver at-least-once. You get correctness by making sinks idempotent and by advancing checkpoints only when sink writes commit. “Exactly once” typically requires transactional guarantees across transport + sink or a carefully designed apply/checkpoint transaction.

How do I handle deletes in CDC pipelines?

Explicitly. Either propagate hard deletes via tombstone/delete events, or implement soft deletes with a deleted flag. If you ignore deletes, your downstream systems will slowly drift, even if updates look correct.

What’s the most common CDC cutover sequence?

Backfill → stream → validate → switch. Load a snapshot into the sink, start CDC from a consistent boundary, apply changes until caught up, validate counts and recent updates, then point consumers/readers at the new sink.

What should I store as my checkpoint?

A monotonic position from the capture system. For log-based CDC, use the log position (LSN/offset/SCN). For timestamp polling, store (updated_at, id) and re-read a safety window. Avoid using wall-clock time alone as a checkpoint.

Cheatsheet

A scan-fast CDC checklist you can keep next to your pipeline dashboard.

Wrap-up

CDC is one of those “invisible” systems that, when done right, feels like magic: data is fresh everywhere and nobody talks about it. When done wrong, it slowly erodes trust because numbers don’t match and nobody knows why.

The practical recipe is simple: choose the right capture method, treat changes as an ordered stream with checkpoints, make your sink idempotent, plan for deletes and schema changes, and use a safe cutover sequence (backfill → stream → validate → switch).

Related posts

• SQL That Scales: Joins, Windows, and Real-World Patterns
• Airflow vs Prefect: Picking an Orchestrator
• NoSQL When It’s Actually the Right Choice
• Schema Migrations Without Downtime
• Explain Plans Demystified: Find the Slow Part Fast
• Star Schema vs Snowflake: Modeling for Analytics