Change Data Capture (CDC)

BATCH ETL (Old Approach):
┌────────────────────────────────────────────────┐
│  Production Database (PostgreSQL)              │
│  • INSERT, UPDATE, DELETE happen continuously  │
└─────────────┬──────────────────────────────────┘
              │
              ▼ Every 1 hour
┌────────────────────────────────────────────────┐
│  ETL Job: SELECT * WHERE updated_at > ?        │
│  • Full table scan (slow)                      │
│  • Impacts production (locks, CPU)             │
│  • Misses deletes (no timestamp)               │
└─────────────┬──────────────────────────────────┘
              │
              ▼
┌────────────────────────────────────────────────┐
│  Data Warehouse / Lake                         │
│  • Data 1-24 hours old                         │
└────────────────────────────────────────────────┘

Key Issues:
• High latency (hours)
• Production database impact
• Can't track deletes
• Inefficient (scans entire table)

CDC Pipeline:
┌────────────────────────────────────────────────┐
│  Production Database (PostgreSQL)              │
│  • INSERT → WAL log                            │
│  • UPDATE → WAL log                            │
│  • DELETE → WAL log                            │
└─────────────┬──────────────────────────────────┘
              │ Read transaction log
              ▼ Continuous streaming
┌────────────────────────────────────────────────┐
│  CDC Engine (Debezium/DMS)                     │
│  • Reads transaction log                       │
│  • Zero impact on production                   │
│  • Captures all changes (INSERT/UPDATE/DELETE) │
└─────────────┬──────────────────────────────────┘
              │ Stream events (Kafka, Kinesis)
              ▼
┌────────────────────────────────────────────────┐
│  Downstream Systems                            │
│  • Data Lake (S3)                              │
│  • Search Index (Elasticsearch)                │
│  • Cache (Redis)                               │
│  • Analytics (Snowflake)                       │
│  • Microservices                               │
│  Data freshness: < 1 second                    │
└────────────────────────────────────────────────┘

CDC Advantages:
• Real-time (milliseconds to seconds)
• Zero production impact (reads logs only)
• Captures all change types (INSERT, UPDATE, DELETE)
• Efficient (no full table scans)

# postgresql.conf changes:
wal_level = logical
max_replication_slots = 10
max_wal_senders = 10

# Restart PostgreSQL
sudo systemctl restart postgresql

# Create replication user
CREATE USER debezium_user WITH REPLICATION LOGIN;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium_user;

# Debezium runs as Kafka Connect connector
# docker-compose.yml:
version: '3'
services:
zookeeper:
image: confluentinc/cp-zookeeper:latest
environment:
  ZOOKEEPER_CLIENT_PORT: 2181

kafka:
image: confluentinc/cp-kafka:latest
depends_on: [zookeeper]
environment:
  KAFKA_BROKER_ID: 1
  KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
  KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092

debezium:
image: debezium/connect:latest
depends_on: [kafka]
ports:
  - 8083:8083
environment:
  BOOTSTRAP_SERVERS: kafka:9092
  GROUP_ID: 1
  CONFIG_STORAGE_TOPIC: connect_configs
  OFFSET_STORAGE_TOPIC: connect_offsets

# Start: docker-compose up -d

import requests
import json

connector_config = {
"name": "postgres-connector",
"config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres.example.com",
    "database.port": "5432",
    "database.user": "debezium_user",
    "database.password": "secret",
    "database.dbname": "mydb",
    "database.server.name": "mydb",
    "table.include.list": "public.users,public.orders",
    "plugin.name": "pgoutput",  # PostgreSQL 10+ logical decoding plugin
    "publication.name": "dbz_publication",
    "slot.name": "debezium_slot"
}
}

# Register connector via REST API
response = requests.post(
"http://localhost:8083/connectors",
headers={"Content-Type": "application/json"},
data=json.dumps(connector_config)
)

print(f"Connector registered: {response.status_code}")

# Check connector status
status = requests.get("http://localhost:8083/connectors/postgres-connector/status")
print(status.json())

"""
Response:
{
"name": "postgres-connector",
"connector": {"state": "RUNNING"},
"tasks": [{"id": 0, "state": "RUNNING"}]
}
"""

# Connector is now streaming all changes from users and orders tables to Kafka!

from kafka import KafkaConsumer
import json

# Connect to Kafka
consumer = KafkaConsumer(
'mydb.public.users',  # Topic: <server>.<schema>.<table>
bootstrap_servers=['localhost:9092'],
value_deserializer=lambda m: json.loads(m.decode('utf-8')),
auto_offset_reset='earliest'
)

# Process change events
for message in consumer:
event = message.value

# Event structure from Debezium
operation = event['payload']['op']  # c (create), u (update), d (delete), r (read/snapshot)
after = event['payload'].get('after')  # New row state
before = event['payload'].get('before')  # Old row state (for updates/deletes)
source = event['payload']['source']

print(f"\n=== Change Event ===")
print(f"Operation: {operation}")
print(f"Table: {source['table']}")
print(f"LSN: {source['lsn']}")

if operation == 'c':  # INSERT
    print(f"New user created:")
    print(f"  ID: {after['id']}")
    print(f"  Name: {after['name']}")
    print(f"  Email: {after['email']}")

    # Sync to downstream system
    sync_to_elasticsearch(after)

elif operation == 'u':  # UPDATE
    print(f"User updated:")
    print(f"  Before: {before}")
    print(f"  After: {after}")

    # Update cache
    invalidate_cache(after['id'])

elif operation == 'd':  # DELETE
    print(f"User deleted:")
    print(f"  ID: {before['id']}")

    # Remove from search index
    remove_from_elasticsearch(before['id'])

elif operation == 'r':  # READ (initial snapshot)
    print(f"Snapshot row: {after}")

# Example output when user inserted in PostgreSQL:
"""
=== Change Event ===
Operation: c
Table: users
LSN: 0/167B4F8
New user created:
ID: 456
Name: Bob Smith
Email: bob@example.com

[Synced to Elasticsearch in 150ms]
"""

# Example output when user updated:
"""
=== Change Event ===
Operation: u
Table: users
LSN: 0/167B600
User updated:
Before: {'id': 456, 'name': 'Bob Smith', 'email': 'bob@example.com'}
After: {'id': 456, 'name': 'Bob Smith', 'email': 'bob.smith@example.com'}

[Cache invalidated for user 456]
"""

# Example output when user deleted:
"""
=== Change Event ===
Operation: d
Table: users
LSN: 0/167B710
User deleted:
ID: 456

[Removed from Elasticsearch]
"""

import boto3

dms = boto3.client('dms', region_name='us-east-1')
    replication_instance = dms.create_replication_instance(
    ReplicationInstanceIdentifier='my-replication-instance',
    ReplicationInstanceClass='dms.c5.large',
    AllocatedStorage=100,
    VpcSecurityGroupIds=['sg-12345'],
    AvailabilityZone='us-east-1a',
    PubliclyAccessible=False
)

print(f"Replication instance created: {replication_instance['ReplicationInstance']['ReplicationInstanceArn']}")

# Wait for instance to be available
waiter = dms.get_waiter('replication_instance_available')
waiter.wait(Filters=[{'Name': 'replication-instance-id', 'Values': ['my-replication-instance']}])

source_endpoint = dms.create_endpoint(
    EndpointIdentifier='source-postgres',
    EndpointType='source',
    EngineName='postgres',
    ServerName='prod-db.example.com',
    Port=5432,
    DatabaseName='mydb',
    Username='dms_user',
    Password='secret'
)

# Test connection
dms.test_connection(
    ReplicationInstanceArn=replication_instance['ReplicationInstance']['ReplicationInstanceArn'],
    EndpointArn=source_endpoint['Endpoint']['EndpointArn']
)

target_endpoint = dms.create_endpoint(
    EndpointIdentifier='target-s3',
    EndpointType='target',
    EngineName='s3',
    S3Settings={
        'ServiceAccessRoleArn': 'arn:aws:iam::123456789:role/dms-s3-role',
        'BucketName': 'my-data-lake',
        'BucketFolder': 'cdc-data',
        'DataFormat': 'parquet',
        'CompressionType': 'gzip',
        'TimestampColumnName': 'cdc_timestamp',
        'ParquetTimestampInMillisecond': True
    }
)

import json

# Table mappings: which tables to replicate
table_mappings = {
"rules": [
    {
        "rule-type": "selection",
        "rule-id": "1",
        "rule-name": "include-users-orders",
        "object-locator": {
            "schema-name": "public",
            "table-name": "%"  # All tables
        },
        "rule-action": "include"
    },
    {
        "rule-type": "transformation",
        "rule-id": "2",
        "rule-name": "add-column",
        "rule-target": "column",
        "object-locator": {
            "schema-name": "public",
            "table-name": "%"
        },
        "rule-action": "add-column",
        "value": "cdc_timestamp",
        "expression": "$AR_H_TIMESTAMP",
        "data-type": {
            "type": "datetime"
        }
    }
]
}

replication_task = dms.create_replication_task(
    ReplicationTaskIdentifier='postgres-to-s3-cdc',
    SourceEndpointArn=source_endpoint['Endpoint']['EndpointArn'],
    TargetEndpointArn=target_endpoint['Endpoint']['EndpointArn'],
    ReplicationInstanceArn=replication_instance['ReplicationInstance']['ReplicationInstanceArn'],
    MigrationType='full-load-and-cdc',  # Full load then CDC
    TableMappings=json.dumps(table_mappings),
    ReplicationTaskSettings=json.dumps({
        "TargetMetadata": {
            "SupportLobs": True,
            "LobMaxSize": 32  # MB
        },
        "FullLoadSettings": {
            "TargetTablePrepMode": "DROP_AND_CREATE"
        },
        "Logging": {
            "EnableLogging": True
        }
    })
)

print(f"Replication task created: {replication_task['ReplicationTask']['ReplicationTaskArn']}")

dms.start_replication_task(
    ReplicationTaskArn=replication_task['ReplicationTask']['ReplicationTaskArn'],
    StartReplicationTaskType='start-replication'
)

print("Replication started!")

import time

while True:
response = dms.describe_replication_tasks(
    Filters=[{'Name': 'replication-task-arn',
              'Values': [replication_task['ReplicationTask']['ReplicationTaskArn']]}]
)

task = response['ReplicationTasks'][0]
status = task['Status']
stats = task.get('ReplicationTaskStats', {})

print(f"\nStatus: {status}")
print(f"Full Load Progress: {stats.get('FullLoadProgressPercent', 0)}%")
print(f"Tables Loaded: {stats.get('TablesLoaded', 0)}")
print(f"Rows Loaded: {stats.get('FullLoadRowsLoaded', 0)}")

if status == 'running' and stats.get('FullLoadProgressPercent') == 100:
    print("\n✓ Full load complete, CDC replication active!")
    break

time.sleep(10)

"""
Output:
Status: running
Full Load Progress: 45%
Tables Loaded: 5
Rows Loaded: 1,245,890

...

Status: running
Full Load Progress: 100%
Tables Loaded: 12
Rows Loaded: 5,000,000

✓ Full load complete, CDC replication active!

Result:
• All existing data copied to S3 as Parquet
• Ongoing changes streaming continuously
• Latency: < 10 seconds
• Zero downtime for source database
"""

# Stream database changes to Kinesis for real-time processing
target_kinesis = dms.create_endpoint(
EndpointIdentifier='target-kinesis',
EndpointType='target',
EngineName='kinesis',
KinesisSettings={
    'StreamArn': 'arn:aws:kinesis:us-east-1:123456789:stream/db-changes',
    'MessageFormat': 'json',
    'ServiceAccessRoleArn': 'arn:aws:iam::123456789:role/dms-kinesis-role',
    'IncludeTransactionDetails': True,
    'IncludeTableAlterOperations': True
}
)

# Now every database change flows to Kinesis in real-time
# Consume with Lambda, Flink, or any Kinesis consumer

# Example: Process with AWS Lambda
import boto3
import json

def lambda_handler(event, context):
"""Process CDC events from Kinesis"""
for record in event['Records']:
    # Decode Kinesis record
    payload = json.loads(record['kinesis']['data'])

    # DMS change event
    operation = payload['metadata']['operation']  # insert, update, delete
    table = payload['metadata']['table-name']
    data = payload['data']

    print(f"Change detected: {operation} on {table}")

    if operation == 'insert':
        # Send to Elasticsearch
        index_to_elasticsearch(table, data)

    elif operation == 'update':
        # Invalidate cache
        invalidate_cache(table, data['id'])

    elif operation == 'delete':
        # Remove from search
        remove_from_elasticsearch(table, data['id'])

return {'statusCode': 200}

# Result: Sub-second latency from database to downstream systems

Traditional State Storage vs Event Sourcing:

TRADITIONAL (Current State Only):
┌────────────────────────────────┐
│  Database                      │
│  users table:                  │
│  ┌──────┬────────┬───────────┐ │
│  │ id   │ name   │ balance   │ │
│  ├──────┼────────┼───────────┤ │
│  │ 123  │ Alice  │ $1,000    │ │  ← Current state only
│  └──────┴────────┴───────────┘ │
└────────────────────────────────┘

Lost information:
• How did balance reach $1,000?
• What were previous balances?
• What operations occurred?


EVENT SOURCING (Event Log):
┌─────────────────────────────────────────────┐
│  Event Store                                │
│  ┌──────────┬─────────────────────────────┐ │
│  │Timestamp │ Event                       │ │
│  ├──────────┼─────────────────────────────┤ │
│  │10:00     │AccountCreated(id=123)       │ │
│  │10:05     │Deposited(amount=500)        │ │
│  │10:15     │Deposited(amount=300)        │ │
│  │10:30     │Withdrew(amount=100)         │ │
│  │10:45     │Deposited(amount=300)        │ │
│  └──────────┴─────────────────────────────┘ │
└─────────────────────────────────────────────┘

Current state: Sum all events = $1,000
Historic state: Replay events to any point in time

Benefits:
• Complete audit trail
• Time travel (replay to any point)
• Event-driven architecture
• Easy debugging ("what changed?")
• Source of truth for downstream systems

# Use CDC as event stream for event sourcing
from kafka import KafkaConsumer
import json
from datetime import datetime, timezone

class EventStore:
    """Store all CDC events as immutable event log"""

    def __init__(self):
        self.events = []  # In production: use database or S3

    def append(self, event):
        """Append event to immutable log"""
        event['event_id'] = len(self.events)
        event['stored_at'] = datetime.now(timezone.utc).isoformat()
        self.events.append(event)

    def get_events(self, entity_id, entity_type):
        """Get all events for an entity"""
        return [e for e in self.events
                if e.get('entity_id') == entity_id
                and e.get('entity_type') == entity_type]

    def replay_to_point(self, entity_id, entity_type, timestamp):
        """Reconstruct state at specific point in time"""
        relevant_events = [
            e for e in self.get_events(entity_id, entity_type)
            if e['timestamp'] <= timestamp
        ]

        # Replay events to build state
        state = {}
        for event in relevant_events:
            state = apply_event(state, event)

        return state


# Consume CDC events and store in event log
event_store = EventStore()

consumer = KafkaConsumer(
    'mydb.public.accounts',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    cdc_event = message.value['payload']

    # Convert CDC event to domain event
    if cdc_event['op'] == 'c':  # Insert
        domain_event = {
            'event_type': 'AccountCreated',
            'entity_id': cdc_event['after']['account_id'],
            'entity_type': 'Account',
            'timestamp': cdc_event['ts_ms'],
            'data': cdc_event['after']
        }

    elif cdc_event['op'] == 'u':  # Update
        before = cdc_event['before']
        after = cdc_event['after']

        # Detect what changed
        if before['balance'] != after['balance']:
            amount = after['balance'] - before['balance']
            event_type = 'Deposited' if amount > 0 else 'Withdrew'

            domain_event = {
                'event_type': event_type,
                'entity_id': after['account_id'],
                'entity_type': 'Account',
                'timestamp': cdc_event['ts_ms'],
                'data': {
                    'amount': abs(amount),
                    'balance_before': before['balance'],
                    'balance_after': after['balance']
                }
            }

    # Store event in immutable log
    event_store.append(domain_event)
    print(f"Event stored: {domain_event['event_type']} for account {domain_event['entity_id']}")


# Query event history
account_events = event_store.get_events(entity_id=123, entity_type='Account')
print(f"\nAccount 123 event history ({len(account_events)} events):")
for event in account_events:
    print(f"  {event['timestamp']}: {event['event_type']} - {event['data']}")

"""
Output:
Account 123 event history (5 events):
  1704034800000: AccountCreated - {'account_id': 123, 'balance': 0}
  1704034860000: Deposited - {'amount': 500, 'balance_before': 0, 'balance_after': 500}
  1704035460000: Deposited - {'amount': 300, 'balance_before': 500, 'balance_after': 800}
  1704036060000: Withdrew - {'amount': 100, 'balance_before': 800, 'balance_after': 700}
  1704036660000: Deposited - {'amount': 300, 'balance_before': 700, 'balance_after': 1000}
"""

# Time travel: What was balance at 10:15 AM?
state_at_1015 = event_store.replay_to_point(
    entity_id=123,
    entity_type='Account',
    timestamp=1704035460000  # 10:15 AM
)
print(f"\nBalance at 10:15 AM: ${state_at_1015['balance']}")  # $800

# Keep Elasticsearch in sync with PostgreSQL using CDC
from kafka import KafkaConsumer
from elasticsearch import Elasticsearch
import json

es = Elasticsearch(['http://localhost:9200'])

consumer = KafkaConsumer(
    'mydb.public.products',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    event = message.value['payload']

    if event['op'] == 'c' or event['op'] == 'r':  # INSERT or Snapshot
        # Index new document
        doc = event['after']
        es.index(
            index='products',
            id=doc['product_id'],
            document={
                'name': doc['name'],
                'description': doc['description'],
                'price': doc['price'],
                'category': doc['category']
            }
        )
        print(f"Indexed product {doc['product_id']}")

    elif event['op'] == 'u':  # UPDATE
        # Update document
        doc = event['after']
        es.update(
            index='products',
            id=doc['product_id'],
            doc={
                'name': doc['name'],
                'description': doc['description'],
                'price': doc['price'],
                'category': doc['category']
            }
        )
        print(f"Updated product {doc['product_id']}")

    elif event['op'] == 'd':  # DELETE
        # Delete document
        doc = event['before']
        es.delete(index='products', id=doc['product_id'])
        print(f"Deleted product {doc['product_id']}")

# Result: Elasticsearch always in sync with PostgreSQL (< 1 second latency)

# Automatically invalidate Redis cache on database changes
import redis

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

consumer = KafkaConsumer(
    'mydb.public.users',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    event = message.value['payload']

    if event['op'] in ['u', 'd']:  # UPDATE or DELETE
        # Invalidate cache
        user_id = event['before']['user_id']

        cache_keys = [
            f"user:{user_id}",
            f"user:{user_id}:profile",
            f"user:{user_id}:preferences"
        ]

        for key in cache_keys:
            redis_client.delete(key)
            print(f"Invalidated cache: {key}")

        # Update with new value if UPDATE
        if event['op'] == 'u':
            new_data = event['after']
            redis_client.setex(
                f"user:{user_id}",
                3600,  # 1 hour TTL
                json.dumps(new_data)
            )
            print(f"Updated cache: user:{user_id}")

# Result: Cache always fresh, no stale data

# Sync data across microservices without direct coupling
# Order Service database changes → Inventory Service

consumer = KafkaConsumer(
    'orders_db.public.orders',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    event = message.value['payload']

    if event['op'] == 'c':  # New order created
        order = event['after']

        # Update inventory in Inventory Service
        update_inventory(
            product_id=order['product_id'],
            quantity_change=-order['quantity']
        )

        # Notify Shipping Service
        send_to_shipping_queue({
            'order_id': order['order_id'],
            'customer_address': order['shipping_address']
        })

        print(f"Order {order['order_id']} processed across services")

# Result: Services stay in sync without tight coupling or API calls