CDC Pipeline — PostgreSQL to Data Warehouse

A production-grade Change Data Capture (CDC) pipeline that streams row-level changes from PostgreSQL to any downstream store (S3, BigQuery, Snowflake, Kafka) with sub-second latency.

What it does

• Uses PostgreSQL logical replication (pgoutput plugin) — no polling, no triggers, no performance impact on the source DB
• Processes INSERT, UPDATE, and DELETE events as a typed stream
• Applies configurable transform steps (filter, flatten, enrich)
• Buffers writes and flushes in configurable batches (by size or time)
• Checkpoints the replication LSN to disk — exactly-once delivery after restarts
• Swappable sinks: stdout (dev), S3 (NDJSON), BigQuery, Kafka

Architecture

PostgreSQL WAL
     │
     ▼
PgCDCSource          ← logical replication slot, pgoutput plugin
     │
     ▼
[Transform chain]    ← filter_deleted, flatten_jsonb_columns,
     │                  add_ingestion_timestamp, ...
     ▼
BatchSink            ← buffers, flushes on size OR timer
     │
     ▼
Target (S3 / BQ / stdout)

Quick start

# 1. Create replication slot and publication on Postgres
psql -c "SELECT pg_create_logical_replication_slot('brad_cdc', 'pgoutput');"
psql -c "CREATE PUBLICATION brad_pub FOR TABLE orders, users, products;"

# 2. Install deps
pip install psycopg2-binary boto3

# 3. Run (dev mode — prints to stdout)
PG_DSN="host=localhost dbname=mydb user=replicator password=secret" \
    python cdc_pipeline.py --sink stdout

# 4. Run to S3
python cdc_pipeline.py --sink s3 --s3-bucket my-data-lake --dsn "$PG_DSN"

Adding a custom transform

from cdc_pipeline import ChangeEvent, CDCPipeline, PgCDCSource, BatchSink

def mask_pii(event: ChangeEvent):
    """Redact email fields before they reach the warehouse."""
    if event.after and "email" in event.after:
        event.after["email"] = "***"
    return event

# Plug it in:
pipeline = CDCPipeline(
    source=PgCDCSource(dsn="..."),
    transforms=[mask_pii],
    sink=BatchSink(flush_fn=stdout_sink),
)
pipeline.run()

Configuration

Env var / flag	Default	Description
PG_DSN / --dsn	—	Postgres connection string
--slot	brad_cdc	Replication slot name
--pub	brad_pub	Publication name
--sink	stdout	stdout or s3
--s3-bucket	—	S3 bucket (required if --sink=s3)

BatchSink parameters are set in code (or extend for config file support):

• max_batch_size — flush when buffer hits this count (default: 500)
• flush_interval_s — flush at least this often in seconds (default: 5)

Production notes

• Run with a dedicated Postgres replication user (REPLICATION privilege)
• Keep slot lag monitored — unread WAL causes disk fill-up on Postgres
• For BigQuery target, replace s3_sink_factory with a BigQuery Storage Write API sink (the BatchSink interface stays the same)
• For Kafka target, use confluent-kafka-python in the flush function