Case 5 — Architecture Evolution: Batch to Real-Time Streaming Migration

# Case 5 — Architecture Evolution: Batch to Real-Time Streaming Migration ## Executive Context & Business Prompt You are a Principal Data Architect at **QuickPay International**. The company's fraud risk scoring pipeline runs as an hourly batch Airflow DAG that queries transactions from Postgres and scores them in Python. Fraudulent actors have exploited this 1-hour detection window to execute card-testing rings, resulting in $340,000 in chargeback losses last month. The Chief Risk Officer requires payment transactions to be evaluated within 15 seconds of occurrence. However, executive BI dashboards, tax reporting, and customer statements all rely on the existing batch tables. Chief Information Officer Prompt: > *"Migrate our fraud detection pipeline to real-time event streaming with Kafka, but do not break the analytics team's batch reporting. Run a dual-write architecture, implement deduplication for at-least-once message delivery, and build an automated reconciliation script asserting numbers match across both paths."* --- ## Explicit Portfolio Deliverables 1. **Migration Architecture Runbook (`STREAMING_MIGRATION_RUNBOOK.md`):** - Details dual-run phase, consumer group design, partition strategy (keyed on `account_id` to ensure per-account ordering), and cutover milestones. 2. **Python Streaming Consumer (`kafka_fraud_consumer.py`):** - Connects to `payments.transactions` topic. - Evaluates high-velocity velocity rules (e.g. $>3$ transactions in $ 0.01\%$. --- ## Starter Code Scaffold ```python from confluent_kafka import Consumer, KafkaError import json import time conf = { 'bootstrap.servers': 'localhost:9092', 'group.id': 'fraud-scoring-engine', 'auto.offset.reset': 'earliest', 'enable.auto.commit': False # Manual offset commit for delivery guarantees } consumer = Consumer(conf) consumer.subscribe(['payments.transactions']) def process_stream(): try: while True: msg = consumer.poll(timeout=1.0) if msg is None: continue if msg.error(): if msg.error().code() == KafkaError._PARTITION_EOF: continue else: print(f"Consumer error: {msg.error()}") break transaction = json.loads(msg.value().decode('utf-8')) # TODO: Implement in-memory velocity evaluation # TODO: Commit offset after successful persistence consumer.commit(asynchronous=False) finally: consumer.close() ``` --- ## 100-Point Grading Rubric | Criteria | Points | Passing Standard | |---|:---:|---| | **1. Partition Key Strategy & Ordering** | 25 pts | Partitions Kafka topic by entity key (`account_id`) ensuring strictly ordered event delivery per customer account. | | **2. Consumer Reliability & Offset Commits** | 25 pts | Implements manual offset management (`enable.auto.commit=False`); commits offsets only after downstream persistence succeeds. | | **3. At-Least-Once Deduplication** | 25 pts | Employs idempotency keys / transaction UUIDs to deduplicate messages and prevent double-counting on consumer restarts. | | **4. Dual-Run Reconciliation & Cutover Plan** | 25 pts | Delivers reconciliation script proving data consistency between batch and streaming tables; provides step-by-step cutover runbook. |