Data Engineering & Batch Pipelines

A pipeline isn't production-ready because it ran successfully once. It is production-ready when it runs correctly every time, survives retries without corruption, and fails loudly when data quality doesn't meet the contract.

---

Why this matters

An ML team's recommendation model suddenly showed higher offline recall@10 (0.087 to 0.095) with no model changes. After a week of investigation, the root cause: an Airflow task retry triggered the training data preparation pipeline twice for the same date partition. The pipeline was not idempotent. It appended rather than upserted. Every user event appeared twice in the training set, double-weighting the most active users.

The model shipped. In production, click-through rate (CTR) fell 14% below the previous version. Diagnosing it took 18 days: no content hash on the training dataset, no audit log of the pipeline's execution count per partition, no idempotency test in the test suite.

This is not a rare edge case. It is a predictable failure mode of pipelines designed as one-shot scripts rather than production software systems.

---

The data engineering lifecycle

Every batch pipeline moves data through five stages. Most failures happen at stage interfaces, not inside the compute engine.

Stage 1: Generation (source systems). The origin: application databases, event logs, third-party APIs, user interactions. Critical properties for ML: schema stability, event-time vs. ingestion-time distinction, and mobile event delay (mobile clients buffer offline; 95th percentile delay for ShopSense is 8 hours, with a tail to 72 hours).

Stage 2: Storage. Two-layer pattern: an immutable raw zone (source-native payloads, never overwritten) and a curated zone (typed, versioned Parquet files, analytics and ML ready). Never fix raw data by overwriting it. Produce a new curated version instead.

Stage 3: Ingestion. Batch (daily/hourly), CDC, or streaming: default to batch. The streaming tax (operational complexity, exactly-once semantics, backfill difficulty) is worth paying only when feature value degrades significantly after one hour.

Stage 4: Transformation. ELT is the modern default for cloud warehouses. ETL is necessary for heavy parsing, binary formats, or streaming transforms. Wide denormalized tables are ML-convenient but create inconsistency risk, so use a metrics layer (dbt) to centralize business logic.

Stage 5: Serving. Training pipelines must read features with point-in-time correctness, the version that existed at the label time. Serving should fail loudly if input tables are older than their declared freshness SLA.

---

Think About It: When does ShopSense need streaming vs. batch?

ShopSense is building three ML systems:

1. Weekly recommendation model retraining: the model is retrained every Sunday with data from the past 90 days. Features include user purchase history, product engagement, session behavior.

2. Real-time fraud detection: a model that scores every transaction as it arrives, deciding in < 200ms whether to block it.

3. Demand forecasting for inventory: a model that predicts weekly SKU demand for each of 12 regional distribution centers, retrained monthly.

Work through these before reading on:

1. For which of these three systems is batch ingestion clearly sufficient? For which is streaming required? For which is it ambiguous?

2. The fraud detection model needs to check "how many transactions has this user made in the past 5 minutes?" Can a daily batch pipeline provide this feature? What architecture provides it instead?

3. The recommendation model training pipeline runs weekly. The raw user events land in S3 daily in batch. Is there any scenario where streaming ingestion of user events would improve the recommendation model's quality? Quantify: how much does 24-hour event staleness affect a 90-day training window?

4. "Always use streaming because it future-proofs the architecture." What's the practical cost of this advice for a team of 20 ML engineers? What does a 3–5× operational complexity multiplier mean in staffing terms?

The core trade-off: streaming delivers lower feature latency but higher operational cost. The right choice depends on whether the ML model's quality degrades meaningfully within the batch window, not on what's technically possible.

---

Requirements-first: six questions before writing code

Before selecting any tool or writing any pipeline code, lock down six requirements:

1. Consumers and SLA. Who uses the output? Maximum staleness? For ShopSense's recommendation training: data complete through T–24h, with 1-day retry tolerance.

2. Latency requirement. Daily freshness vs. minutes-level. This decision determines whether streaming is justified. Default to batch.

3. Correctness model. Event-time or processing-time semantics? A feature using processing-time systematically undercounts weekend events due to mobile delay.

4. Data volume and skew. Daily row count and peak multiplier. Partition skew (one partition 10× average size) causes OOM failures in jobs sized for average partitions.

5. Failure tolerance. RTO (how quickly to recover), RPO (how much data loss is acceptable), DLQ (dead letter queue for unprocessable records).

6. Security and privacy. PII fields? Read access roles? PII partition retention schedule?

The rule: if you don't have SLAs and time semantics defined before picking tools, you will ship a pipeline that runs but produces wrong answers.

---

Idempotency: the non-negotiable reliability property

Idempotency is the property that running a pipeline with the same inputs multiple times produces the same output as running it once. It is the most important correctness property for batch pipelines, because pipelines fail and Airflow retries automatically.

Why non-idempotency happens

Three failure patterns:

1. Append-only writes without deduplication:

# NON-IDEMPOTENT: running twice doubles the rows
df.write.mode("append").parquet("s3://bucket/user_events/date=2024-11-15/")  # df is a Spark DataFrame

2. INSERT without clearing the target:

-- NON-IDEMPOTENT: if rows for this date already exist, counts double
INSERT INTO user_purchase_counts
SELECT user_id, COUNT(*) as purchase_count
FROM user_events WHERE date = '2024-11-15'
GROUP BY user_id

3. Non-deterministic record IDs: Using datetime.now() or auto-increment IDs makes records different on each run.

Idempotency patterns

Pattern 1: Delete-then-insert partition

def write_partition_idempotent(df, partition_date: str, output_path: str):
    partition_path = f"{output_path}/date={partition_date}"
    # Delete existing partition before writing
    bucket, prefix = partition_path.replace("s3://", "").split("/", 1)
    boto3.resource("s3").Bucket(bucket).objects.filter(Prefix=prefix).delete()
    df.write.mode("overwrite").parquet(partition_path)

Pattern 2: Deterministic key + MERGE

-- IDEMPOTENT: MERGE on (user_id, date) is deterministic
MERGE INTO user_purchase_counts AS target
USING (
  SELECT user_id, '2024-11-15' as date, COUNT(*) as purchase_count
  FROM user_events WHERE date = '2024-11-15'
  GROUP BY user_id
) AS source
ON target.user_id = source.user_id AND target.date = source.date
WHEN MATCHED THEN UPDATE SET target.purchase_count = source.purchase_count
WHEN NOT MATCHED THEN INSERT *

Pattern 3: Content-hashed output paths

run_hash = hashlib.sha256(json.dumps(config, sort_keys=True).encode()).hexdigest()[:12]
output_path = f"s3://bucket/training_data/v={run_hash}/"
# Running twice with same config writes to same path — idempotent by construction

Pattern 4: Dead Letter Queue for poison records

def process_event_safe(event: dict) -> tuple[dict | None, dict | None]:
    try:
        return transform_event(event), None
    except Exception as e:
        return None, {
            "original_record": event,
            "error_type": type(e).__name__,
            "error_message": str(e),
            "failed_at": datetime.now(timezone.utc).isoformat(),
        }

---

Think About It: Diagnosing idempotency failures in the ShopSense pipeline

ShopSense's compute_user_engagement_features Airflow task runs daily at 2am UTC. On a Wednesday morning, you notice that the user_session_count_7d feature for the previous 7 days has values approximately 1.8× higher than expected. The task ran successfully (status: success in Airflow). No model has been retrained yet.

Diagnosis challenge: work through these before reading on:

1. What are the three most likely explanations for features that are ~1.8× higher than expected? (Not just non-idempotency; think broadly about what could cause a systematic inflation.)

2. You check the Airflow task logs. The task shows it ran twice on Tuesday: once at 2:00am (failed due to Spark OutOfMemoryError), once at 2:47am (succeeded). What does this tell you? Is the task definitely non-idempotent, or just possibly?

3. The task's output is written as: spark_df.write.mode("append").parquet(output_path). What write mode should it be? What's the minimum code change to make this idempotent?

4. After fixing the write mode to be idempotent, you realize the task has been running non-idempotently for 6 weeks; 3 previous incidents caused retries. All 3 retry runs completed successfully and are now reflected in the feature table. The feature table contains data for the past 90 days. How do you assess the blast radius? Which dates are affected, and how do you identify them?

5. You need to backfill the feature table for all affected dates. Write the backfill command for an Airflow DAG named user_engagement_features for dates from 2024-09-15 to 2024-11-13.

Why this matters: Non-idempotency failures are particularly insidious because they look like successful runs in Airflow. The only way to catch them is either through a content-hash comparison or through a downstream model that underperforms and triggers investigation.

---

Time semantics: three times every pipeline must track

Event time: when the action happened (user clicked at 14:32:07 UTC). Ingestion time: when the event arrived at the warehouse (mobile was offline; arrived 4 hours later). Processing time: when the pipeline job ran (daily batch at 02:00 UTC).

Correctness for ML features requires event-time semantics.

-- WRONG (processing time): systematically undercounts mobile-heavy days
SELECT COUNT(*) FROM user_events
WHERE processed_date BETWEEN '2024-11-09' AND '2024-11-15';

-- CORRECT (event time with late-arrival cutoff)
SELECT COUNT(*) FROM user_events
WHERE event_timestamp >= '2024-11-09T00:00:00Z'
  AND event_timestamp <  '2024-11-16T00:00:00Z'     -- 7 days: Nov 9 through Nov 15
  AND ingestion_timestamp < '2024-11-16T08:00:00Z'  -- late-arrival cutoff: window end + 8h

Late-arriving data handling:

Strategy	Description	Tradeoff
Fixed look-back + hard cutoff	Wait N hours after window closes	Simple, reproducible; misses events after N hours
Incremental backfill	Recompute last 7 days each run	Maximum coverage; training data changes over time
Late-event partition	Separate is_late: true partition	Flexible; downstream consumers choose what to include

---

Data quality and validation in pipelines

DQ validation must be a blocking gate in the pipeline, not background monitoring. If validation fails, the pipeline stops. Training does not proceed on suspect data.

Minimum DQ check set

# Blocking DQ checks before training data is consumed
DQ_CHECKS = [
    # Schema conformance
    "table user_events has columns: [user_id, event_type, event_timestamp, product_id, session_id]",
    # Row count
    "count of user_events WHERE date='2024-11-15' BETWEEN 150_000_000 AND 220_000_000",
    # Null rates
    "null rate of user_events.user_id < 0.001",
    "null rate of user_events.event_timestamp == 0.0",
    # Enum validity
    "user_events.event_type IN ('view', 'click', 'add_to_cart', 'purchase', 'search')",
    # Freshness
    # (daily runs only; skip when backfilling a past date)
    "max(user_events.ingestion_timestamp) > NOW() - INTERVAL 2 HOURS",
    # Uniqueness
    "(user_id, session_id, event_timestamp) is unique",
]

Great Expectations integration

import great_expectations as gx

context = gx.get_context()
batch = (
    context.data_sources.add_pandas("shopsense_events")
    .add_dataframe_asset("user_events")
    .add_batch_definition_whole_dataframe("daily_partition")
    .get_batch(batch_parameters={"dataframe": user_events_df})
)
suite = context.suites.add(gx.ExpectationSuite(name="user_events_training_suite"))

suite.add_expectation(gx.expectations.ExpectTableColumnsToMatchOrderedList(
    column_list=["user_id", "event_type", "event_timestamp", "product_id", "session_id"],
))
suite.add_expectation(gx.expectations.ExpectTableRowCountToBeBetween(
    min_value=150_000_000, max_value=220_000_000,
))
# mostly=0.999 allows at most a 0.1% null rate
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="user_id", mostly=0.999))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeInSet(
    column="event_type",
    value_set=["view", "click", "add_to_cart", "purchase", "search"],
))

validation_result = batch.validate(suite)
if not validation_result.success:
    raise PipelineDataQualityError(
        "DQ validation failed — blocking training pipeline."
    )

Distribution drift checks

from scipy import stats

def check_distribution_drift(current_df, baseline_df, column, ks_threshold=0.05):
    statistic, p_value = stats.ks_2samp(
        current_df[column].dropna(),
        baseline_df[column].dropna(),
    )
    return p_value < ks_threshold, p_value  # (is_drifted, p_value)

A K-S p-value < 0.05 indicates statistically significant drift. On batches with millions of rows nearly every difference is significant, so also gate on the K-S statistic (the size of the gap) against a calibrated threshold. Drift should block training with a human review step; drift can be legitimate (holiday season) or malignant (upstream corruption).

---

Data versioning: DVC, Delta Lake, LakeFS

DVC: Git-like content-addressed dataset versioning. A .dvc file (content hash pointer) committed alongside training code. dvc pull downloads the exact dataset for any Git commit. Model-to-training-data linkage: git checkout model-v4-training-data && dvc pull.

Delta Lake: ACID-compliant, versioned table operations on S3/HDFS (Armbrust et al., VLDB 2020). Time-travel queries: spark.read.format("delta").option("versionAsOf", 42).load(...). Every write appends a transaction log entry, full version history at the table level.

LakeFS: Git-like branch/commit/merge on S3 object storage. Pipelines work on isolated branches; merge to main only after validation. Prevents experimental runs from corrupting production training data paths.

---

Orchestration: Airflow, Prefect, Dagster

Airflow (Beauchemin, Airbnb, 2014): DAG of tasks defined in Python. Four production principles: (1) first-class backfill support, (2) idempotent tasks so retries are safe, (3) separate orchestration from compute, (4) DAG definitions as code with CI tests.

Prefect: Python-native @flow and @task decorators with native retry caching: @task(retries=3, cache_policy=INPUTS). Best for dynamic task graphs and teams preferring managed orchestration (Prefect Cloud).

Dagster: software-defined assets (SDAs). Pipeline tasks are described in terms of the data assets they produce and consume. The asset graph is the lineage graph, every asset knows its upstream dependencies. Best for ML platforms where data lineage is a first-class requirement.

Property	Airflow	Prefect	Dagster
Abstraction	Tasks	Flows	Assets
Data lineage	Partial (assets, OpenLineage provider)	None	First-class
Setup	High	Low	Medium
Best for	Stable ETL	Dynamic ML	Asset-centric ML

---

Think About It: Which orchestrator should ShopSense adopt?

ShopSense's ML platform is currently a collection of cron jobs and ad-hoc Spark scripts. The team wants to formalize their pipeline orchestration. The engineering lead has shortlisted three options:

• Option A: Apache Airflow: most of the team has used it before, large operator ecosystem, but requires self-hosted deployment (scheduler, webserver, workers, metadata DB)
• Option B: Prefect Cloud: managed service, Python-native, lower ops overhead, good for dynamic pipelines
• Option C: Dagster: software-defined assets, built-in lineage tracking, requires rebuilding existing pipelines as asset definitions

Context: ShopSense has 20 ML engineers, 50 Spark jobs, 3 Airflow engineers on the data platform team, 200 datasets, compliance requirements (payment data), and a goal to make data lineage visible across all pipelines within 6 months.

Work through these before reading on:

1. The team's existing Airflow expertise suggests Option A. What is the cost of switching to Option B or C? What is the cost of staying with Option A?

2. The compliance requirement (data lineage for payment data) is a strong signal for one of the three options. Which one, and why? How does this change the evaluation?

3. If ShopSense adopts Dagster, existing Spark scripts need to be wrapped as Dagster assets. Estimate: how long would it take to migrate 50 existing Spark jobs to Dagster assets for a team of 3 data platform engineers? What is the risk during migration?

4. A pragmatic middle-ground: adopt Airflow today, add lineage by integrating OpenLineage (which emits column-level lineage events from Airflow tasks via a plugin). Does this make Option A viable for the compliance requirement? What does it lack that Dagster's asset model provides natively?

There is no universally correct answer here. The goal is to reason through trade-offs, not to pick the "right" tool.

---

Testing batch pipelines: four levels

Level 1 (Unit tests). Test transform functions with fixture DataFrames. Assert correct business logic, edge cases, and window boundary behavior.

Level 2 (DQ contract tests). Test schema contracts, null rates, enum validity, uniqueness constraints.

Level 3 (Pipeline integration tests). Run the full pipeline on a 1K-row fixture. Assert output columns, row count, and idempotency (run twice, assert same output hash).

Level 4 (Schema regression tests). Validate upstream table schema against the registered contract in CI. Fail the build if a breaking change is detected upstream.

---

Villain case study: the non-idempotent pipeline

Setting: An ML team at a logistics recommendation platform, 2023. Daily Airflow DAG computes user engagement features for model training.

The incident: At 02:00 UTC, the pipeline triggered. At 02:47, a network timeout caused the compute_features task to fail, but only after partially writing 87M rows to the output Parquet file. Airflow auto-retried at 03:02. The retry ran the task from scratch and appended another 180M rows to the same file.

Training dataset: 267M rows instead of 180M. The 87M rows from the partial first run (the highest-traffic first 2 hours of the day, dominated by the most active users) were duplicated.

The model trained on the contaminated dataset showed recall@10 = 0.095. The team shipped it. Production CTR dropped 14%. Investigation took 18 days.

What would have prevented this:

• Delete-then-insert write pattern: retry overwrites the partial output instead of appending to it
• Idempotency CI test: double-run test catches append-on-retry before it reaches production
• Content hash logged to MLflow: different hash between contaminated and clean dataset flagged immediately

Reflection exercises:

1. The network timeout is what triggered the retry. But the root cause is the write mode choice. Why is mode="append" such a common default in ML pipelines? What are the legitimate use cases for append-mode writes in ML data pipelines, and what is the invariant that must hold for append to be safe?

2. The partial first run wrote 87M rows: the first 2 hours of the day's events. The model "improved" because these events over-represented the most active users (morning commuters on mobile). If you were debugging this incident and saw recall@10 improve but CTR drop, what distribution shift would you look for in the serving logs vs. training logs?

3. After restoring correct features via a full backfill, the team proposes adding an alert: "if the training dataset row count changes by > 5% between runs, page oncall." Is this a sufficient safeguard against non-idempotency? What would it catch? What would it miss?

Expert analysis: The alert suggested in question 3 is a reasonable second-order safety check but is not a substitute for idempotency. It would have caught the 267M/180M = 48% inflation in this specific incident. But a partial retry that wrote 183M of 180M rows (1.7% inflation) would pass the 5% threshold. An idempotency test is the only first-order safeguard: it directly asserts the property rather than monitoring a proxy.

---

Case study: DoorDash's declarative feature engineering (Fabricator)

(Source: DoorDash Engineering. "Introducing Fabricator: A Declarative Feature Engineering Framework." DoorDash Engineering Blog.)

DoorDash's ML platform grew to 100+ active feature pipelines across multiple teams. The common complaint: each team had implemented the same patterns (rolling aggregation, feature store writes, schema validation) differently. Idempotency guarantees varied. Backfill procedures were inconsistent. Monitoring was team-by-team.

Fabricator standardized this by making feature engineering pipelines declarative: engineers specify what features to compute (source, aggregation, window, group-by, output schema), and Fabricator handles how (orchestration with Dagster, Spark execution on Databricks, feature store materialization to Redis). DoorDash's post describes this platform design. The YAML spec below and the four design decisions that follow are an illustrative reconstruction to practice with, not Fabricator's actual syntax or documented behavior.

feature_set: restaurant_delivery_features
source:
  type: kafka_topic
  topic: delivery_completed_events

aggregations:
  - name: avg_delivery_time_30m
    field: delivery_duration_seconds
    operation: AVG
    window: 30m
    group_by: [restaurant_id]

  - name: order_count_1h
    field: order_id
    operation: COUNT_DISTINCT
    window: 1h
    group_by: [restaurant_id]

output:
  type: feature_store
  key: restaurant_id
  ttl: 2h
  schema_version: 1.2

DoorDash reports 100+ pipelines, 500+ unique features, and 100B+ daily feature values.

Design decision 1: declarative over imperative

Fabricator trades expressiveness for standardization. Teams cannot write arbitrary Spark transformations; they must express their feature computation within Fabricator's aggregation vocabulary. Teams with genuinely complex features (multi-table joins, ML-generated sub-features) must use Fabricator's escape hatch: a custom_transform block that embeds Python code, but loses the automatic idempotency and backfill guarantees for that component.

Analysis question 1: What is the primary motivation for making feature pipelines declarative rather than imperative? Give one benefit and one limitation of this approach. When would a team genuinely need the custom_transform escape hatch?

Design decision 2: backfill as a first-class operation

Fabricator pipelines support backfill natively: running a Fabricator job with --backfill --start 2024-09-01 --end 2024-11-15 recomputes all features for that date range using the same idempotency guarantees as a live run. This is possible because the declarative specification contains enough information to determine the exact computation for any historical date range.

Analysis question 2: Contrast Fabricator's declarative backfill with running a traditional Airflow DAG backfill. What property of declarative pipelines makes backfill simpler? What class of pipeline (give an example) cannot be safely backfilled regardless of whether it is declarative or imperative?

Design decision 3: schema version as a first-class field

The schema_version field in the Fabricator output spec is not documentation; it is an enforced contract. When the feature schema changes (new field added, field renamed, type changed), the schema_version must be incremented. Fabricator validates the output against the registered schema version before writing to the feature store, and rejects the write if the output does not match.

Analysis question 3: What class of bug does schema_version enforcement prevent? What would happen if DoorDash's serving pipeline expected avg_delivery_time_30m (type: float) but a Fabricator pipeline update changed it to average_delivery_duration_30m (same data, different name) without bumping the schema version?

Design decision 4: 100B+ daily feature values at scale

Fabricator's scale (100B+ feature values per day) requires that individual feature computation pipelines be efficient at storage I/O, not just compute. DoorDash's feature store writes are batched and columnar; the feature values are stored in a format optimized for low-latency key lookups at serving time (Redis for online, Parquet for offline).

Analysis question 4: At 100B feature values per day, what would happen if each feature value write to Redis were an individual RPC call? How does batch writing change the storage I/O pattern? At ShopSense's scale (10M DAU × 50 features = ~500M feature values per day), would Redis be the appropriate online store choice, or would you consider alternatives?

---

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)?

---