Data Engineering & Batch Pipelines: exercises

Advanced Exercise 1: Design the ShopSense training data pipeline from requirements

(Paid | XP: 40)

ShopSense is building a new batch pipeline to produce the training dataset for its recommendation model v5. The training pipeline will:

• Ingest user_events (180M events/day), product_catalog (50M SKUs), and user_profiles (15M users)
• Compute 12 derived features (rolling engagement windows, category preferences, session statistics)
• Produce a training dataset: one row per (user_id, candidate_item_id) pair, with 12 features and a binary label (did the user purchase this item within 24h?)
• Write to S3 in Parquet format, partitioned by training_date
• Run daily at 02:00 UTC

Part A (Requirements specification): Fill in the requirements-first checklist for this pipeline. For each item, provide a concrete answer (not a generic description):

1. Consumers and SLA: Who consumes this pipeline's output? What is the maximum acceptable staleness? What happens if the pipeline fails on a Monday: can training be delayed 24 hours, or must it be blocked entirely?

2. Latency requirement: Is daily batch acceptable, or does the training frequency need to change if user behavior patterns can shift significantly within 24 hours? (Hint: how much does user behavior change on Black Friday vs. the day before?)

3. Correctness model: For the feature user_purchase_count_7d, should this be computed in event-time or processing-time? What is the late-arrival handling strategy for mobile purchase events (which arrive up to 72 hours late)?

4. Data volume and skew: What is the expected training dataset row count? (Estimate: 15M users × top-K candidate items per user.) What partition skew risk exists if the dataset is partitioned by user_id prefix?

5. Failure tolerance: Define RTO and RPO. Does this pipeline need a DLQ? What happens to malformed user_events records (e.g., records where event_timestamp is in the future)?

6. Security: Which source tables contain PII? Which features in the training dataset are PII-derived? What is the retention schedule for training dataset partitions?

Part B (Pipeline architecture): Design the DAG for this pipeline with tasks, dependencies, and expected execution times. Include:

• At least one DQ validation gate before the feature computation stage
• An idempotency mechanism for the feature computation outputs
• A data versioning step that writes a version manifest (schema version, content hash, row count, lineage) after successful completion
• An alerting strategy: which task failures send immediate pages vs. which failures escalate after 2 retries?

Draw the task dependency graph (can be ASCII).

Part C (Idempotency design): The feature computation task reads from user_events/date=2024-11-15/ and writes to training_features/date=2024-11-15/. Write the Python function signature and implementation approach for write_features_idempotently() that:

• Accepts a date string, input path, output path, and a computed Spark DataFrame
• Deletes any existing output for that date partition before writing
• Writes the DataFrame to Parquet
• Writes a _SUCCESS marker file with row count and content hash
• Returns the content hash

Part D (SLA monitoring): Define the three most critical SLA metrics to monitor for this pipeline:

1. A freshness metric (measures how old the training data is at training time)
2. A volume metric (detects data loss or duplication at ingestion)
3. A quality metric (detects upstream schema drift before training is triggered)

For each, define: metric name, formula, SLA threshold, and action on breach.

---

Advanced Exercise 2: Time semantics debugging

(Paid | XP: 40)

ShopSense's recommendation model offline evaluation shows recall@10 = 0.091 on the validation set. When deployed to production, the 7-day running CTR is 2.1%, significantly below the expected 2.6% based on offline metrics.

You suspect training-serving skew related to time semantics. You have access to:

• Training dataset: user_purchase_velocity_7d feature computed for all users as of 2024-11-15
• Serving feature store: user_purchase_velocity_7d feature for the same users as of 2024-11-17 (2 days after training cutoff)
• Feature computation code:

# Training pipeline (batch, runs at 02:00 UTC)
def compute_purchase_velocity_training(events_df, training_date):
    cutoff = pd.Timestamp(training_date)
    window_start = cutoff - pd.Timedelta(days=7)
    return (
        events_df[
            (events_df['event_timestamp'] >= window_start) &
            (events_df['event_timestamp'] < cutoff) &
            (events_df['event_type'] == 'purchase')
        ]
        .groupby('user_id')
        .size()
        .rename('velocity_7d')
        .reset_index()
    )

# Serving feature computation (online, runs per-request)
def compute_purchase_velocity_serving(user_id, redis_client):
    raw_events = redis_client.lrange(f"user:{user_id}:events", 0, -1)
    # Redis list contains events from the past 7 days, updated via streaming ingestion
    purchase_events = [e for e in raw_events if json.loads(e)['event_type'] == 'purchase']
    return len(purchase_events)

Part A (Identify the time semantics bug): The training pipeline uses event_timestamp for the window (event time). The serving pipeline reads from a Redis list that contains "events from the past 7 days." Are these semantically equivalent? What is the difference, and how would it manifest as a distribution shift between training and serving?

Part B (PSI analysis): You compute the PSI (Population Stability Index) for user_purchase_velocity_7d between training (2024-11-15) and serving (2024-11-17) distributions. The PSI is 0.23. Based on the standard thresholds (< 0.10: no action, 0.10–0.20: investigate, > 0.20: significant shift), what should happen next?

You also compute the feature distribution statistics:

Cohort	Training (N=15M)	Serving (N=12M)
velocity = 0	42%	51%
velocity = 1	31%	27%
velocity = 2–5	20%	16%
velocity ≥ 6	7%	6%

What pattern do you see? Form a hypothesis about what changed between training (Nov 15) and serving (Nov 17).

Part C (Root cause and fix): The training window is event-timestamp-based with a hard cutoff at 2024-11-15 00:00:00 UTC. The serving Redis list is populated by a streaming job that processes events as they arrive, using ingestion time as the window key.

1. Diagram the discrepancy: a mobile user purchased on Nov 14 at 10pm but the purchase event arrived (was ingested) on Nov 15 at 6am. In training, does this event fall inside the window? In serving (Redis), does it fall inside the window?

2. At scale, what percentage of purchases fall into this "event time in window, ingestion time outside window" category? Use ShopSense's mobile delay statistics (median: 47 min, p95: 8h, max: 72h).

3. Write the fix for the serving feature computation that aligns it with the training pipeline's event-time semantics. Note: you cannot change the training pipeline's logic; it is already correct.

Part D (Prevention): Write a pytest function test_training_serving_velocity_semantic_alignment() that:

• Generates a synthetic event with event_timestamp = T–1h and ingestion_timestamp = T+6h
• Computes purchase_velocity_7d using both the training function and the serving function
• Asserts that both return the same value (they should both count or both not count this event)
• Fails with a descriptive message if there is a discrepancy

---

Advanced Exercise 3: DQ validation suite design

(Paid | XP: 35)

ShopSense's ML team has experienced three data quality incidents in the past quarter:

Incident 1 (October): The product_catalog table had a batch ingestion failure that resulted in a 40% reduction in row count. Training proceeded on the reduced dataset, producing a model that couldn't recommend ~20M SKUs. Detected in production after 4 days.

Incident 2 (October): A vendor API change caused product_category to return ISO codes instead of display names (the product_category → category_primary issue from Ch 3.2). The training pipeline ran successfully with the new format. The trained model degraded 22% in CTR. Detected after 10 days.

Incident 3 (November): The user_events table had a timezone configuration change in the ingestion pipeline that shifted all event_timestamp values by +8 hours. Training features computed on wrong timestamps. Model trained on data from the wrong temporal window. Detected after 6 days.

Your task: Design a Great Expectations validation suite that would have caught each of these three incidents before training was triggered.

Part A, Incident 1 (row count drop): Write three GE expectations that would have caught the 40% row count drop in product_catalog. Include:

• An absolute minimum row count expectation
• A relative row count expectation (today's count vs. yesterday's count, flagging if the change exceeds a threshold)
• A coverage expectation (the catalog covers at least X% of the product_ids that appeared in yesterday's user_events)

Part B, Incident 2 (schema/format drift): Write two GE expectations that would have caught the product_category format change from display names ("Electronics") to ISO codes ("ELECTRONICS_COMPUTING"):

• An expectation on the value set (or value pattern) for product_category
• An expectation that catches high-cardinality inflation (47 values → 200+ ISO codes)

Part C, Incident 3 (timestamp timezone shift): Write two GE expectations that would have caught the +8h timezone shift in event_timestamp:

• An expectation on the range of event_timestamp values for a given partition date (events for date=2024-11-15 should have timestamps in [2024-11-15 00:00, 2024-11-15 23:59], not shifted into Nov 16)
• An expectation on the median event_timestamp hour-of-day (events should be roughly uniformly distributed across the day, not shifted toward the evening)

Part D (Pipeline integration): Write the Airflow operator call and task dependency structure that runs this validation suite as a blocking gate before the compute_features task. The validation task should:

• Run after both user_events and product_catalog are available
• Fail loudly (raise an exception, not just log a warning) if any critical expectation fails
• Write Data Docs to S3 for every run (whether pass or fail)
• Send a Slack alert with the failed expectation summary if validation fails

---

Real-World Implementation 1: Idempotent Airflow DAG with DQ gates

(Paid | XP: 45)

Implement a production-ready Airflow DAG for ShopSense's daily training data pipeline with idempotency guarantees, DQ validation gates, and DVC data versioning.

"""
shopsense_training_data_dag.py

Daily pipeline: raw user_events → validated → feature computation → versioned training dataset
"""
from __future__ import annotations

import hashlib
import json
import logging
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any

import boto3
import great_expectations as gx
import pandas as pd
from airflow import DAG
from airflow.operators.python import PythonOperator  # Airflow 3: airflow.providers.standard.operators.python

S3_BUCKET = "shopsense-training-data"
RAW_PREFIX = "raw/user_events"
FEATURES_PREFIX = "features/user_engagement"
TRAINING_PREFIX = "training/recommendation_v5"

logger = logging.getLogger(__name__)


def validate_raw_events(partition_date: str, **context) -> bool:
    """
    Run Great Expectations validation suite on user_events for partition_date.
    Returns True on success, raises PipelineDataQualityError on failure.

    The suite must check:
    - All required columns present (user_id, event_type, event_timestamp, product_id, session_id)
    - Row count in [150M, 220M]
    - null rate of user_id = 0
    - null rate of event_timestamp = 0
    - event_type in allowed value set
    - max(ingestion_timestamp) within 6 hours of partition_date end

    Push validation result to XCom for downstream logging.
    """
    ...


def compute_engagement_features(partition_date: str, **context) -> dict[str, Any]:
    """
    Compute user engagement features for partition_date using IDEMPOTENT writes.

    Features to compute:
    - user_session_count_7d: sessions in trailing 7 days (event time)
    - user_purchase_velocity_7d: purchases in trailing 7 days (event time)
    - user_click_rate_7d: clicks / views in trailing 7 days
    - user_search_rate_7d: searches per session in trailing 7 days

    Write pattern:
    1. Delete any existing data at output_path/date={partition_date}/
    2. Compute features using PySpark (or pandas for < 500M rows)
    3. Write Parquet with overwrite mode to output_path/date={partition_date}/
    4. Write _SUCCESS marker with row count + content hash

    Return: dict with output_path, row_count, content_hash
    """
    ...


def version_training_dataset(partition_date: str, **context) -> str:
    """
    Create a versioned training dataset snapshot using DVC.

    Steps:
    1. Pull the computed features for partition_date from S3
    2. Run `dvc add` on the local copy to generate a .dvc pointer file
    3. Write a version manifest YAML containing:
       - version_id: f"{partition_date}_{content_hash[:8]}"
       - partition_date
       - feature_schema_version: from schema registry
       - source_table_versions: {user_events: schema_version, product_catalog: schema_version}
       - content_hash
       - row_count
       - dvc_pointer_path
    4. Run `dvc push` to sync to S3 remote
    5. Return the version_id

    Push version_id to XCom.
    """
    ...


def emit_lineage_events(partition_date: str, **context) -> None:
    """
    Emit OpenLineage events documenting the pipeline run.

    Emit:
    - A RunEvent with eventType=COMPLETE for the feature computation job
    - Input dataset: user_events (with schema version from context)
    - Output dataset: user_engagement_features (with column lineage)
    - Column lineage: user_session_count_7d ← [user_id, event_timestamp, session_id]
    - job_name: "shopsense.compute_engagement_features"
    - run_id: from context['run_id']
    """
    ...


# DAG definition
with DAG(
    dag_id="shopsense_training_data_pipeline",
    schedule="0 2 * * *",
    start_date=datetime(2024, 1, 1),
    catchup=True,                 # enables backfill
    max_active_runs=1,            # prevent concurrent runs for same date
    default_args={
        "retries": 2,
        "retry_delay": timedelta(minutes=15),
        "email_on_failure": True,
        "email": ["ml-oncall@shopsense.com"],
    },
    tags=["training-data", "recommendation", "daily"],
) as dag:

    validate = PythonOperator(
        task_id="validate_raw_events",
        python_callable=validate_raw_events,
        op_kwargs={"partition_date": "{{ ds }}"},
    )

    compute = PythonOperator(
        task_id="compute_engagement_features",
        python_callable=compute_engagement_features,
        op_kwargs={"partition_date": "{{ ds }}"},
    )

    version = PythonOperator(
        task_id="version_training_dataset",
        python_callable=version_training_dataset,
        op_kwargs={"partition_date": "{{ ds }}"},
    )

    lineage = PythonOperator(
        task_id="emit_lineage_events",
        python_callable=emit_lineage_events,
        op_kwargs={"partition_date": "{{ ds }}"},
    )

    validate >> compute >> version >> lineage

Your task:

1. Implement validate_raw_events(): Use Great Expectations to validate user_events for the given partition_date. Load the appropriate partition from S3 (path: s3://{S3_BUCKET}/{RAW_PREFIX}/date={partition_date}/). Run the expectation suite (define it inline or load from a YAML file). Raise PipelineDataQualityError with a formatted message if validation fails. Push validation statistics to XCom.

2. Implement compute_engagement_features(): Compute the four features listed. Use an idempotent write pattern: delete the existing date={partition_date} partition from S3 before writing. Write a _SUCCESS marker file containing JSON with {"row_count": ..., "content_hash": ..., "computed_at": ...}. Return the output metadata dict.

3. Implement version_training_dataset(): Generate the version manifest YAML and run DVC commands (dvc add, dvc push) via subprocess.run. The manifest must include all fields listed. Push the version_id to Airflow XCom.

4. Implement emit_lineage_events(): Construct and emit an OpenLineage RunEvent to Marquez (use http://localhost:5000/api/v1/lineage as endpoint; skip if Marquez unavailable). The event must include input/output dataset names, column lineage for at least one feature column, and the run_id from the Airflow context.

5. Write an idempotency test test_compute_engagement_features_idempotency() that runs compute_engagement_features() twice on the same execution date with a synthetic fixture, and asserts that the _SUCCESS marker's content_hash is identical between the two runs.

---

Real-World Implementation 2: Great Expectations custom expectation for temporal integrity

(Paid | XP: 35)

Implement a custom Great Expectations Expectation that validates temporal integrity in the user_events dataset: events in the partition for date=2024-11-15 should have event_timestamp values within [2024-11-15 00:00:00 UTC, 2024-11-16 06:00:00 UTC], allowing for up to 6 hours of late-arrival, but flagging any events with timestamps that are implausibly far from the partition date (indicating a timezone bug or pipeline error).

Note: the scaffold below targets the Great Expectations 1.x custom-expectation API (verified against GX 1.23): parameters are class fields, and a row-level check is a ColumnMapExpectation backed by a ColumnMapMetricProvider, so GX itself reports the affected row count.

"""
expect_timestamps_within_partition_window.py

Custom Great Expectations (1.x) Expectation that validates timestamp integrity.

Usage:
    start, end = compute_window("2024-11-15", allow_early_hours=0, allow_late_hours=6)
    expectation = ExpectTimestampsWithinPartitionWindow(
        column="event_timestamp",
        window_start=start.isoformat(),   # 2024-11-15T00:00:00+00:00
        window_end=end.isoformat(),       # 2024-11-16T06:00:00+00:00
    )
    result = batch.validate(expectation)   # result.result has unexpected_count, etc.
"""

from datetime import datetime, timedelta, timezone

import great_expectations as gx
import pandas as pd
from great_expectations.execution_engine import PandasExecutionEngine
from great_expectations.expectations.expectation import ColumnMapExpectation
from great_expectations.expectations.metrics.map_metric_provider import (
    ColumnMapMetricProvider,
    column_condition_partial,
)


def compute_window(
    partition_date: str,
    allow_early_hours: int = 0,
    allow_late_hours: int = 6,
) -> tuple[datetime, datetime]:
    """
    Compute the valid timestamp window for a given partition date.

    window_start = partition_date midnight UTC - allow_early_hours
    window_end = (partition_date + 1 day) midnight UTC + allow_late_hours
    """
    ...


class ColumnValuesWithinPartitionWindow(ColumnMapMetricProvider):
    """Row-level condition: True where the timestamp is inside the window."""

    condition_metric_name = "column_values.within_partition_window"
    condition_value_keys = ("window_start", "window_end")

    @column_condition_partial(engine=PandasExecutionEngine)
    def _pandas(cls, column, window_start, window_end, **kwargs):
        # TODO: return a boolean Series that is True for timestamps in
        # [window_start, window_end]. Parse the two ISO strings with pd.Timestamp
        # and make sure the column is timezone-aware UTC before comparing.
        ...


class ExpectTimestampsWithinPartitionWindow(ColumnMapExpectation):
    """
    Validates that timestamps in a partition fall within the expected time window.

    Catches:
    - Timezone configuration bugs (timestamps shifted by N hours)
    - Events from the wrong date mixed into the partition
    - Future-dated events (clock skew, timestamp generation bugs)

    The result reports unexpected_count, unexpected_percent and a sample of
    partial_unexpected_list values, so the "affected count" comes for free.
    """

    # In GX 1.x, Expectation parameters are declared as (pydantic) class fields.
    window_start: str
    window_end: str

    map_metric = "column_values.within_partition_window"
    success_keys = ("window_start", "window_end", "mostly")
    args_keys = ("column", "window_start", "window_end")

Your task:

1. Implement compute_window(): compute window_start as datetime(partition_year, month, day, tzinfo=utc) - timedelta(hours=allow_early_hours) and window_end as the next day's midnight UTC + timedelta(hours=allow_late_hours). Handle month/year rollover correctly (hint: datetime.fromisoformat plus timedelta(days=1)).

2. Implement ColumnValuesWithinPartitionWindow._pandas(): return a boolean Series that is True for timestamps inside [window_start, window_end]. The expectation inherits unexpected_count, unexpected_percent and partial_unexpected_list from ColumnMapExpectation, so you do not compute counts yourself. Future-dated events fall outside window_end, so they are caught by the same check.

3. Write tests for these cases:

   • All timestamps within the window → pass
   • 0.1% of timestamps shifted +8 hours (timezone bug) → fail, and the result shows the affected count in unexpected_count
   • 1 record with timestamp 2 years in the future → fail with unexpected_count: 1
   • Empty DataFrame → pass (no timestamps means no violations)

4. Register and use the expectation in a GX Expectation Suite (suite.add_expectation(...)), then validate a test DataFrame with batch.validate(suite) as in the data-quality section above.

---

Production Challenge 1: Pipeline cascade failure during Black Friday

(Paid | XP: 40)

Scenario: It is 3:47am on Black Friday. ShopSense's daily training data pipeline has been running since 2:00am. PagerDuty fires:

CRITICAL: shopsense_training_data_pipeline / compute_engagement_features
Status: FAILED after 2 retries
Error: org.apache.spark.SparkException: Job aborted due to stage failure:
  Task 847 in stage 12.0 failed 4 times... java.lang.OutOfMemoryError: GC overhead limit exceeded

Context:

• Black Friday traffic is 2.3× normal (expected: 180M events; actual: 414M events today)
• The Spark cluster is sized for 220M events (10% headroom above normal max)
• The task is partitioned by user_id prefix (a-z), but 31% of ShopSense users have IDs starting with 'a' through 'c' (partition skew from legacy ID generation)
• The training DAG is blocking the recommendation model's weekly retraining, which is scheduled for 6am
• The recommendation model currently in production was trained 8 days ago

Your task:

Part A, Immediate triage (< 30 min). Write the triage runbook entry for this incident. Include:

• How to determine if this is partition skew vs. genuine memory insufficiency
• The Spark configuration change that would mitigate partition skew without requiring cluster resize: what does spark.sql.shuffle.partitions need to be changed to, and why?
• An immediate fix that allows the pipeline to complete for Black Friday without full cluster scaling

Part B (Backfill strategy). The pipeline failed at 3:47am. Assume you fix it and rerun by 5:30am. The model retraining is scheduled at 6:00am. The retraining pipeline requires "training data complete through T-24h" as its input SLA.

Given:

• "Training data complete" means all feature partitions for dates [today-90d, today-1d] are present and have passed DQ validation
• Today-1d = Nov 28 partition (yesterday): this is the partition that failed
• All other partitions (Nov 27 and earlier) were computed successfully on previous days

Write the partial backfill command to compute only the missing Nov 28 partition. What Airflow flag or configuration prevents a backfill from re-running all 90 previous days? (Airflow 2: airflow dags backfill; Airflow 3: airflow backfill create. Hint: bound the date range, and check how each version treats already-successful runs.)

Part C (Capacity planning). The root cause is under-provisioning for peak traffic. Design a capacity monitoring approach that would have predicted this failure:

1. What metric would you monitor to detect partition skew before it causes failures? (Not just "cluster memory usage"; something more specific to partition distribution.)
2. What threshold on this metric would have triggered a pre-emptive cluster resize before Black Friday?
3. Black Friday traffic peaks are predictable (they happen every year). Write a Terraform or Airflow variable configuration that automatically scales the Spark cluster to 2.5× capacity on the 4th Thursday of November (US Thanksgiving / Black Friday period).

Part D (Architecture improvement). The partition skew (31% of users in 3 prefix buckets) is a structural problem with the ID generation scheme. Three options:

1. Re-partition by content hash of user_id (uniform distribution): requires backfilling 90 days of feature data
2. Salt the partition key (add a random suffix to the partition key to distribute load); requires changes to all downstream joins
3. Use a dynamic partitioning scheme that targets fixed partition sizes rather than fixed partition counts

Evaluate each option: what is the implementation cost, what is the risk, and which would you recommend for the 30-day horizon?

---

Production Challenge 2: Retroactive validation after 3 months of bad data

(Paid | XP: 30)

Scenario: It is December 1st. The ShopSense ML team discovers that a bug introduced on September 1st has been silently corrupting the user_purchase_velocity_7d feature for approximately 8% of users (those with user_id prefixes 'm' and 'n') due to a partition filtering bug that incorrectly excluded their purchase events.

The bug has been in production for 91 days. During this time:

• 12 recommendation model evaluation runs included this feature
• 3 model versions were trained and deployed with corrupted training data (recommendation-v3, v4, v5)
• recommendation-v5 is currently serving 100% of production traffic

The bug is now fixed. The team has clean feature data from December 1st forward.

Your task:

Part A (Blast radius assessment). Quantify the impact of the corrupted data:

1. What fraction of the training dataset was corrupted? (Users with prefix 'm' and 'n' = 8% of users; their purchase history was zero-inflated; the training label rate for these users is ~40% lower than average since purchases were under-counted.)
2. Which model evaluation metrics would have been artificially optimistic due to this corruption? Which would have been pessimistic?
3. How would you test whether the recommendation-v5 model trained on corrupted data is degraded compared to what it would have been with clean data?

Part B (Remediation plan). Write a prioritized remediation plan:

1. Immediate (today): What data integrity action do you take today to ensure no more corrupted features are used in training?
2. This week: Backfill the feature table for the corrupted date range (Sept 1 – Nov 30) for affected users. Write the Spark filter that identifies affected users (prefix 'm' or 'n'), and the DVC command to create a new dataset version with the corrected features.
3. Next week: Retrain the recommendation model on clean data. Do you need to retrain all 3 affected model versions (v3, v4, v5), or just v5? Justify your answer.

Part C (Detection test). Write a Great Expectations expectation that would have caught the original bug (purchase velocity = 0 for a disproportionate fraction of users). The expectation should check:

• The fraction of users with user_purchase_velocity_7d = 0 is within an acceptable range (historically, ~42% of users have no purchases in any 7-day window)
• Segment this check by user_id_prefix to detect prefix-specific anomalies (the 'm'/'n' cohort should have ≈42% zero-velocity, not the 73% that the bug produced)

Part D (Process improvement). The bug was undetected for 91 days because the DQ validation suite only checked overall statistics (not stratified by user_id prefix). Write a 30-day plan to add stratified data quality monitoring to the training data pipeline. Each item should be specific and implementable.

---

Interview Question 1: The streaming vs. batch decision

(Paid | XP: 20)

Question: "Your ML team's feature pipeline is batch-based: features are computed daily. A product manager asks why we can't update features every 5 minutes, wouldn't that make our recommendations better? How do you respond?"

What makes a strong answer:

• Acknowledge the valid underlying goal: fresher features can genuinely improve model quality in certain use cases. Don't dismiss the question.
• Quantify the actual freshness benefit: for recommendation models, how much does feature staleness affect quality? Is the improvement from 5-minute vs. 24-hour features measurable? This depends on the model's use case.
• Be concrete about the streaming tax: don't just say "it's more complex." Give 3 specific operational costs of streaming.
• Frame the trade-off correctly: streaming is a resource allocation decision, not a technical binary. What is the cost per marginal quality gain?

Your task: Write the full response as you would deliver it in a cross-functional meeting with the PM, the engineering manager, and two ML engineers. Then provide:

1. A decision framework: under what specific conditions would you recommend switching from daily batch to 5-minute micro-batch or true streaming? Give 3 concrete criteria.

2. Three operational costs of streaming that a PM likely has not considered (beyond "it's more complex").

3. For ShopSense's recommendation model specifically: if the model is retrained weekly (not daily), what is the actual freshness benefit of 5-minute feature updates vs. daily feature updates? Why is this less significant than it sounds?

---

Interview Question 2: Debugging a silent data pipeline failure

(Paid | XP: 25)

Question: "Walk me through how you would debug a recommendation model that shows 15% lower CTR in production compared to offline evaluation, starting from the hypothesis that the training data is corrupt."

What makes a strong answer:

• Structured debugging methodology: don't just list checks. Show a prioritized, hypothesis-driven investigation workflow.
• Specific technical checks: give concrete queries and metrics, not vague descriptions ("check the data quality").
• Multiple hypothesis coverage: training data corruption is one hypothesis. A strong answer considers at least 3 distinct failure modes that could produce this symptom.
• Escalation criteria: know when to escalate and how to make the rollback decision.

Your task: Write a structured debugging runbook for this scenario. The runbook should have:

Phase 1, Initial triage (< 30 min): What 3 checks do you run first, and what does each check's result tell you?

Phase 2 (Hypothesis testing): Generate 4 distinct hypotheses for why production CTR is 15% below what offline evaluation predicted. For each hypothesis:

• What metric or artifact do you examine?
• What result confirms or rules out the hypothesis?
• How long does it take to check?

Phase 3 (Root cause confirmation): If the investigation confirms training data corruption (hypothesis: non-idempotent pipeline produced a contaminated dataset), what evidence do you need to write the postmortem? List the 5 specific artifacts you need to collect.

Phase 4 (Rollback decision): When is it appropriate to roll back to the previous model version while investigation continues, and when is it better to keep the current model? Give the decision criteria.

---

Interview Question 3: Designing a data pipeline testing strategy from scratch

(Paid | XP: 20)

Question: "We're a team of 8 ML engineers. We have 15 Airflow DAGs for training data preparation, none of which have any tests. Our engineering manager says we have 2 engineer-weeks to add testing. What do you build first, and why?"

What makes a strong answer:

• Prioritize by risk, not by completeness: 2 engineer-weeks is not enough to test everything. The answer requires a principled prioritization.
• Distinguish test types by value: which test type prevents the most expensive failures? (Not the most frequent failures; the most expensive ones.)
• Address the ROI question explicitly: what does 2 engineer-weeks of testing buy? What failure modes does it prevent?
• Be honest about what you're NOT testing: acknowledging trade-offs is more credible than claiming full coverage.

Your task: Design the 2-engineer-week testing sprint:

Week 1 priorities (highest ROI): What do you build in week 1? Give 3 specific tests with justification for why each is high-priority. For each test, estimate: lines of code to write, time to implement, and the failure mode it prevents.

Week 2 priorities: What do you build in week 2, given that week 1 is already done?

What you explicitly decide NOT to test in this sprint: Name 2 test types that are genuinely valuable but below the priority threshold for this sprint, with justification.

The ROI case: After the sprint, an executive asks: "We spent 2 engineer-weeks on testing. What incidents did this prevent?" How do you answer this honestly (without overclaiming) and compellingly (without underclaiming)?

---