Implementing Change Data Capture with Debezium and Airbyte: Streaming Relational Database Changes into Your Cloud Warehouse
Introduction
Imagine you're a data engineer at a mid-sized e-commerce company. Your production PostgreSQL database is the operational heartbeat of the business — orders are placed, inventory is updated, customer records are modified, and payments are processed, all day long. Your analytics warehouse in Snowflake tells the story of that activity, but with a frustrating asterisk: it's showing you yesterday's story. Your nightly batch job runs at 2 AM, which means a product manager asking "how many orders came in during this morning's flash sale?" at 11 AM gets a helpless shrug.
The traditional fix — tightening up your batch window, running syncs every 15 minutes — creates its own set of problems. You're hammering your production database with SELECT queries, missing deletes entirely, and still not solving the real question: what actually changed, and when? Change Data Capture (CDC) solves this elegantly by reading the database's own write-ahead log (WAL) rather than querying the data itself. Instead of asking "what does the data look like right now," CDC asks "what mutations happened, in what order?"
In this lesson, you'll build a production-grade CDC pipeline using Debezium (for log-based CDC) and Airbyte (for managed connector infrastructure), landing real-time change events into Snowflake where your transformation layer can act on them. By the end, you'll understand not just the how but the why behind every design decision, and you'll know exactly what can go wrong and how to catch it before it bites you.
What you'll learn:
- How database write-ahead logs work and why they're the right foundation for CDC
- How to configure PostgreSQL for logical replication and connect Debezium to it
- How Airbyte's CDC mode differs from standard incremental sync, and when each is appropriate
- How to model incoming CDC events in your warehouse for both real-time and historical analysis
- How to handle schema evolution, deletes, and late-arriving events without breaking your pipeline
Prerequisites
You should be comfortable with:
- Running SQL against a relational database (PostgreSQL preferred)
- Basic Kafka concepts (topics, consumers, producers) — you don't need to be an expert, but knowing what a consumer group is will help
- Cloud data warehouse fundamentals — Snowflake or BigQuery both work for this lesson
- Data ingestion concepts with tools like Airbyte
You'll need:
- A PostgreSQL instance (version 10+) where you have superuser access — a local Docker install works fine for the exercises
- An Airbyte instance (Cloud or self-hosted OSS)
- A Snowflake or BigQuery destination (free tier is fine)
How CDC Actually Works: The Write-Ahead Log
Before touching any tooling, you need to understand what's happening under the hood. This isn't optional background reading — it directly informs every configuration decision you'll make.
Every production-grade relational database maintains a log of changes before applying them to the actual data files. PostgreSQL calls this the Write-Ahead Log (WAL); MySQL calls it the binlog; SQL Server calls it the transaction log. The primary purpose of this log is crash recovery — if the database process dies mid-write, it can replay the log to restore a consistent state.
The critical insight is that this log is a complete, ordered record of every INSERT, UPDATE, and DELETE that's ever happened. CDC tools don't poll your tables — they tail this log, similar to how you'd tail -f a log file on Linux. This has enormous practical consequences:
- Zero query load on your tables. CDC doesn't run SELECT statements. It reads log files.
- You capture deletes. A batch sync based on a
updated_at cursor will never see deleted rows. The WAL records them.
- Event ordering is preserved. You know the exact sequence of changes, which matters for reconstructing state at any point in time.
- Minimal latency. Events can reach your warehouse in seconds, not hours.
PostgreSQL's WAL supports a feature called logical replication, which decodes the raw binary WAL entries into a human-readable change stream. You need to enable this explicitly — PostgreSQL ships with it disabled by default because it has storage and performance implications.
Note: Logical replication slots in PostgreSQL retain WAL segments until a consumer acknowledges them. If your CDC consumer goes down for an extended period, WAL files accumulate on disk. Left unchecked, this will fill your disk and crash PostgreSQL. This is one of the most common production incidents with Debezium. We'll cover how to monitor this.
The Shape of a CDC Event
Whether you're using Debezium, Airbyte's native CDC, or any other tool, every CDC event carries the same essential payload:
{
"before": {
"id": 10042,
"customer_id": 5581,
"status": "pending",
"total_amount": 149.99,
"updated_at": "2024-03-15T09:22:11Z"
},
"after": {
"id": 10042,
"customer_id": 5581,
"status": "confirmed",
"total_amount": 149.99,
"updated_at": "2024-03-15T09:23:44Z"
},
"op": "u",
"ts_ms": 1710494624000,
"source": {
"db": "ecommerce_prod",
"table": "orders",
"lsn": 87234921
}
}
The op field tells you the operation type: c for create (INSERT), u for update (UPDATE), d for delete (DELETE), r for read (initial snapshot). The before and after fields give you the complete row state on either side of the change. The lsn (Log Sequence Number) is your position in the WAL — this is how the consumer tracks where it left off.
For an INSERT, before is null. For a DELETE, after is null. This matters when you're writing your transformation logic downstream.
Setting Up PostgreSQL for Logical Replication
Let's configure a PostgreSQL instance from scratch. If you're working with an existing production database, the changes are the same — you'll just need DBA approval for the configuration changes and a brief restart.
Step 1: Enable Logical Replication
Edit your postgresql.conf (location varies by OS; SHOW config_file; will tell you):
# postgresql.conf
wal_level = logical
max_replication_slots = 4
max_wal_senders = 4
wal_level = logical is the key setting. The default is replica, which only supports physical (binary) replication. logical enables the decoded change stream that Debezium needs. After editing, restart PostgreSQL:
# For systemd-based Linux
sudo systemctl restart postgresql
# For Docker
docker restart your_postgres_container
Step 2: Create a Dedicated Replication User
Never use your application user for CDC. Create a dedicated user with exactly the permissions it needs:
-- Create the CDC user
CREATE USER debezium_cdc WITH
REPLICATION
LOGIN
PASSWORD 'strong_password_here';
-- Grant SELECT on the tables you want to capture
GRANT SELECT ON TABLE public.orders TO debezium_cdc;
GRANT SELECT ON TABLE public.order_items TO debezium_cdc;
GRANT SELECT ON TABLE public.customers TO debezium_cdc;
GRANT SELECT ON TABLE public.products TO debezium_cdc;
-- Grant USAGE on the schema
GRANT USAGE ON SCHEMA public TO debezium_cdc;
Tip: If you want to capture all current and future tables in a schema, use ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO debezium_cdc;. This saves you from the inevitable moment six months from now when someone adds a new table and wonders why it's not flowing.
By default, PostgreSQL only records the primary key in the before image of an UPDATE event. This means for an UPDATE, you'll see the full after row but only the PK in before, making it impossible to reconstruct what changed. For CDC to be genuinely useful, set replica identity to FULL on your captured tables:
ALTER TABLE public.orders REPLICA IDENTITY FULL;
ALTER TABLE public.order_items REPLICA IDENTITY FULL;
ALTER TABLE public.customers REPLICA IDENTITY FULL;
ALTER TABLE public.products REPLICA IDENTITY FULL;
Warning: REPLICA IDENTITY FULL increases WAL volume because it writes the complete row before-image. On high-write tables, this can be significant — measure WAL generation rates before and after in production. For very large, frequently-updated tables, consider capturing only specific columns or using a targeted replica identity index instead.
Step 4: Create a Publication
A PostgreSQL publication defines which tables are included in the logical replication stream:
CREATE PUBLICATION debezium_pub
FOR TABLE public.orders, public.order_items, public.customers, public.products;
Alternatively, publish all tables: CREATE PUBLICATION debezium_pub FOR ALL TABLES; — though this is usually too broad for production.
Deploying Debezium
Debezium is a set of Kafka Connect source connectors. It tails the database WAL and publishes structured change events to Kafka topics. Let's stand up the minimal infrastructure to make this work.
Docker Compose Stack
For this exercise, you'll run Kafka, Zookeeper, and Debezium in Docker. In production, you'd use a managed Kafka service (Confluent Cloud, MSK, Redpanda Cloud), but the concepts are identical:
# docker-compose.yml
version: "3.8"
services:
zookeeper:
image: confluentinc/cp-zookeeper:7.5.0
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ports:
- "2181:2181"
kafka:
image: confluentinc/cp-kafka:7.5.0
depends_on:
- zookeeper
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_LOG_RETENTION_HOURS: 168
ports:
- "29092:29092"
kafka-connect:
image: debezium/connect:2.4
depends_on:
- kafka
environment:
BOOTSTRAP_SERVERS: kafka:9092
GROUP_ID: debezium-connect
CONFIG_STORAGE_TOPIC: debezium_connect_configs
OFFSET_STORAGE_TOPIC: debezium_connect_offsets
STATUS_STORAGE_TOPIC: debezium_connect_status
KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
ports:
- "8083:8083"
Start the stack: docker compose up -d
Registering the Debezium Connector
Once Kafka Connect is running, register your PostgreSQL connector via the REST API:
curl -X POST http://localhost:8083/connectors \
-H "Content-Type: application/json" \
-d '{
"name": "ecommerce-postgres-connector",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"plugin.name": "pgoutput",
"database.hostname": "host.docker.internal",
"database.port": "5432",
"database.user": "debezium_cdc",
"database.password": "strong_password_here",
"database.dbname": "ecommerce_prod",
"database.server.name": "ecommerce",
"table.include.list": "public.orders,public.order_items,public.customers,public.products",
"publication.name": "debezium_pub",
"slot.name": "debezium_slot",
"snapshot.mode": "initial",
"heartbeat.interval.ms": "10000",
"decimal.handling.mode": "double",
"time.precision.mode": "connect",
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.add.fields": "op,ts_ms,source.db,source.table,source.lsn",
"transforms.unwrap.delete.handling.mode": "rewrite",
"transforms.unwrap.drop.tombstones": "false"
}
}'
Let's unpack the key configuration decisions:
plugin.name: pgoutput — This is the built-in PostgreSQL logical decoding plugin (available since PG 10). The alternative is wal2json, which requires a separate installation. Prefer pgoutput unless you have a specific reason.
snapshot.mode: initial — When the connector starts for the first time, it takes a full consistent snapshot of all configured tables, then switches to streaming. initial_only takes the snapshot and stops. never skips the snapshot (use this if you're confident your warehouse already has the data). always snapshots on every restart — almost never what you want in production.
transforms.unwrap.type: ExtractNewRecordState — By default, Debezium emits the full envelope (before/after/op/source). The ExtractNewRecordState Single Message Transform (SMT) flattens this to just the after state, with the metadata fields added back as top-level fields. This is much easier to work with in most downstream systems. The add.fields configuration brings along the operation type and timestamp so you don't lose that context.
heartbeat.interval.ms: 10000 — This sends a heartbeat message to the Kafka topic every 10 seconds even when there are no changes. Without this, idle databases can lag in WAL acknowledgment, causing slot retention issues.
Verifying the Stream
Check that topics are being created and populated:
# List topics - you should see one per table
docker exec -it kafka-container kafka-topics --bootstrap-server localhost:9092 --list
# Consume from the orders topic
docker exec -it kafka-container kafka-console-consumer \
--bootstrap-server localhost:9092 \
--topic ecommerce.public.orders \
--from-beginning \
--max-messages 5
Make a change in PostgreSQL and watch it appear:
-- In your PostgreSQL client
UPDATE orders SET status = 'shipped' WHERE id = 10042;
Within a second or two, you should see a change event appear in the Kafka topic.
Configuring Airbyte for CDC
Here's where the architecture branches. You have two options for getting Kafka CDC events into your warehouse:
- Debezium → Kafka → Airbyte Kafka Source → Warehouse: Use Airbyte's Kafka source connector to consume from the topics Debezium is writing.
- Airbyte's Native CDC: Skip the Kafka middleware entirely and let Airbyte handle the PostgreSQL logical replication directly, using its own built-in CDC logic.
Both are legitimate. The Kafka-based approach gives you more flexibility (other consumers can tap the same stream, you have replay capability) but requires more infrastructure. Airbyte's native CDC is simpler to operate when Airbyte is already your ingestion platform.
Key insight: If you already have a Kafka infrastructure investment, or if multiple teams need access to the change stream (not just your warehouse), the Debezium-first approach wins. If your warehouse is the sole consumer and you want to minimize operational surface area, Airbyte's native CDC is the pragmatic choice. For this lesson, we'll walk through both paths so you understand the tradeoffs. The concepts from the incremental sync configuration article apply directly here.
Path A: Airbyte Native CDC (Postgres Source)
In the Airbyte UI, create a new Postgres source connection:
- Navigate to Sources → New Source → search for Postgres
- Fill in your connection details (host, port, database, user, password)
- Under Replication Method, select Logical Replication (CDC)
- Set Replication Slot to
debezium_slot (or create a new one — Airbyte will create its own slot)
- Set Publication to
debezium_pub
- Set Initial Load Timeout (hours) to something appropriate for your table sizes
When you configure the destination (Snowflake or BigQuery), navigate to Streams configuration. For each table, you'll choose a sync mode:
- Full Refresh | Overwrite: Replaces entire table on every sync. Never use this for CDC.
- Full Refresh | Append: Appends entire table on every sync. Useful for small reference tables only.
- Incremental | Append: Only new/changed records land in the warehouse. The source table's cursor field determines what's "new."
- Incremental | Append + Dedup: CDC events land and Airbyte deduplicates using the primary key. This is the mode you want for most CDC use cases.
Select Incremental | Append + Dedup for your transactional tables. With this mode, Airbyte:
- Lands every CDC event (including the operation metadata) in a raw staging table
- Runs a deduplication query that keeps only the latest version of each row by primary key
The result in your destination looks like a normal, current-state table — but you have the raw events available too if you need them.
Path B: Airbyte Kafka Source (Debezium → Kafka → Airbyte)
For this path, Debezium is already running and writing to Kafka topics. Configure an Airbyte Kafka source:
- Navigate to Sources → New Source → search for Kafka
- Set Bootstrap Servers to your Kafka broker address
- Set Subscription Type to Subscribe (not Assign — you want Kafka to manage partition assignment)
- Set Topic Pattern to
ecommerce\.public\.* (regex matching all your tables)
- Set Consumer Group ID to
airbyte-warehouse-consumer
- Configure deserialization format — JSON if you followed our Debezium config above
The Kafka source will consume messages and forward them to your warehouse destination. Configure the sync connection with Incremental | Append mode, and handle deduplication in your transformation layer using dbt.
Landing and Modeling CDC Events in the Warehouse
Raw CDC events aren't analytics-ready. They're a stream of mutations, not a coherent table of current state. The transformation layer — where dbt earns its keep — is responsible for turning that stream into something useful.
The Raw Events Table
After Airbyte syncs, your Snowflake raw layer will look something like this (Airbyte appends _airbyte_* metadata columns):
-- What lands in the raw layer: ecommerce_raw.orders_raw
SELECT
id,
customer_id,
status,
total_amount,
updated_at,
_airbyte_op, -- 'c', 'u', or 'd'
_airbyte_extracted_at, -- when Airbyte captured it
_airbyte_emitted_at, -- when it was written to the warehouse
_airbyte_data -- full JSON payload (if enabled)
FROM ecommerce_raw.orders_raw
ORDER BY _airbyte_extracted_at
LIMIT 5;
id | customer_id | status | total_amount | _airbyte_op | _airbyte_extracted_at
10042 | 5581 | pending | 149.99 | c | 2024-03-15 09:22:12
10042 | 5581 | confirmed | 149.99 | u | 2024-03-15 09:23:45
10042 | 5581 | shipped | 149.99 | u | 2024-03-15 14:17:03
10043 | 7293 | pending | 89.50 | c | 2024-03-15 09:31:19
Order 10042 has three events. Your analytics model should show it once, as shipped. But your historical analysis model might want all three events to understand the order's journey through the funnel.
The Current-State Model
This is what replaces your traditional orders table in the silver/marts layer:
-- models/staging/stg_orders.sql
WITH source AS (
SELECT
id,
customer_id,
status,
total_amount,
updated_at,
_airbyte_op AS cdc_operation,
_airbyte_extracted_at AS cdc_extracted_at,
ROW_NUMBER() OVER (
PARTITION BY id
ORDER BY _airbyte_extracted_at DESC, _airbyte_emitted_at DESC
) AS row_num
FROM {{ source('ecommerce_raw', 'orders_raw') }}
WHERE _airbyte_op != 'd' -- exclude deleted rows from current state
),
deduped AS (
SELECT * FROM source WHERE row_num = 1
)
SELECT
id AS order_id,
customer_id,
status AS order_status,
total_amount,
updated_at AS last_updated_at,
cdc_operation,
cdc_extracted_at
FROM deduped
Warning: Don't use updated_at as your deduplication tiebreaker unless you're certain your application always sets it correctly. Application bugs often result in updated_at not being updated on every write. Use _airbyte_extracted_at or the WAL LSN position — these are set by the CDC system, not your application code, and are therefore more reliable.
The Event History Model
For funnel analysis, fraud detection, or debugging, you often want every event preserved:
-- models/staging/stg_orders_history.sql
SELECT
id AS order_id,
customer_id,
status AS order_status,
total_amount,
updated_at AS event_timestamp,
_airbyte_op AS cdc_operation,
_airbyte_extracted_at AS cdc_extracted_at,
LAG(status) OVER (
PARTITION BY id
ORDER BY _airbyte_extracted_at
) AS previous_status,
-- Time between status transitions
DATEDIFF(
'minute',
LAG(_airbyte_extracted_at) OVER (
PARTITION BY id
ORDER BY _airbyte_extracted_at
),
_airbyte_extracted_at
) AS minutes_in_previous_status
FROM {{ source('ecommerce_raw', 'orders_raw') }}
This model powers questions like "how long does an order typically spend in pending state before confirmation?" — questions that a current-state table simply can't answer.
This approach maps naturally to a medallion architecture where raw CDC events form your bronze layer, deduplicated current state is silver, and your aggregated metrics live in gold.
Handling Deletes
The delete case requires a decision: do you hard-delete from your warehouse, or soft-delete? Hard deletes are almost always wrong for analytics — they destroy historical information and break time-based analyses. The standard pattern is to flag deleted records:
-- models/staging/stg_orders.sql (with delete handling)
WITH source AS (
SELECT
id,
customer_id,
status,
total_amount,
updated_at,
_airbyte_op AS cdc_operation,
_airbyte_extracted_at AS cdc_extracted_at,
ROW_NUMBER() OVER (
PARTITION BY id
ORDER BY _airbyte_extracted_at DESC
) AS row_num
FROM {{ source('ecommerce_raw', 'orders_raw') }}
),
deduped AS (
SELECT * FROM source WHERE row_num = 1
)
SELECT
id AS order_id,
customer_id,
status AS order_status,
total_amount,
updated_at AS last_updated_at,
cdc_operation = 'd' AS is_deleted,
CASE WHEN cdc_operation = 'd'
THEN cdc_extracted_at
END AS deleted_at,
cdc_extracted_at
FROM deduped
Now your downstream models can filter WHERE NOT is_deleted for operational reports while audit queries can still find deleted records. This pattern aligns with slowly changing dimension handling — you're making the same tradeoff between current state and historical accuracy.
Incremental dbt Models on Top of CDC Data
The current-state model above does a full table scan on every run. For large tables, that's expensive. The right answer is an incremental dbt model that only processes new CDC events:
-- models/staging/stg_orders.sql (incremental version)
{{
config(
materialized = 'incremental',
unique_key = 'order_id',
incremental_strategy = 'merge',
cluster_by = ['order_status', 'cdc_extracted_at']
)
}}
WITH source AS (
SELECT
id,
customer_id,
status,
total_amount,
updated_at,
_airbyte_op AS cdc_operation,
_airbyte_extracted_at AS cdc_extracted_at,
ROW_NUMBER() OVER (
PARTITION BY id
ORDER BY _airbyte_extracted_at DESC
) AS row_num
FROM {{ source('ecommerce_raw', 'orders_raw') }}
{% if is_incremental() %}
-- Only process events newer than what we've already processed
WHERE _airbyte_extracted_at > (
SELECT MAX(cdc_extracted_at) FROM {{ this }}
)
{% endif %}
),
deduped AS (
SELECT * FROM source WHERE row_num = 1
)
SELECT
id AS order_id,
customer_id,
status AS order_status,
total_amount,
updated_at AS last_updated_at,
cdc_operation = 'd' AS is_deleted,
cdc_extracted_at
FROM deduped
The merge incremental strategy will UPSERT into your destination table — inserting new order IDs and updating existing ones when their CDC events arrive. This is exactly the semantics you want for a current-state view.
Tip: When using is_incremental() with a MAX(cdc_extracted_at) watermark, add a small buffer: WHERE _airbyte_extracted_at > (SELECT MAX(cdc_extracted_at) - INTERVAL '10 minutes' FROM {{ this }}). Late-arriving events (where Kafka delivery was delayed) can otherwise slip through and never be processed. The 10-minute buffer means you reprocess a small window of events on each run, but the merge strategy handles the resulting duplicates correctly.
Monitoring WAL Retention and Replication Lag
This is where CDC pipelines go quietly wrong in production. Two metrics require active monitoring:
Replication Slot Lag
-- Run this on PostgreSQL to monitor slot lag
SELECT
slot_name,
plugin,
slot_type,
active,
pg_size_pretty(
pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)
) AS retained_wal_size,
pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) AS lag_bytes
FROM pg_replication_slots
ORDER BY lag_bytes DESC;
If retained_wal_size is growing and the slot shows active = false, your consumer has gone down and WAL is accumulating. Set an alert threshold — 5 GB of retained WAL is a reasonable warning level; 20 GB should be a pager alert.
In Kafka, monitor consumer group lag:
docker exec -it kafka-container kafka-consumer-groups \
--bootstrap-server localhost:9092 \
--group airbyte-warehouse-consumer \
--describe
The LAG column shows how many messages the consumer is behind. A growing lag means your Airbyte sync can't keep up with Debezium's output rate — you may need to increase Airbyte's sync frequency or reduce Kafka topic retention to avoid unbounded growth.
Debezium Connector Status
# Check connector status via Kafka Connect REST API
curl http://localhost:8083/connectors/ecommerce-postgres-connector/status | jq .
# Expected output:
{
"name": "ecommerce-postgres-connector",
"connector": {
"state": "RUNNING",
"worker_id": "kafka-connect:8083"
},
"tasks": [
{
"id": 0,
"state": "RUNNING",
"worker_id": "kafka-connect:8083"
}
]
}
A connector in FAILED state with a stopped task is your most common failure mode. Add health checks to whatever orchestration system you're using — Airflow, Kubernetes liveness probes, or a simple cron job calling this endpoint and alerting on non-RUNNING states.
This kind of systematic monitoring becomes even more important as your stack grows. Embedding CDC monitoring into your broader data freshness SLA framework means you're alerted proactively rather than reacting to a downstream analyst noticing stale numbers.
Hands-On Exercise: Building a Real-Time Order Funnel Dashboard
Let's tie everything together with a concrete project. You're going to build a pipeline that tracks orders through their lifecycle in near-real-time, producing a funnel analysis that refreshes every few minutes.
Step 1: Seed your PostgreSQL database
-- Create and populate the orders table
CREATE TABLE public.orders (
id SERIAL PRIMARY KEY,
customer_id INTEGER NOT NULL,
status VARCHAR(50) NOT NULL DEFAULT 'pending',
total_amount DECIMAL(10,2) NOT NULL,
item_count INTEGER NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
ALTER TABLE public.orders REPLICA IDENTITY FULL;
-- Insert 100 sample orders
INSERT INTO public.orders (customer_id, status, total_amount, item_count)
SELECT
(RANDOM() * 10000)::INTEGER,
'pending',
ROUND((RANDOM() * 500 + 10)::NUMERIC, 2),
(RANDOM() * 10 + 1)::INTEGER
FROM generate_series(1, 100);
Follow the Airbyte native CDC setup described earlier. Set your sync frequency to Every 5 minutes. Run the initial sync to capture the snapshot.
Step 3: Simulate order lifecycle changes
-- Confirm 40 orders
UPDATE public.orders
SET status = 'confirmed', updated_at = NOW()
WHERE id IN (SELECT id FROM public.orders ORDER BY RANDOM() LIMIT 40);
-- Ship 20 of the confirmed ones
UPDATE public.orders
SET status = 'shipped', updated_at = NOW()
WHERE status = 'confirmed'
AND id IN (SELECT id FROM public.orders WHERE status = 'confirmed' ORDER BY RANDOM() LIMIT 20);
-- Cancel 5 orders
UPDATE public.orders
SET status = 'cancelled', updated_at = NOW()
WHERE id IN (SELECT id FROM public.orders WHERE status = 'pending' ORDER BY RANDOM() LIMIT 5);
-- Delete 2 fraudulent orders entirely
DELETE FROM public.orders WHERE id IN (
SELECT id FROM public.orders ORDER BY RANDOM() LIMIT 2
);
Step 4: Build the funnel model in dbt
-- models/marts/order_funnel_metrics.sql
WITH current_orders AS (
SELECT * FROM {{ ref('stg_orders') }}
WHERE NOT is_deleted
),
funnel AS (
SELECT
order_status,
COUNT(*) AS order_count,
SUM(total_amount) AS total_value,
ROUND(AVG(total_amount), 2) AS avg_order_value,
-- Conversion rate from previous stage
COUNT(*) * 100.0 / NULLIF(
SUM(COUNT(*)) OVER (), 0
) AS pct_of_all_orders
FROM current_orders
GROUP BY order_status
)
SELECT
order_status,
order_count,
total_value,
avg_order_value,
ROUND(pct_of_all_orders, 1) AS pct_of_all_orders,
NOW() AS report_generated_at
FROM funnel
ORDER BY
CASE order_status
WHEN 'pending' THEN 1
WHEN 'confirmed' THEN 2
WHEN 'shipped' THEN 3
WHEN 'cancelled' THEN 4
ELSE 5
END
Run dbt run --select stg_orders order_funnel_metrics and then trigger another Airbyte sync. Within minutes, your funnel metrics should reflect the state changes you made in PostgreSQL, including the is_deleted = true records being excluded from the count.
Key insight: Compare the order count in current_orders with what you see by querying orders_raw directly. You should see more rows in orders_raw (all events) than in the deduped model (one row per order ID, latest state). The difference tells you how many change events have been processed. This ratio is a useful sanity check — a deduped count higher than your raw event count indicates a bug in your deduplication logic.
Common Mistakes & Troubleshooting
Mistake 1: Forgetting REPLICA IDENTITY FULL and getting empty before images
Symptom: Your UPDATE events in Kafka only contain the primary key in the before field — everything else is null.
Fix: ALTER TABLE your_table REPLICA IDENTITY FULL; Note that this requires a brief table lock (ACCESS EXCLUSIVE) on PostgreSQL. Plan for this on busy tables.
Mistake 2: Using the wrong snapshot mode on connector restart
Symptom: After restarting Debezium, your warehouse has duplicates — records that existed before the restart appear twice.
Cause: snapshot.mode: always re-snapshots the entire table on every connector start.
Fix: Use snapshot.mode: initial (default) — it only snapshots when no offset exists. Use snapshot.mode: never if you're confident in your existing state and just want to resume streaming.
Mistake 3: Ignoring WAL accumulation
Symptom: PostgreSQL disk fills up overnight and the database crashes.
Cause: Your replication slot (debezium_slot) stopped being acknowledged — perhaps because Debezium was down — and PostgreSQL couldn't purge old WAL files.
Fix: Monitor pg_replication_slots lag daily. Consider setting max_slot_wal_keep_size in postgresql.conf to limit how much WAL a slot can retain (at the cost of potentially invalidating the slot if the consumer falls too far behind):
max_slot_wal_keep_size = 10GB
Mistake 4: Schema changes breaking the connector
Symptom: The Debezium connector fails with a schema evolution error after a column was added to a source table.
Cause: By default, Debezium detects schema changes and may fail if the schema registry isn't configured to handle evolution.
Fix: Set schema.history.internal.kafka.topic to a durable Kafka topic for schema history. When using ExtractNewRecordState without a schema registry (plain JSON), added columns will flow through automatically. Dropped columns will produce null values in the event. Alert on DDL changes to your published tables — they're a coordination point between your application team and your data team.
Mistake 5: Deduplicating on updated_at instead of a system-generated timestamp
Symptom: You have duplicate current-state records, or some updates are silently dropped.
Cause: Application bugs where updated_at isn't always incremented on write, combined with using updated_at as your dedup key.
Fix: Always use _airbyte_extracted_at or the WAL LSN position as your authoritative ordering field. These are set by the CDC infrastructure, not your application.
Summary & Next Steps
You've built a production-grade CDC pipeline from PostgreSQL through Debezium (or Airbyte native CDC) into Snowflake, with a dbt transformation layer that delivers both current state and event history. The core ideas to carry forward:
- Log-based CDC is fundamentally different from query-based incremental sync. It reads the database's own change log rather than querying data, which eliminates query load, captures deletes, and preserves ordering.
- WAL slot management is your biggest operational risk. Monitor it actively. Set retention limits. Alert on inactive slots.
- Raw events and current state serve different analytical needs. Keep both. The event history model is often more valuable than the current state model for advanced analytics.
- Use system-assigned timestamps for deduplication, not application-assigned ones.
From here, natural next steps include:
CDC is one of those techniques that, once you implement it correctly, fundamentally changes what your data platform can do. The gap between "what happened yesterday" and "what's happening now" closes, and an entirely new class of analytics becomes possible.
