Data Engineering & AI: Comprehensive Interview Scenarios

This is a curated collection of real-world interview scenarios covering Data Engineering, AI Integration, System Design, and Production Operations. Each scenario includes context, what the interviewer is testing, and a strong answer framework.

Category 1: System Design Scenarios

Scenario 1: Design a Real-Time Fraud Detection Pipeline

The Ask: "We process 50,000 transactions per second globally. Design a pipeline that detects fraudulent transactions with <100ms latency."

What's Being Tested: Streaming architecture, ML serving latency, system tradeoffs

The "Why" Behind the Design: Before jumping into the architecture, consider why fraud detection is hard: It's a race against time. If you take 2 seconds to decide, the transaction is already approved by the payment gateway. You need to calculate complex aggregations ("how much has this user spent in the last hour across all merchants?") instantly.

Real-World Example: Imagine a user buys a $5 coffee in New York at 10:00 AM, and at 10:05 AM, there's a $2,000 electronics purchase in London on the same card. A batch system running nightly won't catch this until tomorrow. A streaming system sees the New York transaction, updates the user's "last known location" state in memory, and immediately flags the London transaction 5 minutes later because the "velocity distance" is physically impossible.

Strong Answer Framework:

Architecture:
Transaction API → Kafka (partitioned by card_number) 
    → Flink Job (feature computation + model inference)
    → Fraud Score Topic → Alert Service → Block/Allow Decision

Key Design Decisions:

1. Kafka partitioning by card_number ensures all events for one card 
   go to same partition → enables stateful feature computation

2. Flink maintains per-card rolling windows (last 1h, 24h, 7d spend)
3. ML model (XGBoost or neural net) deployed as Flink UDF — no 
   external API call, <5ms inference

4. Feature store serves historical patterns offline-computed daily
5. Result written to Redis (TTL 5min) for deduplication
6. Fallback: if model unavailable, rules-based fallback kicks in

Tradeoffs discussed:

- Kafka vs Kinesis: Kafka if multi-cloud/on-prem, Kinesis if AWS-native
- Online vs Offline features: velocity features online (Flink), 
  historical patterns offline (feature store)

- Model serving: embedded in stream job vs external microservice —
  embedded wins for latency, external wins for model updates

Scenario 2: Build a Data Platform for 100TB/day Ingestion

The Ask: "Your startup just acquired 5 companies. Each has different data stacks. You need to unify 100TB/day of data. Where do you start?"

What's Being Tested: Data lake architecture, multi-source ingestion, prioritization

Strong Answer:

Phase 1 (Week 1-2): Assess and Don't Break Things

- Audit each company's data: schema, volumes, quality, SLAs
- Identify business-critical datasets (revenue, customers, products)
- Set up observability FIRST — can't manage what you can't measure

Phase 2 (Week 3-8): Bronze Layer

- Raw ingestion to S3/GCS with source-specific prefixes
  └── s3://datalake/bronze/{company_name}/{source_system}/{date}/

- Use Fivetran/Airbyte for managed connectors (SaaS sources)
- Custom Spark jobs for proprietary databases
- All data preserved as-is (no transformation in bronze)

Phase 3 (Month 2-3): Silver Layer  

- Common entity resolution: customer_id across 5 companies
- Schema standardization (ISO dates, USD currency)
- Data quality validation with Great Expectations
- Lineage tracking with OpenLineage

Phase 4 (Month 4-6): Gold + Analytics

- Unified customer 360 view
- Consolidated revenue model
- Cross-company analytics enablement

Technology Choices:

- Storage: S3 + Iceberg (ACID, time travel, GDPR compliance)
- Orchestration: Airflow on EKS (scalable, team familiarity)
- Transformation: dbt for SQL, Spark for heavy compute
- Catalog: AWS Glue Catalog + Datahub for lineage

Scenario 3: Text-to-SQL System for Non-Technical Users

The Ask: "Our CEO wants to query data in plain English. Build a system where she types 'What was last quarter's revenue by region?' and gets an answer."

What's Being Tested: LLM integration, safety guardrails, production reliability

The "Why" Behind the Design: Text-to-SQL is fundamentally a translation problem, but with catastrophic downside risk. If an LLM hallucinates in a chatbot, the user gets a weird response. If an LLM hallucinates a SQL query, it might run a Cartesian join that crashes your database, or worse, execute a DROP TABLE command.

Real-World Example: If a user asks, "Show me our top 5 customers," the LLM needs context to know that "top" means "highest revenue" and "customers" lives in dim_customers joined with fct_sales. It also needs to be blocked from running DELETE FROM customers WHERE rank > 5 if it gets confused.

Strong Answer:

# Architecture Components:

# 1. Schema Context Builder (runs at startup/on change)
class SchemaContextBuilder:
    def build_context(self, tables: list[str]) -> str:
        """Build rich context for the LLM."""
        context = []
        for table in tables:
            cols = self.get_column_descriptions(table)
            samples = self.get_sample_values(table, limit=3)
            business_context = self.get_business_context(table)
            context.append(f"""
Table: {table}
Business Purpose: {business_context}
Columns: {cols}
Sample Values: {samples}
Common Joins: {self.get_common_joins(table)}
""")
        return "\n".join(context)

# 2. Query Generator with safety
def generate_safe_sql(question: str, schema_context: str) -> dict:
    response = claude.messages.create(
        model="claude-3-5-sonnet-20241022",
        system="""You are a SQL expert. Rules:

        - ONLY generate SELECT statements
        - NEVER use DROP, DELETE, UPDATE, INSERT
        - Always include LIMIT (max 10000)
        - Return JSON: {"sql": "...", "explanation": "...", "confidence": 0-1}""",
        messages=[{
            "role": "user",
            "content": f"Schema:\n{schema_context}\n\nQuestion: {question}"
        }]
    )
    return json.loads(response.content[0].text)

# 3. Validation layer
def validate_sql(sql: str) -> bool:
    """Multi-layer SQL safety validation."""
    dangerous_keywords = ['DROP', 'DELETE', 'UPDATE', 'INSERT', 'TRUNCATE', 'ALTER']
    sql_upper = sql.upper()
    if any(kw in sql_upper for kw in dangerous_keywords):
        return False
    if not sql_upper.strip().startswith('SELECT'):
        return False
    # Parse and validate with sqlglot
    try:
        import sqlglot
        parsed = sqlglot.parse_one(sql)
        return True
    except:
        return False

# Key considerations to mention:
# - Cache common queries (Redis, 1h TTL)
# - Rate limiting per user
# - Query cost estimation before execution
# - Audit log of all AI-generated queries
# - Human review workflow for sensitive data
# - Fallback to curated question library if confidence < 0.7

Category 2: Data Quality & Reliability Scenarios

Scenario 4: Silent Data Corruption Diagnosis

The Ask: "Finance reported our daily revenue report has been wrong for 3 weeks. How do you diagnose and fix this?"

What's Being Tested: Debugging methodology, lineage understanding, data observability

Strong Methodology:

Step 1: Quantify the problem

- Compare reported revenue vs source system (billing DB)
- Which date range is affected? What's the magnitude of discrepancy?
- Is it consistent error (off by X%) or variable?

Step 2: Trace the lineage

- Revenue Report ← fct_revenue (dbt) ← stg_orders ← raw.orders (Kafka consumer)
- Check each layer: does the discrepancy appear at the same stage?

Step 3: Data audit at each stage
SELECT 
    DATE(order_date) as date,
    COUNT(*) as row_count,
    SUM(revenue) as total_revenue
FROM each_layer_table
WHERE order_date >= '3 weeks ago'
GROUP BY 1 ORDER BY 1;

Step 4: Look for the pattern
Common culprits to check:

- Timezone bug: server timezone changed, timestamps shifted
- Duplicate records: Kafka consumer delivered twice (check offset reset)
- Currency conversion bug: forex rate table not updating
- Schema change: upstream added column, downstream query broke silently
- Filter bug: dbt model WHERE clause changed
- Backfill collision: someone ran historical backfill, overwrote fresh data

Step 5: Git blame the pipeline

- `git log --all --since="3 weeks ago" -- models/marts/fct_revenue.sql`
- Check Airflow run history for that period

Preventive fix:

- Add reconciliation job: compare revenue sum at each pipeline stage daily
- Alert if >1% discrepancy from previous day (anomaly detection)
- Implement data contracts between pipeline stages

Scenario 5: Handle Late-Arriving Data at Scale

The Ask: "Our IoT sensors sometimes send data 48 hours late due to connectivity issues. Our hourly aggregations are wrong when late data arrives. Fix this."

What's Being Tested: Streaming windowing, watermarking, Lambda/Kappa architecture

Strong Answer:

# Option A: Spark Structured Streaming with Watermarking
from pyspark.sql.functions import window, col, sum as _sum

stream = (
    spark.readStream
    .format("kafka")
    .option("subscribe", "iot_events")
    .load()
    .withColumn("event_time", col("value.timestamp").cast("timestamp"))
)

# hour watermark: accept late data up to 48 hours after window end
windowed = (
    stream
    .withWatermark("event_time", "48 hours")
    .groupBy(
        window(col("event_time"), "1 hour"),  # 1-hour tumbling windows
        col("sensor_id"),
        col("region")
    )
    .agg(_sum("reading").alias("hourly_reading"))
)

# Write to Delta with merge (upsert) for late data correction
query = (
    windowed.writeStream
    .outputMode("append")  # With watermark, can use append
    .foreachBatch(lambda batch, epoch_id: upsert_to_delta(batch, epoch_id))
    .start()
)

def upsert_to_delta(batch_df, epoch_id):
    """Merge late-arriving data, correcting existing aggregations."""
    from delta.tables import DeltaTable
    target = DeltaTable.forPath(spark, "s3://curated/iot_hourly/")

    target.alias("target").merge(
        batch_df.alias("source"),
        "target.window = source.window AND target.sensor_id = source.sensor_id"
    ).whenMatchedUpdateAll() \
     .whenNotMatchedInsertAll() \
     .execute()

# Option B: Lambda Architecture (for simpler ops)
# Batch layer: reprocesses full history daily (always correct)
# Speed layer: real-time approximation (may be wrong until batch corrects)
# Serving layer: query router that blends both

# Trade-off discussion:
# Kappa (streaming only): simpler, but late data handling complex
# Lambda: reliable but dual codepath maintenance burden
# Recommendation: Kappa + Delta merge for most cases

Scenario 6: Handle PII Data Compliantly (GDPR)

The Ask: "A user submits a GDPR 'right to erasure' request. They appear in 50 tables across your data lake. How do you handle this in 30 days?"

Strong Answer:

Architecture: PII-aware from Day 1

1. PII Registry (built proactively):
   - Catalog every table/column containing PII
   - Tag in data catalog: customer_id in [orders, clicks, returns, sessions...]
   - Maintain reverse map: email → all records

2. Erasure Pipeline:
   a. Receive request → log in compliance system with deadline
   b. Query PII registry for all locations of user_id
   c. For each table:

      - If structured (Parquet/Delta): row-level delete (Delta supports this)
      - If archived S3: rewrite Parquet file without user rows
      - If Kafka: mark offset range for compaction
   d. For analytics aggregates:

      - If user contributed to aggregate, recalculate without them
      - If aggregate is anonymized (cohort of >5 users), may be exempt
   e. For ML training data:

      - If model was trained on this user's data: flag for retraining
      - Some models support "machine unlearning" techniques
   f. Generate compliance audit trail

3. Verification:
   SELECT COUNT(*) FROM each_table WHERE user_id = 'erased_user_id'
   -- Must return 0 across all 50 tables

Prevention:

- Use surrogate keys in analytics, not email/name
- Pseudonymization: store mapping outside main lake (deletable separately)
- Avoid storing raw PII in aggregations
- Data minimization: only collect what you need

Category 3: AI Integration Scenarios

Scenario 7: Build an Intelligent Data Catalog

The Ask: "Our data engineers spend 2 hours per week finding the right tables. Build an AI-powered catalog that makes discovery instant."

Strong Answer:

# Core Components:

# 1. Metadata Harvesting Agent
class CatalogHarvestingAgent:
    def harvest_table(self, table_name: str) -> dict:
        """Automatically enrich table metadata using AI."""

        # Get raw metadata
        schema = self.get_schema(table_name)
        sample_data = self.get_sample(table_name, rows=50)
        lineage = self.get_lineage(table_name)
        query_history = self.get_query_history(table_name, days=30)

        # AI enrichment
        description = claude.messages.create(
            model="claude-3-5-haiku-20241022",
            messages=[{
                "role": "user",
                "content": f"""Generate a business-friendly description for this table.
Schema: {schema}
Sample Data: {sample_data.to_markdown()}
Most common queries: {query_history}

Return JSON: {{
  "description": "2-3 sentence business description",
  "use_cases": ["use case 1", "use case 2", "use case 3"],
  "tags": ["tag1", "tag2"],
  "sensitivity": "public|internal|confidential|restricted",
  "data_domain": "finance|marketing|operations|product"
}}"""
            }]
        )

        return json.loads(description.content[0].text)

# 2. Semantic Search Index
def build_search_index(catalog: list[dict]):
    """Embed all catalog entries for semantic search."""
    embeddings_model = BedrockEmbeddings(model_id="amazon.titan-embed-text-v2:0")

    documents = []
    for entry in catalog:
        # Rich text representation for embedding
        text = f"""
        {entry['table_name']}: {entry['description']}
        Use cases: {', '.join(entry['use_cases'])}
        Domain: {entry['data_domain']}
        Tags: {', '.join(entry['tags'])}
        Columns: {', '.join(entry['column_descriptions'])}
        """
        documents.append({"text": text, "metadata": entry})

    # Upsert to OpenSearch/Pinecone
    return vector_store.upsert(documents)

# 3. Conversational Catalog Interface
def catalog_chat(question: str, user_context: dict) -> dict:
    results = semantic_search(question, top_k=5)

    response = claude.messages.create(
        model="claude-3-5-sonnet-20241022",
        messages=[{
            "role": "user",
            "content": f"""A data engineer asked: "{question}"

Relevant tables found:
{json.dumps(results, indent=2)}

User context: {json.dumps(user_context)}

Recommend the best table(s) to use and explain why. 
Include: table name, why it fits, any caveats, example query to get started."""
        }]
    )
    return {"recommendation": response.content[0].text, "tables": results}

Scenario 8: Production LLM Cost Optimization

The Ask: "Your team spent $50,000 on LLM API calls last month. The CFO is asking questions. How do you optimize?"

What's Being Tested: Cost engineering, system design, practical AI operations

Strong Answer:

Audit First (Week 1):

- Log every LLM call: model, input tokens, output tokens, use case, latency
- Identify top 20% of calls generating 80% of cost
- Categorize: classification tasks? Generation? Question answering?

Optimization Strategies:

1. Model Right-sizing (fastest win)
   - Classification tasks → Use Haiku/GPT-3.5 instead of Sonnet/GPT-4
   - Simple extraction → fine-tuned smaller model (80% cheaper)
   - Complex reasoning → keep using Sonnet/GPT-4
   - Typical savings: 40-60%

2. Prompt Optimization
   - Reduce input tokens: trim system prompt, use shorter examples
   - Cache static context (schema descriptions, instructions)
   - Typical savings: 20-30%

3. Response Caching
   - Cache identical or near-identical queries (Redis, 1h TTL)
   - For text-to-SQL: hash(question + schema) → cache SQL
   - Typical savings: 15-25% for repetitive workloads

4. Batch Processing
   - Instead of 1000 individual API calls → batch of 1000 (Anthropic Batch API: 50% discount)
   - For non-real-time enrichment jobs

5. Implement Semantic Caching
   - If new question is semantically similar to cached (cosine sim > 0.95) → return cached
   - LangChain has built-in semantic cache
   - Typical savings: 30-50% for common domains

6. Self-hosted models for sensitive/high-volume
   - Deploy Llama 3.1 70B on GPU instances for internal data
   - Break-even vs. API: typically at ~$8,000/month API spend

ROI Tracking:

- Cost per query by use case
- Value generated per dollar (revenue attributed to AI insights)
- Human hours saved × burdened rate vs. API cost

Scenario 9: AI Feature Engineering at Scale

The Ask: "We want to add 50 new ML features derived from customer transaction history to our churn prediction model. These need to be available for both training and real-time inference."

Strong Answer:

# Feature Store Architecture with Feast

# 1. Define feature groups
from feast import FeatureView, Entity, Field, FileSource, PushSource
from feast.types import Float64, Int64, String, Bool
from datetime import timedelta

customer = Entity(name="customer_id", join_keys=["customer_id"])

# Offline source (batch computation)
customer_transaction_source = FileSource(
    path="s3://feature-store/customer_transaction_features/",
    timestamp_field="feature_timestamp",
    created_timestamp_column="created_timestamp"
)

# Online source (pushed in real-time after each transaction)
customer_realtime_source = PushSource(
    name="customer_realtime_push",
    batch_source=customer_transaction_source
)

customer_behavior_fv = FeatureView(
    name="customer_behavior_v2",
    entities=[customer],
    ttl=timedelta(days=7),
    schema=[
        # Recency features
        Field(name="days_since_last_purchase", dtype=Int64),
        Field(name="days_since_last_login", dtype=Int64),

        # Frequency features  
        Field(name="purchases_last_7d", dtype=Int64),
        Field(name="purchases_last_30d", dtype=Int64),
        Field(name="purchases_last_90d", dtype=Int64),
        Field(name="sessions_last_7d", dtype=Int64),

        # Monetary features
        Field(name="avg_order_value_30d", dtype=Float64),
        Field(name="total_spend_90d", dtype=Float64),
        Field(name="max_single_order_value", dtype=Float64),
        Field(name="refund_rate_90d", dtype=Float64),

        # Behavioral features
        Field(name="preferred_category", dtype=String),
        Field(name="mobile_vs_desktop_ratio", dtype=Float64),
        Field(name="support_tickets_30d", dtype=Int64),
        Field(name="promo_usage_rate", dtype=Float64),

        # Derived AI features (LLM-generated from reviews)
        Field(name="review_sentiment_score", dtype=Float64),
        Field(name="product_satisfaction_embedding_dim_1", dtype=Float64),
        # ... more embedding dimensions
    ],
    source=customer_realtime_source,
    online=True  # Available for real-time serving
)

# 2. Compute features in batch (daily)
def compute_customer_features(spark, date: str) -> DataFrame:
    orders = spark.read.parquet(f"s3://raw/orders/date={date}/")
    sessions = spark.read.parquet(f"s3://raw/sessions/date={date}/")
    reviews = spark.read.parquet(f"s3://raw/reviews/date={date}/")

    features = orders.groupBy("customer_id").agg(
        datediff(lit(date), max("order_date")).alias("days_since_last_purchase"),
        count(when(col("order_date") >= date_sub(lit(date), 30), True)).alias("purchases_last_30d"),
        avg(when(col("order_date") >= date_sub(lit(date), 30), col("revenue"))).alias("avg_order_value_30d"),
        # ... all other features
    )

    return features

# 3. Real-time push after each transaction
def push_realtime_features(transaction: dict):
    """Called by transaction service after each completed order."""
    store = FeatureStore(repo_path="./feature_repo")

    # Compute delta features (incremental update)
    updated_features = compute_incremental_features(transaction)

    store.push("customer_realtime_push", 
               pd.DataFrame([updated_features]),
               to=PushMode.ONLINE)  # Only update online store

Category 4: Performance & Scale Scenarios

Scenario 10: Debug a Slow Spark Job

The Ask: "A Spark job that used to run in 20 minutes now takes 4 hours. Production data is backed up. Debug it live."

Systematic Debugging Approach:

Step 1: Check Spark UI immediately

- Stage view: which stage is slow?
- If stage has 1 task doing all work → Data Skew
- If all tasks slow → Resource constraint or shuffle issue

Step 2: Data Skew Detection
SELECT partition_value, COUNT(*) as record_count
FROM source_table  
GROUP BY partition_value
ORDER BY 2 DESC LIMIT 20;

If top value has 1000x more rows than median → Skew

Fix: Salting
from pyspark.sql.functions import concat, lit, (rand()*100).cast("int")

# Add salt to skewed key
df_salted = df.withColumn(
    "salted_key", 
    concat(col("customer_id"), lit("_"), (rand()*100).cast("int").cast("string"))
)
# Join on salted key with expanded lookup
lookup_expanded = lookup.crossJoin(
    spark.range(100).withColumnRenamed("id", "salt")
).withColumn("salted_key", concat(col("customer_id"), lit("_"), col("salt").cast("string")))

Step 3: Check for shuffle

- Look for "Exchange" nodes in explain plan
- If unexpected shuffle: check join order, join type

explain_plan = df.explain(mode="extended")
# Look for: SortMergeJoin (bad for small tables) vs BroadcastHashJoin (good)

Fix: Force broadcast for small tables
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "512mb")

Step 4: Check input data size

- Did source data grow 10x? → Expected slowdown
- Did partition count change? → Adjust spark.sql.shuffle.partitions

Step 5: Check resource utilization

- Executor OOM → Increase memory or reduce parallelism
- Low CPU utilization → Network bottleneck (data not co-located)

Scenario 11: Design a Multi-Region Data Architecture

The Ask: "We're expanding to Europe and must comply with GDPR data residency — EU customer data cannot leave EU. But we need global analytics. How?"

Strong Answer:

Architecture: Regional Isolation + Federated Analytics

Regional Data Planes:
├── US Region (us-east-1)
│   ├── US customer data (stays in US)
│   ├── US analytics warehouse
│   └── US ML training
├── EU Region (eu-west-1)  
│   ├── EU customer data (stays in EU, GDPR compliant)
│   ├── EU analytics warehouse
│   └── EU ML training
└── APAC Region (ap-southeast-1)
    └── APAC data (stays in APAC)

Global Analytics Layer (no PII, only aggregates):
├── Anonymized metrics flow to global warehouse
│   - Revenue by region (no customer-level detail)
│   - Aggregate cohort metrics
│   - Product performance
└── Cross-region queries via federated query engine (Trino/Athena)

Data Classification:

- PII (name, email, address) → stays in origin region
- Behavioral aggregates (purchase count, avg value) → can cross regions
- Product/inventory data → global by default

Implementation:

1. Tag every field: pii_level = [none|low|medium|high]
2. Automated check: any cross-region job touching PII → blocked
3. Anonymization service: strips/masks PII before cross-region transfer
4. Audit log: every cross-region data movement logged

Governance:

- Data residency policy as code (OPA policies)
- Automated compliance reports
- DPIA (Data Protection Impact Assessment) for new data flows

Category 5: AI & LLM-Specific Scenarios

Scenario 12: RAG Pipeline for Internal Knowledge Base

The Ask: "Build a system where data engineers can ask 'How do we calculate customer LTV?' and get an accurate answer from our internal documentation."

Architecture Approach:

The "Why" Behind the Design: A RAG (Retrieval-Augmented Generation) pipeline solves the "knowledge cutoff" problem of LLMs by giving them a highly specific open-book test. Instead of training the LLM on your docs (which is expensive and hard to update), you search your docs for the answer, hand the relevant paragraphs to the LLM, and say "Answer the user's question using ONLY this text."

Real-World Example: A new engineer asks, "How do I deploy an Airflow DAG?"

1. Retrieval: The system searches the vector database and finds a markdown file named deploying_airflow.md created 2 days ago.
2. Augmentation: It grabs the top 3 paragraphs from that file.
3. Generation: The LLM reads those paragraphs and formulates a polite, accurate response: "To deploy an Airflow DAG, you need to push to the main branch, which triggers the CI/CD pipeline as outlined in deploying_airflow.md."

# Production RAG Pipeline
import anthropic
from langchain_community.vectorstores import OpenSearchVectorSearch
from langchain_aws import BedrockEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter

class InternalKnowledgeRAG:
    def __init__(self):
        self.embeddings = BedrockEmbeddings(
            model_id="amazon.titan-embed-text-v2:0"
        )
        self.vectorstore = OpenSearchVectorSearch(
            opensearch_url="https://search-internal-docs.es.amazonaws.com",
            index_name="engineering-docs",
            embedding_function=self.embeddings
        )
        self.claude = anthropic.Anthropic()

    def ingest_document(self, content: str, metadata: dict):
        """Ingest documentation with rich metadata."""
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["\n## ", "\n### ", "\n\n", "\n", " "]
        )
        chunks = splitter.create_documents(
            [content],
            metadatas=[{**metadata, "chunk_index": i}]
            for i, _ in enumerate(splitter.split_text(content))
        )
        self.vectorstore.add_documents(chunks)

    def answer_question(self, question: str, user: str) -> dict:
        # Retrieve relevant chunks
        docs = self.vectorstore.similarity_search_with_score(question, k=6)

        # Filter by relevance threshold
        relevant = [(doc, score) for doc, score in docs if score > 0.75]

        if not relevant:
            return {"answer": "No relevant documentation found.", "sources": []}

        context = "\n\n".join([
            f"[Source: {doc.metadata['source']}]\n{doc.page_content}"
            for doc, _ in relevant
        ])

        response = self.claude.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1000,
            system="""You are a knowledgeable internal assistant for a data engineering team.
Answer questions based ONLY on the provided context.
If the context doesn't contain the answer, say so clearly.
Always cite your sources.""",
            messages=[{
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {question}"
            }]
        )

        return {
            "answer": response.content[0].text,
            "sources": list(set([doc.metadata['source'] for doc, _ in relevant])),
            "confidence": "high" if len(relevant) >= 3 else "medium"
        }

Scenario 13: Detecting AI Model Drift in Production

The Ask: "Our churn prediction model was 89% accurate when deployed. Three months later, it's flagging the wrong customers. How do you detect and respond?"

Strong Answer:

# Model monitoring pipeline
import boto3
import numpy as np
from scipy import stats

class ModelDriftDetector:
    def __init__(self, model_name: str, baseline_window_days: int = 30):
        self.model_name = model_name
        self.baseline = self.load_baseline_distribution(baseline_window_days)

    def check_for_drift(self, recent_predictions: pd.DataFrame) -> dict:
        drift_report = {}

        # 1. Prediction Distribution Drift (PSI - Population Stability Index)
        psi = self.calculate_psi(
            expected=self.baseline['prediction_scores'],
            actual=recent_predictions['churn_probability']
        )
        drift_report['psi'] = psi
        drift_report['prediction_drift'] = 'HIGH' if psi > 0.2 else 'LOW' if psi < 0.1 else 'MEDIUM'

        # 2. Feature Drift (KS Test per feature)
        feature_drift = {}
        for feature in self.baseline['features']:
            ks_stat, p_value = stats.ks_2samp(
                self.baseline['feature_distributions'][feature],
                recent_predictions[feature].dropna()
            )
            feature_drift[feature] = {
                'ks_statistic': ks_stat,
                'p_value': p_value,
                'drifted': p_value < 0.05
            }
        drift_report['feature_drift'] = feature_drift

        # 3. Outcome Drift (if we have labels)
        if 'actual_churn' in recent_predictions:
            actual_rate = recent_predictions['actual_churn'].mean()
            baseline_rate = self.baseline['churn_rate']
            drift_report['churn_rate_change'] = abs(actual_rate - baseline_rate) / baseline_rate

        # 4. Business Impact Metrics
        drift_report['precision'] = self.calculate_precision(recent_predictions)
        drift_report['recall'] = self.calculate_recall(recent_predictions)

        return drift_report

    def calculate_psi(self, expected, actual, bins=10):
        """Population Stability Index - >0.2 indicates significant drift."""
        breakpoints = np.percentile(expected, np.arange(0, 110, 10))
        expected_pct = np.histogram(expected, breakpoints)[0] / len(expected)
        actual_pct = np.histogram(actual, breakpoints)[0] / len(actual)

        # Avoid log(0)
        expected_pct = np.where(expected_pct == 0, 0.0001, expected_pct)
        actual_pct = np.where(actual_pct == 0, 0.0001, actual_pct)

        return np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))

# Response Playbook:
# PSI < 0.1: Normal, no action
# 0.1 ≤ PSI < 0.2: Monitor closely, investigate features
# PSI ≥ 0.2: Trigger retraining pipeline, alert ML team
# Precision drops >5%: Emergency retraining + rollback plan

Category 6: Culture & Process Scenarios

Scenario 14: Building Data Engineering Culture

The Ask: "You're the first data engineer at a startup. There's no infrastructure, no processes, no data culture. Where do you start?"

Strong Framework:

Month 1: Foundation & Trust

- Don't build anything. Learn the business first.
- Identify the 3 decisions leadership makes most often
- Find where data currently lives (spreadsheets, SaaS dashboards)
- Build ONE reliable metric: the company's North Star metric
- Get it into a simple dashboard. Make leadership trust data.

Month 2-3: Core Infrastructure

- Cloud: AWS/GCP/Azure data warehouse (Redshift/BigQuery/Synapse)
- ELT: Fivetran for SaaS sources + dbt for transformations
- BI: Metabase or Looker for self-service
- Simple Airflow for batch jobs
- Principle: boring technology that just works

Month 4-6: Data Culture

- Weekly "Data Office Hours" — teach non-engineers to query
- Data Dictionary: document the 20 most important metrics
- SLA: commit to "revenue data available by 8am"
- Quality Dashboard: public-facing data health metrics

Month 7-12: Scale

- Based on business priorities, build the next most valuable thing
- Avoid premature optimization
- Hire carefully: first data engineer hire is the most important

Principles to never compromise:

1. Reliability > Features (a wrong number is worse than no number)
2. Document decisions (build an ADR — Architecture Decision Record)
3. Make data accessible (hoarding data creates political problems)
4. Measure the value your work creates (not just the work itself)

Scenario 15: AI Ethics in Data Engineering

The Ask: "Your ML model predicts loan default risk. An audit finds it has an 18% higher false-positive rate for certain demographic groups. What do you do?"

Strong Answer:

Immediate Actions (Day 1):

1. Pause the model in production — don't let discriminatory outcomes continue
2. Escalate to legal, compliance, and executive leadership immediately
3. Document everything: when discovered, who was informed, what was done

Investigation (Week 1-2):

- Audit training data for historical bias:
  * Was training data from a period of discriminatory lending?
  * Are protected attributes (race, gender) proxied by zip code, school name?
  * Is there differential data quality for demographic groups?

- Bias metrics to calculate:
  * Demographic Parity: P(approve | group A) ≈ P(approve | group B)
  * Equal Opportunity: P(approve | creditworthy, group A) ≈ P(approve | creditworthy, group B)
  * Individual Fairness: similar individuals should be treated similarly

Technical Fixes:

- Re-examine feature selection: remove proxy features
- Reweighting: assign higher weight to underrepresented groups in training
- Constraint optimization: add fairness constraints to loss function
- Post-processing: threshold adjustment by demographic group (check legality)
- Regular fairness audits: scheduled bias testing in monitoring pipeline

Governance:

- Fairness metrics added to model card
- Mandatory bias audit before any model deployment
- External audit by third party for high-stakes decisions
- Data collection improvements to reduce representation gaps

Long-term:

- Build diverse data collection practices
- Bias testing as part of CI/CD pipeline
- Training for all ML engineers on fairness-aware ML

Quick-Fire Interview Questions

Q: Explain the difference between a data lake and a data warehouse.

A data lake stores raw, unstructured/semi-structured data at massive scale cheaply (schema-on-read). A data warehouse stores structured, cleaned, modeled data optimized for analytical queries (schema-on-write). Modern lakehouses combine both using open table formats.

Q: What is the CAP theorem and how does it apply to your data systems?

In distributed systems, you can only guarantee 2 of 3: Consistency, Availability, Partition tolerance. In data engineering: Kafka sacrifices consistency for availability+partition (eventual consistency). RDBMS sacrifices availability for consistency+partition.

Q: When would you use Kafka vs Kinesis?

Kafka: multi-cloud, on-prem, long retention (weeks), complex routing, ecosystem integrations. Kinesis: AWS-native, simpler ops, managed service, tightly integrated with AWS (Lambda, EMR, Glue).

Q: What is data lineage and why does it matter?

Lineage tracks the origin, transformations, and destinations of data — the full journey from source to consumer. Critical for: debugging wrong reports, GDPR compliance (data origin auditing), impact analysis (if I change table X, what breaks?).

Q: How do you handle schema evolution in a production pipeline?

Use Avro/Protobuf schemas with a Schema Registry (Confluent) for Kafka. Use Iceberg/Delta for lakes (supports add column, rename, type evolution). Use dbt on_schema_change: 'sync_all_columns' for transformations. Principle: backwards-compatible changes (add columns) are safe; breaking changes (remove/rename) need versioning.

Q: What is the difference between a star schema and a snowflake schema?

Star: dimension tables directly connected to fact table (denormalized, faster queries). Snowflake: dimensions further normalized into sub-dimensions (less storage, more joins). Star preferred for analytics (query speed > storage). Snowflake used when storage is constrained.

Q: How would you implement idempotent data pipelines?

Each pipeline run with same input produces same output without side effects. Techniques: write to temp location, validate, atomic rename/swap; use INSERT OVERWRITE vs INSERT (replaces, not appends); include run_id in outputs for deduplication; use date-partitioned paths that overwrite the partition.

Q: What is the difference between hot, warm, and cold data storage?

Hot: frequently accessed, low latency, high cost (Redis, DynamoDB, in-memory). Warm: regularly accessed, moderate latency/cost (S3 Standard, data warehouse). Cold: rarely accessed, high latency, very low cost (S3 Glacier, tape). Tiering saves significant cost.

These scenarios represent the kinds of problems you'll encounter at senior data engineering interviews at companies like Netflix, Airbnb, Databricks, AWS, Google, and well-funded startups. The key is not memorizing answers but deeply understanding the why behind each architectural decision.

This content is reserved for Premium Members.