Implement CDC for real-time data synchronization. Covers Debezium, log-based CDC, query-based CDC, outbox pattern, event sourcing, and CDC pipeline architecture.
Change Data Capture (CDC) is a set of techniques for identifying and propagating changes made to a data store to downstream systems. This approach matters when you need to maintain consistency between systems, support real-time analytics, or build event-driven architectures where data movement must be both timely and complete.
Log-based CDC reads transaction logs from the source database to detect changes. Most relational databases support this through built-in log mechanisms.
For PostgreSQL, enable logical replication with:
CREATE PUBLICATION mypub FOR TABLE users, orders;Then connect using pg_recvlogical:
pg_recvlogical -d mydb -S myslot -R -f -t 10The -S flag specifies the replication slot name. If a replication slot is not available, the command fails with:
ERROR: replication slot "myslot" does not existThe pg_recvlogical tool does not handle slot creation automatically; you must create it first using pg_create_logical_replication_slot.
Polling CDC reads changes by querying a timestamp or version column. This method works with any system that supports SQL.
Example query for detecting new or updated rows:
SELECT * FROM users WHERE updated_at > '2023-01-01T00:00:00Z';For efficient polling, use a cursor with a limit:
SELECT id, name, updated_at
FROM users
WHERE updated_at > $1
ORDER BY updated_at ASC
LIMIT 1000;When using a timestamp column, ensure it has sufficient precision. A DATE type will silently truncate to day-level granularity, leading to missed changes.
Trigger-based CDC uses database triggers to capture changes into a dedicated audit table. This method works with any SQL system that supports triggers.
Example trigger function:
CREATE OR REPLACE FUNCTION audit_trigger()
RETURNS TRIGGER AS $$
BEGIN
IF TG_OP = 'INSERT' THEN
INSERT INTO audit_log (table_name, operation, row_data)
VALUES (TG_TABLE_NAME, 'INSERT', row_to_json(NEW));
RETURN NEW;
END IF;
END;
$$ LANGUAGE plpgsql;Create the trigger:
CREATE TRIGGER users_audit_trigger
AFTER INSERT OR UPDATE OR DELETE ON users
FOR EACH ROW EXECUTE FUNCTION audit_trigger();If a trigger function raises an exception, the original operation fails and must be retried manually. The error message will show the trigger's failure, but not the specific row that caused it.
A change stream processes each modified row individually. This is useful for systems requiring fine-grained change handling.
Example change processing function:
def process_change(change):
operation = change['operation']
table = change['table']
data = change['data']
if operation == 'INSERT':
# Handle insert logic
pass
elif operation == 'UPDATE':
# Handle update logic
pass
elif operation == 'DELETE':
# Handle delete logic
passWhen processing changes, ensure that operations are idempotent. A repeated change event should not alter the target system's state.
Batched processing collects multiple changes and applies them together, reducing overhead.
Example batch processing:
def process_batch(changes):
# Group by table for efficient batch operations
grouped = {}
for change in changes:
table = change['table']
if table not in grouped:
grouped[table] = []
grouped[table].append(change)
for table, table_changes in grouped.items():
# Process each table's changes as a batch
passWhen batching, be careful about ordering. Changes within a batch must maintain their original sequence to ensure correct application.
Resolution handles conflicts between different change sources or versions.
Example conflict resolution logic:
def resolve_conflict(changes):
# Sort by timestamp and apply in order
sorted_changes = sorted(changes, key=lambda x: x['timestamp'])
# Apply each change in sequence
for change in sorted_changes:
apply_change(change)If two changes modify the same row with different timestamps, the later one overwrites the earlier one. This can silently drop changes if not properly tracked.
Changes are typically serialized as JSON objects containing metadata and data.
Example change object:
{
"operation": "UPDATE",
"table": "users",
"timestamp": "2023-01-01T12:00:00Z",
"data": {
"id": 1,
"name": "John Doe",
"email": "[email protected]"
},
"old_data": {
"name": "Jane Doe"
}
}The old_data field contains the previous values for update operations. If an operation does not support this, the field will be missing, and downstream systems must handle the absence.
Some systems serialize changes in binary formats for performance.
Example using a custom binary format:
import struct
def encode_change(operation, table, data):
# Pack operation type, table name length, table name, data length, data
return struct.pack(
f'!B{len(table)}sI{len(data)}s',
operation,
table.encode(),
len(data),
data.encode()
)Binary serialization requires both sender and receiver to agree on the format. Mismatched formats will cause decode errors.
Changes are pushed directly to a destination system using HTTP or message queues.
Example HTTP push:
import requests
def push_change(change):
response = requests.post(
'http://destination.example.com/changes',
json=change,
headers={'Content-Type': 'application/json'}
)
if response.status_code != 200:
raise Exception(f"Delivery failed: {response.text}")If the destination system is unavailable, delivery will fail with a 5xx error or timeout. The change will need to be retried.
Changes are stored in a queue before delivery, providing reliability.
Example using a message queue:
import json
def queue_change(change):
# Send to queue system
queue.put(json.dumps(change))Queues provide at-least-once delivery guarantees. If a change is processed but the acknowledgment fails, it may be delivered again.
When changes are lost, recovery must identify and replay missing events.
Example recovery process:
def recover_missing_changes(last_processed_id):
# Query for changes that occurred after last processed event
changes = fetch_changes_after(last_processed_id)
# Apply changes in order
for change in changes:
apply_change(change)If a system lacks a reliable sequence number, recovery becomes difficult and may miss changes.
Duplicate changes must be handled to prevent data corruption.
Example duplicate detection:
def is_duplicate(change):
# Check if this change was already processed
return db.exists('SELECT 1 FROM processed_changes WHERE id = ?', (change['id'],))If a system does not track which changes have been applied, it will process the same change multiple times.
Retry logic handles temporary failures in delivery or processing.
Example retry mechanism:
import time
def retry_with_backoff(func, max_retries=3):
for attempt in range(max_retries):
try:
return func()
except Exception as e:
if attempt == max_retries - 1:
raise e
time.sleep(2 ** attempt)Retries may cause duplicate processing. Ensure that retry logic does not create race conditions.
Track which changes have been processed:
CREATE TABLE change_log (
id BIGSERIAL PRIMARY KEY,
change_id UUID NOT NULL,
processed_at TIMESTAMP DEFAULT NOW(),
status VARCHAR(20) NOT NULL
);If a change is not marked as processed, the system will retry it.
Report errors during change processing:
def process_change_with_error_handling(change):
try:
process_change(change)
mark_as_processed(change['id'])
except Exception as e:
report_error(change, str(e))
raiseErrors must be logged with sufficient context for debugging.
Validate that changes conform to expected formats:
def validate_change(change):
required_fields = ['operation', 'table', 'data']
for field in required_fields:
if field not in change:
raise ValueError(f"Missing required field: {field}")If validation fails, the system should reject the change rather than attempt processing.
Timestamps from different systems may have different time zones or clocks.
Example of problematic timestamp handling:
# This will fail if clocks are not synchronized
def process_changes(changes):
for change in changes:
# Using local time instead of UTC
if change['timestamp'] > datetime.now():
# Future-dated change
passAlways use UTC timestamps and validate that systems have synchronized clocks.
Schema evolution can break change processing.
Example schema change:
ALTER TABLE users ADD COLUMN phone VARCHAR(20);If a change processing system does not handle new columns, it may silently ignore them or fail on unexpected fields.
Processing changes in wrong order can cause data inconsistency.
Example ordering problem:
# Wrong: Processing without maintaining order
def process_changes_unordered(changes):
for change in changes:
apply_change(change)Maintain proper ordering by processing changes in timestamp order or by using sequence numbers.
Processing large volumes of changes can exhaust system resources.
Example resource management:
def process_changes_with_limits(changes, batch_size=1000):
for i in range(0, len(changes), batch_size):
batch = changes[i:i + batch_size]
process_batch(batch)Always limit batch sizes to prevent memory overflow or timeout issues.
The key to successful CDC implementation is understanding the specific requirements of your data flow and handling failures gracefully. Each pattern must be carefully tested for edge cases and integrated with appropriate monitoring to detect and resolve issues quickly.
This page was rewritten on 15 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.
We use cookies for analytics (Google Analytics) and advertising (Google AdSense) to improve your experience and support free content. Privacy Policy