Change Data Capture and Streaming Ingestion

The problem. A second system (a data warehouse, a search index, a cache, another service) needs a copy of data that lives in a primary database. Copying the whole table on a schedule is slow and stale. The question is how to learn precisely what changed, reliably and in order.

Three ways that go wrong. Polling for rows with a newer updated_at misses hard deletes and any change made without updating the column, and it hammers the database. Dual writes (the app writes to the database and also to a queue) can fail halfway, so the two disagree forever. Triggers add cost to every write and are easy to forget. Each leaves silent gaps.

Log-based CDC. Every database already writes each committed change to its transaction log (the write-ahead log or binlog) so it can recover and replicate. A CDC connector reads that log like a replica would and publishes one event per row change to a stream. It sees inserts, updates and deletes, in commit order, without touching the application and with little load.

Shape of a change event. A useful event carries the table and primary key, the operation (create, update, delete), the row before and after (or just after), a position in the log (an offset), and a commit time. The primary key is the ordering key, so all changes to one row go to the same partition and arrive in order. A delete is its own event, and downstream a delete is often also followed by a tombstone so a compacted log can forget the key.

Snapshot, then stream. A new consumer cannot start from the middle of history. It first loads a consistent snapshot of the tables and notes the log position at which the snapshot was taken, then applies every change after that position. Many tools do this as one step and call it the initial load.

table = {}        # downstream copy: key -> (row, position of the change that wrote it)

def apply(event):
    key, op, row, pos = event['key'], event['op'], event.get('row'), event['pos']
    seen = table.get(key)
    if seen and seen[1] >= pos:          # an older or duplicate event: ignore it
        return
    if op == 'delete':
        table[key] = (None, pos)         # keep the position so a late replay cannot resurrect it
    else:
        table[key] = (row, pos)

events = [
    dict(key=1, op='create', row={'name': 'Ana'}, pos=10),
    dict(key=1, op='update', row={'name': 'Anna'}, pos=11),
    dict(key=1, op='update', row={'name': 'Anna'}, pos=11),     # delivered twice
    dict(key=2, op='create', row={'name': 'Ben'}, pos=12),
    dict(key=2, op='delete', pos=13),
    dict(key=2, op='create', row={'name': 'Ben'}, pos=12),      # a late replay of an old event
]
for e in events:
    apply(e)
print({k: v for k, v in table.items() if v[0] is not None})     # {1: ({'name': 'Anna'}, 11)}

Idempotent apply is the safety net. Streams deliver at least once, so duplicates and replays will happen. If each apply is an upsert keyed by the primary key and guarded by the log position (as above), then applying an event twice, or an old event late, leaves the right result. That gives you an exactly-once effect on top of at-least-once delivery.

Alternatives and add-ons. The transactional outbox writes events into an outbox table in the same transaction as the business change, and a relay publishes from that table: it makes events explicit and shaped for consumers, at the cost of an extra table. Streaming ingestion into a lake or warehouse usually writes micro-batches of events as files, then merges them into the table. Reverse ETL does the opposite direction, computing changes in the warehouse and pushing them to business tools while respecting their rate limits.

Operating it. Watch the lag between the database's latest position and the consumer's. Plan for schema changes (a new column must not break consumers), for connector restarts (resume from the stored offset), and for the log retention window (a consumer that is down longer than the log is kept needs a fresh snapshot).