WAL Log Enrichment Pipeline

Design a CDC pipeline that captures WAL entries at 1M writes/sec, enriches them (IDs → names/regions via a Redis-fronted lookup), and delivers them in per-key order with exactly-once semantics (Kafka transactions + LSN dedupe), DLQ, and replay. Design a WAL Log Enrichment Pipeline This system needs to read Write-Ahead Log (WAL) entries from a main database, add extra details (enrichment) to them, and send them to a final database. The system must handle 1 million writes per second. It must be fast and accurate. We assume the source database uses Sharding or has multiple primary nodes. Phase 1: What We Need to Build System Requirements Capture logs: The system must read all WAL entries from the source database immediately. Add context: The system must add extra info to the logs (like table details or user names). Deliver data: The system must send the finished data to the target database in the correct order. Handle changes: The system must not break if the database structure (schema) changes. Replay ability: The system must be able to re-read old data to fix errors. "Enrichment" means turning a raw log (which usually only has IDs) into a full record. For example, it turns a user_id into a record with the user's name and location. Performance Goals Requirement Target Reason Throughput 1M writes/sec Very high volume (like Netflix). Latency < 5 seconds Data needs to be ready almost instantly. Ordering Per-key ordering Updates for the same ID must stay in order. Durability No data loss We cannot lose any updates. Availability 99.99% uptime The system must always be running. Unlike a website where speed is measured in milliseconds, data pipelines usually aim for seconds. A 5-second delay is okay, but losing data is not. Size Estimations Metric Value Write speed 1M events/sec Event size (Raw) ~1 KB Event size (Enriched) ~2 KB (after adding info) Input speed 1M × 1 KB = 1 GB/sec Output speed 1M × 2 KB = 2 GB/sec Daily storage ~86 TB/day Retention (7 days) ~600 TB At 1 million writes per second, the hard parts are: Reading fast enough from the source. Finding extra info (enrichment) without slowing down. Writing to the target without crashing it. Phase 2: How We Store Data Log Entry Structure WALEntry (from source) ├── lsn: BIGINT (Log Sequence Number - unique ID for the log) ├── timestamp: TIMESTAMP ├── transaction_id: BIGINT ├── table: VARCHAR ├── operation: ENUM (INSERT, UPDATE, DELETE) ├── key: JSONB (primary key fields) ├── before: JSONB (values before the change) └── after: JSONB (values after the change) EnrichedEntry (to target) ├── lsn: BIGINT ├── timestamp: TIMESTAMP ├── transaction_id: BIGINT ├── table: VARCHAR ├── operation: ENUM ├── key: JSONB ├── before: JSONB ├── after: JSONB ├── enrichment: JSONB (added details like names/regions) ├── enriched_at: TIMESTAMP └── schema_version: INTEGER Saving Our Place (Checkpoints) We track our progress in two places to ensure we don't lose data: CDC Connector: Saves the last read LSN. Kafka Consumers: Save the last processed message offset. # CDC checkpoint (Saved by the connector) { "connector": "postgres-cdc", "database": "netflix_prod", "lsn": 847293847, "txId": 12345 } # Consumer offset (Managed by Kafka) { "topic": "enriched-changes.orders", "partition": 42, "offset": 1000000, "consumer_group": "sink-clickhouse" } Important Design Choices LSN (Log Sequence Number): This is a unique ID for every change in PostgreSQL. We use it to remove duplicates. We use it to restart processing from the exact right spot. Separate topics per table: Each table gets its own Kafka topic. This helps us scale. Big tables get more partitions. Schema versioning: We save the schema_version. This helps the consumer understand the data even if the format changes later. Phase 3: Interface Design We do not use standard web APIs (REST) for the data flow. We use database protocols and Kafka messages. Data Contracts CDC Connector → Kafka (raw-changes topic) The connector sends events that look like this: { "lsn": 847293847, "timestamp": "2024-02-15T10:30:00Z", "table": "orders", "operation": "INSERT", "key": {"order_id": "ord_123"}, "before": null, "after": {"order_id": "ord_123", "user_id": "usr_456", "amount": 1999} } Enrichment Worker → Kafka (enriched-changes topic) The worker adds the extra info: { "lsn": 847293847, "timestamp": "2024-02-15T10:30:00Z", "table": "orders", "operation": "INSERT", "key": {"order_id": "ord_123"}, "before": null, "after": {"order_id": "ord_123", "user_id": "usr_456", "amount": 1999}, "enrichment": {"user_name": "John Doe", "user_region": "us-west"}, "schema_version": 3 } Admin APIs We use REST APIs only for managing the system, not for moving data. # Force the system to re-read old data POST /admin/replay { "consumer_group": "enrichment_pipeline", "partition": 0, "from_offset": 1000000, "to_offset": 1500000 } # Stop or Start the pipeline POST /admin/pipeline/{action} # action: pause, resume # Check system health GET /admin/metrics { "source_lsn": 847293847, "consumer_lag_seconds": 2.3, "throughput_per_sec": 980000, "error_rate": 0.001 } Phase 4: System Architecture Component Roles CDC Connector (Debezium) Reads the logical replication slot from PostgreSQL. Turns WAL entries into clear change events. Sends these events to Kafka. Kafka Topics raw-changes.{table}: Stores raw data from the source. enriched-changes.{table}: Stores the finished data. Organized by Primary Key to keep order correct. Enrichment Workers Programs that read from raw-changes. They fetch extra info (like resolving a foreign key). They check a Redis Cache first. If the data isn't there, they check the database. They send the result to enriched-changes. Enrichment Cache (Redis) Stores popular data (users, products) in memory. Updated by a cache warmer (a separate tool that watches for changes to user/product tables). If data is missing here (Cache Miss), the worker checks the database. Enrichment Data Source A read-only copy of the database. Used only when data is missing from Redis. Sink Connector Reads from enriched-changes. Writes data to the final database (e.g., ClickHouse). Uses LSN to remove duplicates. Data Flow Example 1. Raw WAL entry: We get an order with a user_id and product_id. { "lsn": 847293847, "table": "orders", "operation": "INSERT", "key": {"order_id": "ord_123"}, "after": { "order_id": "ord_123", "user_id": "usr_456", "product_id": "prod_789", "amount_cents": 1999 } } 2. After Enrichment: We added the user's name and the product's name. { "lsn": 847293847, "table": "orders", "operation": "INSERT", "key": {"order_id": "ord_123"}, "after": { "order_id": "ord_123", "user_id": "usr_456", "product_id": "prod_789", "amount_cents": 1999 }, "enrichment": { "user_name": "John Doe", "user_region": "us-west-2", "user_tier": "premium", "product_name": "Netflix Gift Card", "product_category": "gift_cards" }, "schema_version": 3 } Ensuring Order We guarantee Per-key ordering. This means all changes to order_123 happen in the correct order. We do this by: Partitioning by Primary Key: All events for order_123 go to the same Kafka partition. One Worker per Partition: Only one worker processes that partition. Sequential Commits: We finish one item before confirming it is done. Global ordering (ordering every single event across the whole system) is too slow for 1 million writes per second. We don't need it. Phase 5: Handling High Load & Problems Meeting Performance Goals Throughput: 1M events/sec Strategy Why it works Kafka partitioning 500 partitions = 500 workers running at the same time. Batch processing Process 1000 events at once, instead of one by one. Batch lookups Ask Redis for 1000 keys in one request (MGET). Horizontal scaling Add more workers if the load increases. Math check: 1M events/sec ÷ 500 partitions = 2,000 events/sec per partition. If batch size is 1,000, that is only 2 batches per second per worker. This is very easy for a single worker to handle. Latency: < 5 seconds Strategy Why it works In-memory caching Redis answers in less than 1 millisecond. Small batch sizes Smaller batches send data faster. Pre-computed data Keep the cache ready before the data is requested. Solving Bottlenecks 1. CDC Connector is too slow Solution: Use multiple replication slots (split by table). Solution: Filter tables. Only capture what you really need. 2. Looking up extra info is slow Solution: Use Batch lookups. Get all IDs from a batch and ask Redis once. Solution: Cache warming. Fill the cache before the traffic hits. # Batch lookup pattern keys = [event.user_id for event in batch] enrichments = redis.mget(keys) # One request for 1000 items 3. Writing to target is slow Solution: Write in batches (1000 items at a time). Solution: Use a database meant for analytics (like ClickHouse), not a standard DB. 4. Kafka is too slow Solution: Add more partitions. Solution: Turn on compression (LZ4). It makes data 3-5x smaller. Exactly-Once Processing We must ensure we don't process a duplicate event if a network error occurs. Solution: Kafka Transactions # Enrichment worker pseudo-code def process_batch(events): enriched = enrich(events) # Do this all at once (Atomic) producer.begin_transaction() try: producer.send(enriched_topic, enriched) # Save our place in the stream producer.send_offsets_to_transaction(consumer_offsets) producer.commit_transaction() except: producer.abort_transaction() raise # Try the whole batch again # If it crashes before commit: we try again. # If it crashes after commit: we are done. No duplicates. Safety at the Target: We also use the LSN as a unique key in the final database. If we try to write the same LSN twice, the database ignores the second one. This is our backup safety net. Handling Errors: Dead Letter Queue (DLQ) If one event in a batch fails (e.g., bad data), we shouldn't stop the whole pipeline. Solution: Send the bad event to a separate "Dead Letter Queue" topic and continue. def process_batch(events): # Step 1: Get extra info for everyone keys = extract_lookup_keys(events) cache_results = redis.mget(keys) # ... logic to fetch missing keys from DB ... # Step 2: Add info. Separate good items from bad ones. enriched, failed = [], [] for event in events: try: enriched.append(apply_enrichment(event, cache_results)) except EnrichmentError as e: failed.append((event, str(e))) # Save the failure # Step 3: Send good data to target, bad data to DLQ producer.begin_transaction() try: producer.send(enriched_topic, enriched) if failed: producer.send(dlq_topic, failed) # Send to bad pile producer.send_offsets_to_transaction(consumer_offsets) producer.commit_transaction() except: producer.abort_transaction() raise Schema Changes Problem: The source database adds a new column. Solution: Store schemas in a registry (like Confluent Schema Registry). Update the enrichment worker first to handle the new field. Then, update the source database. Recovery from Failure Connector fails: It restarts and reads from the saved LSN. No data is lost. Worker fails: A new worker takes over. It starts from the last committed offset. Target DB fails: The Sink Connector waits and retries. Data piles up in Kafka until the DB is back. Review Checklist Requirements Functional: Capture, Enrich, Deliver, Replay. Scale: 1M writes/sec. Rules: Order by Key, No data loss. Data Model Explained WAL structure (LSN, before/after). Explained Checkpoints (how we save our spot). API Design Explained we use data streams, not REST APIs. Mentioned Admin APIs for fixing problems. System Architecture Drew the flow: CDC → Kafka → Worker → Sink. Explained Partitioning for ordering. Explained Cache + Batching for speed. Scaling & Problems Proved 1M/sec works with Math (Partitioning). Explained Exactly-Once (Kafka Transactions). Explained DLQ for bad messages. Explained how to handle crashes. Important Takeaways Use CDC: Don't ask the database "what changed?" repeatedly. Read the logs directly. Kafka is key: It handles the durability, ordering, and buffering. Batch everything: Lookups, processing, and writing must be done in groups, not one by one. Order by Key: You don't need to order everything globally, just per ID. Cache for speed: Use Redis for the fast path. Use the DB only when necessary. Double protection: Use Kafka transactions AND unique keys in the target DB to prevent duplicates.