What Is Change Data Capture (CDC) — and How to Orchestrate It

TL;DR — Change data capture (CDC) is a data integration pattern that detects and streams row-level changes — inserts, updates, and deletes — from a database as they happen, instead of re-copying entire tables on a schedule. CDC powers real-time analytics, data warehouse synchronization, and event-driven architectures while keeping the load on source systems minimal.

Most data teams meet CDC the day a nightly full-table copy stops scaling: the sync window grows, the warehouse bill climbs, and the business asks why yesterday’s orders aren’t in this morning’s dashboard. CDC fixes all three at once — you move only what changed, seconds after it changed.

This guide covers how CDC works, and then does what most CDC articles don’t: it shows a complete, runnable pipeline that captures changes from PostgreSQL and merges them into Snowflake.

How CDC works

There are three ways to detect changes in a database, and they are not equal:

Log-based CDC won this comparison years ago, and Debezium is its de-facto open-source implementation: it reads the transaction log of Postgres, MySQL, SQL Server, MongoDB, Oracle, or DB2 and emits one structured change event per row.

Why CDC needs orchestration

Capturing changes is the easy half. A production CDC pipeline also needs to:

• transform raw change events (handle deletes, deduplicate, enrich);
• load them into a destination with correct upsert semantics (MERGE, not blind inserts);
• recover from failures without losing or double-processing events;
• alert someone when the pipeline breaks — not when the business notices.

That layer is orchestration. Here is what it looks like as one declarative flow.

Orchestrate CDC with Kestra: Postgres → Snowflake

This flow uses the Debezium PostgreSQL trigger to poll for change events (one execution per batch), cleans them in Python, stages the result in Snowflake, and merges it into the target table. Deleted rows are flagged by Debezium (deleted field) and handled explicitly.

id: postgres-cdc-to-snowflake
namespace: company.data

triggers:
  - id: postgres_cdc
    type: io.kestra.plugin.debezium.postgres.Trigger
    hostname: "{{ secret('PG_HOST') }}"
    port: "5432"
    database: production
    username: "{{ secret('PG_USERNAME') }}"
    password: "{{ secret('PG_PASSWORD') }}"
    includedTables:
      - public.orders
    format: INLINE
    splitTable: "OFF"
    snapshotMode: INITIAL
    interval: PT1M

tasks:
  - id: to_json
    type: io.kestra.plugin.serdes.json.IonToJson
    from: "{{ trigger.uris.data }}"

  - id: transform
    type: io.kestra.plugin.scripts.python.Script
    inputFiles:
      changes.jsonl: "{{ outputs.to_json.uri }}"
    outputFiles:
      - orders_clean.csv
    script: |
      import csv, json

      with open("changes.jsonl") as src, open("orders_clean.csv", "w", newline="") as out:
          writer = None
          for line in src:
              row = json.loads(line)
              row["is_deleted"] = bool(row.pop("deleted", False))
              if writer is None:
                  writer = csv.DictWriter(out, fieldnames=row.keys())
                  writer.writeheader()
              writer.writerow(row)

  - id: stage
    type: io.kestra.plugin.jdbc.snowflake.Upload
    url: "jdbc:snowflake://{{ secret('SNOWFLAKE_ACCOUNT') }}.snowflakecomputing.com"
    username: "{{ secret('SNOWFLAKE_USERNAME') }}"
    password: "{{ secret('SNOWFLAKE_PASSWORD') }}"
    from: "{{ outputs.transform.outputFiles['orders_clean.csv'] }}"
    fileName: orders_changes.csv
    prefix: cdc
    stageName: "@analytics.public.%orders_staging"

  - id: merge
    type: io.kestra.plugin.jdbc.snowflake.Queries
    url: "jdbc:snowflake://{{ secret('SNOWFLAKE_ACCOUNT') }}.snowflakecomputing.com"
    username: "{{ secret('SNOWFLAKE_USERNAME') }}"
    password: "{{ secret('SNOWFLAKE_PASSWORD') }}"
    sql: |
      COPY INTO analytics.public.orders_staging;
      MERGE INTO analytics.public.orders t
      USING analytics.public.orders_staging s ON t.order_id = s.order_id
      WHEN MATCHED AND s.is_deleted THEN DELETE
      WHEN MATCHED THEN UPDATE SET t.status = s.status, t.updated_at = s.updated_at
      WHEN NOT MATCHED AND NOT s.is_deleted THEN
        INSERT (order_id, status, updated_at) VALUES (s.order_id, s.status, s.updated_at);

errors:
  - id: alert
    type: io.kestra.plugin.notifications.slack.SlackExecution
    url: "{{ secret('SLACK_WEBHOOK') }}"
    channel: "#data-alerts"

A few things worth noticing — because they are exactly the parts a raw Debezium deployment leaves you to build yourself:

• State is managed for you. The trigger stores Debezium offsets in Kestra’s KV Store, so a restart resumes from the last committed position instead of re-snapshotting the table.
• Deletes are first-class. deleted: ADD_FIELD (the default) flags removed rows, and the MERGE handles them — the failure mode of query-based CDC simply doesn’t exist here.
• Failure has a path. The errors block alerts Slack on any failed execution; add retries with exponential backoff per task if the destination is flaky.

Batch or real-time? Two triggers, one decision

The flow above uses Trigger — it polls every minute (interval: PT1M) and creates one execution per batch of changes. Its sibling RealtimeTrigger creates one execution per row, within milliseconds.

Rule of thumb: warehouse sync and analytics want the batch trigger (fewer, larger MERGEs are cheaper in Snowflake); operational reactions — fraud checks, cache invalidation, notifications — want the real-time trigger. It’s a one-line change, which is the point of keeping CDC logic in declarative YAML rather than in connector configuration files.

Where CDC pays off

• Warehouse synchronization — replace nightly full loads with minute-level freshness (this page’s flow; see also blueprint #194: Debezium → Slack notification → Python processing).
• Event-driven architectures — turn database changes into triggers for downstream workflows without touching application code.
• Audit trails & compliance — an immutable stream of every change, with metadata, for free.
• Cache and search-index invalidation — update Redis or Elasticsearch the moment a row changes, via the same pattern with a different final task.

Kestra ships Debezium plugins for PostgreSQL, MySQL, SQL Server, MongoDB, Oracle, and DB2 — the flow above ports to any of them by swapping the trigger type.

Related concepts: Postgres logical replication · Postgres WAL · Snowflake streams · ETL vs ELT · Dead letter queue · Data orchestration

▶ Run this flow in minutes — try Kestra Cloud or docker run kestra/kestra and paste the YAML above.