Case 2 — Postgres CDC to Warehouse (Real-Time Replication)

# Case 2 — Postgres CDC to Warehouse ## The brief You have a Postgres application database. You need its tables replicated to the warehouse with near-real-time freshness (under 5 minutes lag). UPDATEs and DELETEs must be reflected accurately. ## The setup Source: A Postgres database (Docker container) with these tables: - `users` (id, email, name, created_at, updated_at) - `orders` (id, user_id, status, amount, created_at, updated_at) - `order_items` (id, order_id, product_id, quantity, price) Pre-populate with a few hundred rows. Then run a "traffic generator" Python script that: - Inserts new orders every few seconds. - Updates random orders' status (pending → paid → shipped). - Occasionally deletes test orders. Destination: BigQuery / Snowflake / Postgres warehouse (different schema). ## The task ### Step 1: Choose your CDC tool (15 min) Options: - **Debezium** (open source, requires Kafka): production-grade, complex. - **Airbyte CDC** (open source, Kafka-free): simpler, smaller scale. - **DIY with logical replication slots**: most educational, most work. For learning, build DIY with Postgres logical replication. (Pick Airbyte if you want a faster path.) ### Step 2: Set up logical replication (30 min) Enable on Postgres: ```sql -- postgresql.conf wal_level = logical max_wal_senders = 10 max_replication_slots = 10 -- restart Postgres -- create publication CREATE PUBLICATION cdc_pub FOR ALL TABLES; -- create replication slot SELECT pg_create_logical_replication_slot('cdc_slot', 'pgoutput'); ``` ### Step 3: Build the CDC consumer (30 min) Python script that: - Connects to the replication slot. - Reads change events. - For each INSERT/UPDATE/DELETE, writes to warehouse staging table with action type. - Tracks position; resumes from last position on restart. ```python import psycopg2 import psycopg2.extras conn = psycopg2.connect(...) conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT) cur = conn.cursor() cur.execute("START_REPLICATION SLOT cdc_slot LOGICAL 0/0 (proto_version '1', publication_names 'cdc_pub')") while True: msg = cur.consume_stream() if msg: change = parse_pgoutput(msg.payload) write_to_warehouse_staging(change) msg.cursor.send_feedback(flush_lsn=msg.data_start) ``` ### Step 4: Apply changes to warehouse (15 min) Build a dbt model or scheduled merge that: - Reads the staging table (changes log). - Applies them to the warehouse target tables. - Handles INSERT, UPDATE (merge), and DELETE. Run every minute via Airflow or a loop. ## Deliverables - Working CDC pipeline (Postgres → warehouse). - Lag dashboard: how far behind is the warehouse vs source? - Test: run the traffic generator, verify warehouse matches source within 5 min. - A document explaining the gotchas you hit (snapshot vs streaming start, schema changes, dropped tables). ## What you'll learn - Why CDC is harder than it sounds. - Replication slot management (they fill up if consumer dies). - Handling DELETE (most CDC tools don't propagate well without care). - Initial snapshot + ongoing stream coordination. - Why production teams pay for Fivetran or Debezium-managed.