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 $<60$ seconds).
   • Commits offsets reliably using manual offset checkpointing.
3. Dual-Run Data Reconciliation Script (reconcile_batch_stream.py):
   • Compares the total transaction count and dollar volume between the legacy batch table and the new streaming lakehouse table for a given 24-hour test window.
   • Flags discrepancies $>0.01%$.

Starter Code Scaffold

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