Workbook

Data Cataloging, Metadata & Lineage: exercises

Applied exercises from this chapter. Write your own answer first, then open “Expert thinking” under each exercise to compare your reasoning.

Advanced Exercise 1: Design a complete catalog entry for the ShopSense demand forecasting dataset

(Paid | XP: 40)

ShopSense's demand forecasting model predicts daily inventory requirements for 50M SKUs across 12 regional distribution centers. It is trained weekly on 2 years of historical sales data joined to product catalog metadata.

The training dataset has the following source tables:

sales_events (fields: order_id, product_id, quantity, sale_timestamp, store_id, customer_id)
product_catalog (fields: product_id, category_l1, category_l2, brand_id, msrp, weight_kg, is_seasonal)
store_metadata (fields: store_id, region_code, store_type, capacity_tier)
holiday_calendar (fields: date, is_holiday, holiday_name, country_code)

The training dataset is produced by a weekly Airflow DAG that joins these tables, computes 14 derived features (rolling 7d/30d/90d sales averages, year-over-year ratios, seasonal adjustment factors), and writes a Parquet snapshot to S3.

Your task: Design a complete catalog entry for this training dataset that includes all four catalog layers.

Part A, Schema registry entry (required fields): Write a YAML schema registry entry for the training dataset. Include:

All 14 derived feature field names, types, nullability, and descriptions
A pii: true/false classification for each field
A changelog entry for the most recent schema version
Two fields that you would mark as breaking: true on rename, with justification

Part B, Business glossary entry (required terms): Define three terms from the demand forecasting domain that require glossary entries because they are ambiguous across teams:

A definition for "demand" (the model's target variable): what is the precise operationalization?
A definition for "seasonal product": is this determined by is_seasonal flag, by category, by rolling sales variance, or by explicit calendar configuration?
A definition for "effective training cutoff": the last date in the training data, versus the date the training snapshot was created, versus the date the model was deployed

For each term, provide: canonical definition, owner, implementing field, and at least one deprecated definition with the date it was deprecated.

Part C (Column-level lineage path): Write the complete column-level lineage path for the derived feature rolling_sales_7d from source columns through intermediate computations to the final training dataset field. Include the transformation type and any filter predicates.

Part D (Model audit trail entry): Write the YAML model audit trail entry for demand-forecasting-v3. Include: training dataset version hash, all upstream schema versions, code commit, deployment approval chain, and three offline metrics. Add a known_limitations field with at least two real issues you'd expect to find in a demand forecasting model trained on 2 years of historical data that includes a one-off demand shock (for example a pandemic-era buying spike or a major supply shortage).

Acceptance criteria:

Schema registry entry has all required fields, PII classification is correct for all fields
Business glossary entries include canonical definition, owner, implementing field, and deprecated definition
Lineage path is traced to specific source columns in specific source tables
Model audit trail is complete and the known_limitations field reflects real ML issues (distribution shift, out-of-distribution events)

Advanced Exercise 2: Impact analysis: trace the `product_category` rename

(Paid | XP: 40)

ShopSense's data engineering team has proposed renaming the field product_category to category_primary in the product_catalog dataset. They've submitted a schema change request. As the data platform engineer, you need to produce an impact analysis report before the change ships.

Known facts:

product_catalog.product_category is a string field with 47 possible values representing L1 product categories
The rename is motivated by a naming convention standardization effort (all *_category fields will be renamed to category_*)
The proposed migration timeline is 2 weeks

The lineage graph (partial) for product_catalog.product_category:

product_catalog.product_category
  ├── ETL: item_category_features_daily
  │     └── output: item_category_embeddings.category_id (field)
  │           └── feature_view: item_category_embedding_v1 (Feast)
  │                 └── training: recommendation_model_v4 (50% traffic)
  │                 └── training: recommendation_model_v3 (50% traffic)
  │                 └── training: search_ranker_v2 (staging)
  │
  ├── ETL: product_category_stats_weekly
  │     └── output: category_sales_summary.category_name (field)
  │           └── dashboard: Category Revenue by Month (Finance team)
  │           └── dashboard: Inventory Health (Operations team)
  │
  ├── ETL: demand_forecast_features
  │     └── output: product_features.category_l1 (field)
  │           └── training: demand_forecasting_v3 (production)
  │
  └── Query: presto_ad_hoc_queries (estimated 12 unique queries/week referencing this field)

Your task:

Part A (Generate the breaking change report). Write the full impact analysis report that would be sent to the data engineering team before the rename ships. Include:

Complete list of direct consumers (ETL jobs, feature views, pipelines)
Complete list of model versions affected (and their current traffic share)
Complete list of dashboards affected (and their owning teams)
Estimated migration effort per consumer type (simple query change vs. full feature store recompute vs. model retrain)
Migration sequencing recommendation: in what order should consumers be migrated? What is the dependency order?

Part B (Migration plan). Design a 2-week migration plan that allows the rename to ship without service disruption. Answer:

How do you handle the Presto ad hoc queries (12/week from unknown analysts)? You cannot enumerate all of them.
Should recommendation_v3 and recommendation_v4 be retrained with the new field name, or just migrated at the serving feature pipeline level? What are the risks of each approach?
How do you validate that the migration is complete before decommissioning the old field name?

Part C (Schema contract test). Write a Python pytest function test_product_catalog_schema_v3_contract() that validates the product_catalog Parquet file against a YAML schema contract. The test should:

Check that all required fields exist
Validate that category_primary (new name) exists in the new schema version
Assert that product_category (old name) does not exist in schema versions ≥ 3.0
Check that the allowed_values constraint for category_primary has exactly 47 values
Fail with a descriptive message if any check fails

Advanced Exercise 3: Lineage audit: trace a production regression backward

(Paid | XP: 40)

ShopSense's recommendation model's click-through rate dropped from 3.1% to 2.4% (22% regression) over 10 days. You are the on-call ML engineer. You have access to:

DataHub lineage graph (queried via REST API)
MLflow experiment tracking (with model artifact + training dataset hash logged)
Airflow execution logs (last 30 days)
The model's monitoring dashboard (feature distribution drift, prediction distribution)

Evidence gathered so far:

The model was last retrained 6 weeks ago. No retraining has occurred since.
The serving endpoint's p95 latency is normal. No errors in serving logs.
MLflow shows the current production model was trained with training dataset hash sha256:a3b8c91f.
Monitoring dashboard shows PSI > 0.2 for the feature item_category_embedding (significant drift)
PSI for all other features is < 0.1

The DataHub lineage graph for item_category_embedding:

item_category_embedding (feature, PSI=0.31)
  ← computed by: item_category_features_daily (Airflow DAG)
      ← reads from: product_catalog (dataset)
          └── product_catalog.category_primary (field)
              └── ingested by: product_catalog_ingestion (Airflow DAG)
                  └── source: vendor_catalog_api v3.1 (external)

Airflow execution log anomalies (last 30 days):

product_catalog_ingestion upgraded from vendor API v2.8 to v3.1 on day -11 (11 days ago)
vendor API v3.1 release notes: "Restructured category taxonomy. category_primary field now returns ISO category codes (e.g., ELECTRONICS_COMPUTING) instead of display names (e.g., Electronics > Computing)."

Your task:

Part A (Root cause statement). Write a one-paragraph root cause statement suitable for the incident postmortem. Include: what changed, when, in which system, what the downstream effect was, and why the PSI check caught it when it did (after 10 days of degradation, not immediately).

Part B (Impact quantification). The item_category_embedding feature is a 64-dimensional embedding computed from the category_primary field. If vendor API v3.1 changed the category format from display names to ISO codes, and the embedding model was trained on display names, answer:

What would the embedding model output for an ISO code input it has never seen in training?
Would this produce a detectable error (exception) or a silent degradation? Why?
If the previous category had 47 display-name values and the new API returns ~200 ISO codes, how would this affect the serving distribution of the embedding vectors?

Part C (Immediate remediation options). Propose three possible remediation strategies, ordered by implementation speed:

A < 4 hour fix that degrades gracefully but doesn't fully restore performance
A < 24 hour fix that restores performance without retraining the model
A < 1 week fix that fully resolves the root cause with no future recurrence

For each, describe the trade-offs.

Part D (Prevention test). Write a Great Expectations (or equivalent) validation that would have caught this issue at ingestion time, before product_catalog_ingestion wrote the v3.1 data to S3. The test should validate the format of the category_primary field values.

Real-World Implementation 1: Schema registry validator

(Paid | XP: 40)

Implement a schema registry validator that reads a YAML schema definition, validates a Parquet dataset file against it, and generates a breaking change report when comparing two schema versions.

"""
schema_registry_validator.py

Usage:
    python schema_registry_validator.py validate --schema user_events.yaml --data user_events.parquet
    python schema_registry_validator.py diff --old user_events_v3.yaml --new user_events_v4.yaml
"""

import yaml
import pyarrow.parquet as pq
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any


@dataclass
class FieldSpec:
    name: str
    type: str
    nullable: bool
    pii: bool
    allowed_values: list[str] | None = None
    retention_days: int | None = None


@dataclass
class SchemaSpec:
    dataset: str
    version: str
    fields: list[FieldSpec]
    changelog: dict[str, Any]


@dataclass
class ValidationResult:
    is_valid: bool
    missing_fields: list[str]        # in schema but not in data
    extra_fields: list[str]          # in data but not in schema
    type_mismatches: list[tuple[str, str, str]]  # (field_name, expected, actual)
    nullable_violations: list[str]   # non-nullable fields with nulls
    allowed_value_violations: list[tuple[str, list[str]]]  # (field, invalid_values)
    pii_fields: list[str]            # fields flagged as PII

    def summary(self) -> str:
        ...


@dataclass
class BreakingChangeReport:
    old_version: str
    new_version: str
    breaking_changes: list[dict[str, Any]]    # {"field": ..., "change_type": ..., "description": ...}
    non_breaking_changes: list[dict[str, Any]]
    migration_notes: list[str]

    def is_breaking(self) -> bool:
        return len(self.breaking_changes) > 0


def load_schema(schema_path: str) -> SchemaSpec:
    """Load and parse a YAML schema file into a SchemaSpec."""
    ...


def validate_dataset(schema: SchemaSpec, data_path: str) -> ValidationResult:
    """Validate a Parquet file against a SchemaSpec.

    Checks:
    - All required fields present in data
    - No unexpected fields in data (warn, not fail)
    - Type compatibility (mapped from Parquet types to schema types)
    - Null counts for non-nullable fields
    - Allowed value constraints (sample-based, up to 100K rows)
    """
    ...


def diff_schemas(old_schema: SchemaSpec, new_schema: SchemaSpec) -> BreakingChangeReport:
    """Compare two schema versions and classify each change as breaking or non-breaking.

    Breaking changes:
    - Field removed
    - Field renamed (detected as: old field missing + new field with same semantics added)
    - Type changed to incompatible type (e.g., string → int, float → string)
    - nullable: false → nullable: true for a previously required field (consumers that
      assumed no nulls can now break; tightening nullable: true → false is non-breaking
      for readers)
    - allowed_values list changed to a subset (previously valid values now rejected)

    Non-breaking changes:
    - Field added (new optional field)
    - Description/documentation change
    - allowed_values list extended (new values added)
    - retention_days changed
    """
    ...


def generate_migration_notes(report: BreakingChangeReport) -> list[str]:
    """Generate human-readable migration instructions for each breaking change."""
    ...


# ---- Type compatibility mapping ----
# Parquet physical types to schema logical types
# (timestamps are stored as INT64 with a TIMESTAMP logical type in current Parquet;
#  INT96 is a legacy encoding that some older writers still produce)
PARQUET_TO_SCHEMA_TYPE: dict[str, list[str]] = {
    "INT32": ["int", "integer"],
    "INT64": ["long", "bigint", "timestamp", "timestamp_utc"],
    "FLOAT": ["float"],
    "DOUBLE": ["double", "float"],
    "BYTE_ARRAY": ["string", "bytes"],
    "BOOLEAN": ["bool", "boolean"],
    "INT96": ["timestamp", "timestamp_utc"],
}


if __name__ == "__main__":
    import sys

    if sys.argv[1] == "validate":
        schema = load_schema(sys.argv[3])
        result = validate_dataset(schema, sys.argv[5])
        print(result.summary())
        sys.exit(0 if result.is_valid else 1)

    elif sys.argv[1] == "diff":
        old = load_schema(sys.argv[3])
        new = load_schema(sys.argv[5])
        report = diff_schemas(old, new)
        for change in report.breaking_changes:
            print(f"[BREAKING] {change['field']}: {change['description']}")
        for change in report.non_breaking_changes:
            print(f"[non-breaking] {change['field']}: {change['description']}")
        sys.exit(1 if report.is_breaking() else 0)

Your task:

Implement load_schema(): parse the YAML schema file into a SchemaSpec object with all FieldSpec entries. Handle the case where allowed_values and retention_days are optional (not all fields specify them).
Implement validate_dataset(): read the Parquet file with PyArrow, compare fields to the schema, check nullable constraints (count nulls per non-nullable field), and validate allowed values on a sample of up to 100K rows. Use PARQUET_TO_SCHEMA_TYPE to map Parquet physical types to schema logical types.
Implement diff_schemas(): compare two schema versions and classify each difference as breaking or non-breaking. Pay particular attention to: renamed fields (a field removal in old + a new field that looks semantically similar; you can use field-name edit distance as a heuristic), type changes, and allowed_value list changes.
Implement generate_migration_notes(): for each breaking change, produce a human-readable migration instruction. Example: "Field 'product_clicked' was renamed to 'product_id'. Update all references in consumers to use 'product_id'."
Implement ValidationResult.summary(): produce a concise multi-line string summarizing the validation result. Include counts of each issue type and a one-line pass/fail verdict.

Test your implementation against these cases:

Schema with 5 fields; Parquet file missing 1 required field → validation fails
Schema where event_type has allowed_values: [view, click, purchase]; Parquet file contains value "search" → violation detected
Old schema with product_clicked; new schema with product_id (same type) → breaking change with rename detection
Old schema with session_duration: int; new schema with session_duration: float → type compatibility check (int → float is non-breaking; float → int is breaking)

Real-World Implementation 2: OpenLineage event emitter for pandas transforms

(Paid | XP: 35)

Implement a decorator and context manager that automatically emits OpenLineage-compatible column lineage events when a pandas transformation function runs. The emitter should capture input/output datasets, field mappings, and job metadata.

"""
lineage_emitter.py

Wraps pandas transform functions to emit OpenLineage column lineage events.

Usage:
    @emit_column_lineage(
        input_dataset="user_events",
        output_dataset="user_purchase_velocity_7d",
        column_mappings={
            "velocity_7d": ["event_type", "event_timestamp", "user_id"],
            "user_id": ["user_id"],
        },
        transform_description="Aggregate purchase events over trailing 7-day window per user",
    )
    def compute_purchase_velocity(events_df: pd.DataFrame) -> pd.DataFrame:
        ...
"""

import functools
import hashlib
import json
import uuid
from datetime import datetime, timezone
from typing import Any, Callable
import pandas as pd


# OpenLineage event schema (simplified)
OPENLINEAGE_SPEC_VERSION = "1.0.0"


def build_column_lineage_facet(
    output_column: str,
    input_columns: list[str],
    input_dataset: str,
) -> dict[str, Any]:
    """Build the entry for one output column in a ColumnLineageDatasetFacet's `fields` map.

    build_run_event() places the merged entries under
    outputs[].facets["columnLineage"]["fields"].
    """
    return {
        output_column: {
            "inputFields": [
                {
                    "namespace": "shopsense",
                    "name": input_dataset,
                    "field": col,
                    # OpenLineage 1.x: transformation type is DIRECT or INDIRECT, with a subtype
                    "transformations": [{"type": "DIRECT", "subtype": "AGGREGATION"}],
                }
                for col in input_columns
            ],
            "transformationDescription": "",  # filled in by caller
        }
    }


def build_run_event(
    job_name: str,
    run_id: str,
    event_type: str,      # START | COMPLETE | FAIL
    input_dataset: str,
    output_dataset: str,
    input_schema: list[dict],   # [{"name": ..., "type": ...}]
    output_schema: list[dict],
    column_lineage: dict[str, Any],
    row_count: int | None = None,
    started_at: str | None = None,
    ended_at: str | None = None,
    custom_facets: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Build an OpenLineage RunEvent dict compliant with the OL 1.0.0 spec."""
    ...


def emit_to_marquez(event: dict[str, Any], marquez_url: str = "http://localhost:5000") -> bool:
    """POST the OpenLineage event to a Marquez server. Return True on success."""
    ...


def emit_column_lineage(
    input_dataset: str,
    output_dataset: str,
    column_mappings: dict[str, list[str]],  # {output_col: [input_col, ...]}
    transform_description: str = "",
    marquez_url: str = "http://localhost:5000",
    dry_run: bool = False,   # if True, print event JSON instead of POSTing
):
    """Decorator that emits OpenLineage column lineage events around a pandas transform."""
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(df: pd.DataFrame, *args, **kwargs) -> pd.DataFrame:
            run_id = str(uuid.uuid4())
            job_name = func.__name__
            started_at = datetime.now(timezone.utc).isoformat()

            # Extract input schema from the DataFrame
            input_schema = _extract_schema(df)

            # Emit START event
            start_event = build_run_event(
                job_name=job_name,
                run_id=run_id,
                event_type="START",
                input_dataset=input_dataset,
                output_dataset=output_dataset,
                input_schema=input_schema,
                output_schema=[],  # unknown until function completes
                column_lineage={},
                started_at=started_at,
            )
            _emit_event(start_event, marquez_url, dry_run)

            try:
                result_df = func(df, *args, **kwargs)
                ended_at = datetime.now(timezone.utc).isoformat()

                # Extract output schema and build column lineage facet
                output_schema = _extract_schema(result_df)
                column_lineage: dict[str, Any] = {}
                for col, input_cols in column_mappings.items():
                    column_lineage.update(
                        build_column_lineage_facet(col, input_cols, input_dataset)
                    )

                # Emit COMPLETE event
                complete_event = build_run_event(
                    job_name=job_name,
                    run_id=run_id,
                    event_type="COMPLETE",
                    input_dataset=input_dataset,
                    output_dataset=output_dataset,
                    input_schema=input_schema,
                    output_schema=output_schema,
                    column_lineage=column_lineage,
                    row_count=len(result_df),
                    started_at=started_at,
                    ended_at=ended_at,
                )
                _emit_event(complete_event, marquez_url, dry_run)
                return result_df

            except Exception as e:
                # Emit FAIL event
                fail_event = build_run_event(
                    job_name=job_name,
                    run_id=run_id,
                    event_type="FAIL",
                    input_dataset=input_dataset,
                    output_dataset=output_dataset,
                    input_schema=input_schema,
                    output_schema=[],
                    column_lineage={},
                    ended_at=datetime.now(timezone.utc).isoformat(),
                    custom_facets={"errorMessage": {"message": str(e)}},
                )
                _emit_event(fail_event, marquez_url, dry_run)
                raise


        return wrapper
    return decorator


def _extract_schema(df: pd.DataFrame) -> list[dict[str, str]]:
    """Extract field names and types from a DataFrame."""
    return [{"name": col, "type": str(df[col].dtype)} for col in df.columns]


def _emit_event(event: dict, marquez_url: str, dry_run: bool) -> None:
    if dry_run:
        print(json.dumps(event, indent=2))
    else:
        emit_to_marquez(event, marquez_url)

Your task:

Implement build_run_event(): construct the full OpenLineage RunEvent dict following the OpenLineage 1.0 spec structure: eventType, eventTime, run (with runId and facets), job (with namespace, name, facets), inputs (list of InputDataset), outputs (list of OutputDataset with ColumnLineageDatasetFacet). The row_count should be placed in a DataQualityMetricsFacet on the output dataset.
Implement emit_to_marquez(): POST the event dict as JSON to {marquez_url}/api/v1/lineage. Return True if HTTP 200-204, False otherwise. Log the error (don't raise) so the pipeline continues even if Marquez is unavailable.
Write a test test_purchase_velocity_lineage() that:
- Decorates a test version of compute_purchase_velocity() with @emit_column_lineage(..., dry_run=True)
- Calls the function with a synthetic 100-row DataFrame
- Captures stdout (the dry-run JSON output)
- Parses the JSON and asserts that: the eventType is "COMPLETE", the output dataset name matches, the velocity_7d column maps to ["event_type", "event_timestamp", "user_id"] in the column lineage facet, and row_count > 0
Handle the FAIL event correctly: if the wrapped function raises an exception, the decorator must emit a FAIL event with the error message in custom_facets, then re-raise the original exception. Write a test test_lineage_emitter_fail_event() that verifies the FAIL event is emitted before the exception propagates.

Production Challenge 1: The schema version divergence incident

(Paid | XP: 40)

Scenario: It is 6:47am. You are on call. PagerDuty fires: recommendation model serving error rate jumped from 0.02% to 3.8% over the past 20 minutes.

Initial investigation:

Serving endpoint error logs (6:27am – 6:47am):
  KeyError: 'category_primary' in feature_assembler.py:143

feature_assembler.py:143:
  category_emb = self.category_encoder[features['category_primary']]

Checking the catalog:

product_catalog schema registry:
  Current version: 3.0.0 (deployed 6:24am by data-eng team)
  Previous version: 2.9.0

Changelog:
  3.0.0 (2024-11-18):
    change: "Renamed product_category to category_primary (naming convention standardization)"
    breaking: true
    migration: "All consumers update field reference from product_category to category_primary"

The problem: The serving infrastructure reads features from the online feature store (Redis). The feature store is populated by a Feast materialization job that reads from the product_catalog dataset. The Feast feature view definition still references product_category (old name). When the data engineering team shipped schema 3.0.0, the feature store materialization job started failing silently, writing null values to Redis for the category_primary key. The serving pipeline expected category_primary but the value in Redis was null.

Complication: The product_catalog v2.9.0 data has been overwritten by v3.0.0 in S3. You cannot simply roll back the data source.

Your task:

Part A, Immediate fix (< 30 minutes). The serving endpoint must be restored first. The category embedding is used to rank 10 candidate products. Design an emergency fallback strategy that:

Allows the serving endpoint to continue returning recommendations even with null category_primary
Degrades gracefully (uses a global category prior or popularity-based fallback)
Does not require redeployment of the model binary Describe the code change and estimate the serving quality degradation (recommendation quality, not error rate).

Part B, Short-term fix (< 4 hours). Restore full serving quality by fixing the Feast feature view definition to reference category_primary (new field name) and re-materializing the feature store. Write the updated Feast feature view definition. What is the minimum re-materialization window you need to restore production quality? What do you set as the serving-time default for records not yet re-materialized?

Part C (Root cause and prevention). The data engineering team followed the change management process; they set breaking: true in the schema changelog. But the impact analysis missed the Feast feature view as a consumer. Why? (What type of consumer would not appear in a table-level lineage graph?)

Write a detection test test_product_catalog_consumers_are_registered() that would have caught this gap, verifying that all registered Feast feature views that reference product_catalog fields are listed in the catalog's consumer registry.

Part D (Process improvement). Write the postmortem "What we will do differently" section (3–5 bullets). Each bullet should be a concrete, implementable process or tooling change, not a vague recommendation like "improve communication." At least one bullet should address how Feast feature view definitions are kept in sync with schema registry versions.

Production Challenge 2: Retroactive catalog reconstruction

(Paid | XP: 35)

Scenario: ShopSense's ML team just completed a compliance audit. The auditor asked a straightforward question: "For your recommendation model currently in production, what dataset was it trained on, what were the schema versions of the upstream sources, and who approved the deployment?"

The ML team cannot answer. They have MLflow tracking for the model artifact, but the training dataset was not versioned; it was overwritten in S3 by the next training run. The schema registry was implemented 3 months ago, but the model was trained 8 months ago. There is no deployment approval trail.

The auditor has given the team 5 business days to reconstruct this information or the model will be taken out of production.

Available resources:

MLflow: model artifact registered with trained_at: 2024-03-14, mlflow_run_id: abc123
Airflow: full execution logs going back 12 months (task-level, not column-level)
S3: no original training snapshot (overwritten), but S3 object version history enabled → 47 object versions of training_data/latest.parquet
Git: full commit history for the training pipeline code
Slack: searchable message history (the ML team uses a #model-deployments channel)
Feature store: Feast feature view versions registered in Git (8 months of history)

Your task:

Part A (Reconstruction plan). Write a step-by-step reconstruction plan for producing the audit trail. For each piece of evidence, describe: what you are looking for, where you will find it, and what you will do if it is missing. Include:

Identifying the training dataset (which S3 object version corresponds to trained_at: 2024-03-14)
Reconstructing the schema versions at training time (from Git history of the schema YAML files)
Identifying the deployment approver (from Slack #model-deployments history + git tag)
Computing the training dataset hash retroactively (from the S3 object version)

Part B (Gap analysis). List every piece of audit information that you cannot reconstruct reliably, and for each, describe what catalog infrastructure (if it had existed) would have made reconstruction unnecessary.

Part C (30-day prevention plan). Write a prioritized 5-task plan to implement the minimum viable audit trail infrastructure so this situation never recurs. Tasks should be implementable in 30 days with a 2-person team. For each task, estimate days of effort and the audit question it addresses.

Part D (Template for future deployments). Write a deployment checklist template (as a GitHub PR description template or a YAML file) that the ML team will complete for every future model deployment. The template must capture all information required to answer the auditor's question in < 5 minutes.

Interview Question 1: Explaining data lineage to an engineering manager

(Paid | XP: 20)

Question: "Our ML team wants to spend 2 months building a data catalog and lineage system. I understand why, conceptually, but I need to justify this to the VP of Engineering. What's the concrete business case? Give me 3 scenarios where lineage prevents a real incident, and tell me how much each incident would cost without it vs. with it."

What makes a strong answer:

Avoid abstract arguments ("lineage is important for debugging"). The VP of Engineering responds to concrete cost comparisons.
Use realistic numbers: engineering time costs roughly $100–$250/hour fully loaded for senior engineers; production ML incidents can cost $100K–$1M+ in revenue impact
Cover three distinct failure modes that lineage solves, not three variants of the same failure
Acknowledge the build cost: don't pretend a catalog is free; make the ROI case explicit

Your task: Write the verbal answer to this question as you would deliver it in a 10-minute conversation with an engineering manager. Then, separately, provide three specific incident scenarios with:

What happened (1 sentence)
What lineage would have shown (1 sentence)
Estimated cost without lineage (investigation time + production impact)
Estimated cost with lineage (time to identify the issue)

Scenario types to cover (use different ones than each other):

An upstream schema change that silently degraded a model
A compliance/audit request that required reconstructing training data provenance
A data deletion request (GDPR) that required identifying all copies of a user's data across derived datasets

Strong answer anchor: The three scenarios should have combined costs of $200K–$500K without lineage and $5K–$15K with lineage. An investment of 2 engineer-months (about 320 hours) at $200/hr is roughly $65K. This is how you make the ROI case concrete.

Interview Question 2: System design: data catalog for a 30-model ML platform

(Paid | XP: 25)

Question: "We're scaling from 5 models to 30 models over the next 18 months. Three different teams will be building models. How would you design the data catalog and metadata system to support this scale? What would you build vs. adopt? Where would you start?"

What makes a strong answer:

Address both the tooling and the organizational dimensions: a catalog platform without adoption processes is worthless
Be specific about the architecture generation: Gen 1/2/3, and why
Sequence the work correctly: what to build in month 1 vs. month 6 vs. month 18
Identify the hardest parts: not the technology, but the adoption and governance
Call out what you would NOT do: avoiding common mistakes is as important as naming the right approach

Your task: Write a structured answer that covers:

Part A (Architecture choice): For a 30-model, 3-team ML platform with ~300 datasets, which catalog platform would you choose (DataHub, OpenMetadata, Amundsen, or custom)? What are the deciding factors? What would make you choose differently?

Part B, What to build in Phase 1 (month 1–3): Describe the minimum viable catalog infrastructure. What does it include, what does it explicitly exclude, and why? What does "good enough for now" look like?

Part C, Phase 2 (month 3–9): What do you add next? How does the catalog start driving behavior (automation, gates) rather than just documenting it?

Part D (The hardest problem): What is the single hardest part of building a data catalog at this scale, harder than the technology? How do you address it?

Part E (What NOT to do): Name two approaches you would explicitly avoid and why. (Examples: starting with a full DataHub deployment on Day 1; letting each team maintain their own catalog independently.)

Interview Question 3: Metadata-driven governance vs. process-driven governance

(Paid | XP: 20)

Question: "Your team wants to adopt DataHub with the MCE/MAE event-sourced architecture. A senior engineer argues: 'The catalog should be read-only documentation. Production pipelines should never depend on catalog metadata at runtime; that creates a single point of failure.' How do you respond?"

What makes a strong answer:

Take the concern seriously: the senior engineer is not wrong about the failure mode; acknowledge it
Distinguish between different types of catalog runtime dependency: some are low-risk (schema validation at startup), some are higher-risk (serving pipelines querying catalog at request time)
Provide concrete examples of both the failure mode they're worried about and the failure mode you're preventing
Arrive at a nuanced recommendation rather than a binary "catalog should / should not influence production"

Your task: Write the response as a technical discussion, not a debate where you "win." Structure it as:

Acknowledge the valid concern: what is the failure mode the senior engineer is describing? Under what conditions would it happen?
Distinguish three levels of catalog runtime dependency: from safest to riskiest:
- Level 1: Catalog validates pipeline configuration at startup (low risk, fails fast, no cascading failure)
- Level 2: Catalog gates pipeline execution (medium risk: failure blocks the pipeline, not serving)
- Level 3: Serving pipeline queries catalog at request time (high risk, catalog outage = serving degradation)
Provide the concrete case for each level: give an example of what each level enables that is worth the risk
Make the recommendation: which levels are appropriate for ShopSense at its current scale? What operational safeguards make Level 2 acceptable? When (if ever) is Level 3 justified?

Advanced Exercise 1: Design a complete catalog entry for the ShopSense demand forecasting dataset

Advanced Exercise 2: Impact analysis: trace the product_category rename

Advanced Exercise 3: Lineage audit: trace a production regression backward

Real-World Implementation 1: Schema registry validator

Real-World Implementation 2: OpenLineage event emitter for pandas transforms

Production Challenge 1: The schema version divergence incident

Production Challenge 2: Retroactive catalog reconstruction

Interview Question 1: Explaining data lineage to an engineering manager

Interview Question 2: System design: data catalog for a 30-model ML platform

Interview Question 3: Metadata-driven governance vs. process-driven governance

Advanced Exercise 2: Impact analysis: trace the `product_category` rename