Implementing File-Based Ingestion with Schema Inference and Drift Detection: Automatically Adapting Pipelines to Evolving Source File Structures

Introduction

You've built a solid ingestion pipeline for the daily sales export your partner sends every morning. The CSV drops into an S3 bucket, your pipeline picks it up, validates a few rows, loads it into Snowflake, and life is good. Then, three months later, the pipeline crashes at 3 AM. The on-call alert fires. You dig in and find that the upstream team quietly added four new columns, renamed customer_id to cust_id, and dropped the region field entirely — all without telling anyone. Your schema is now a lie, and your downstream dashboard is broken.

This scenario plays out constantly in real data engineering work. File-based ingestion — reading CSVs, JSON dumps, Parquet exports, Excel files from vendors, partners, or internal tools — is ubiquitous precisely because files are the universal language of data handoffs. But files don't come with contracts, and the people generating them don't always understand (or care) what happens when they change the structure. The gap between "the file arrived" and "the file looks like what we expected" is where most file ingestion failures live.

By the end of this lesson, you'll know how to build ingestion pipelines that actively detect when a file's structure has changed, categorize that change as safe or dangerous, and respond intelligently — adapting automatically in some cases, alerting and halting in others. You'll work through a realistic implementation in Python that handles schema inference, schema registry persistence, drift classification, and conditional recovery.

What you'll learn:

• How to infer and serialize schemas from raw files at ingest time
• How to store and version schemas so you can detect drift across runs
• How to classify schema changes as additive (safe), subtractive (dangerous), or destructive (breaking)
• How to build automated responses: auto-adapt for safe changes, quarantine or halt for breaking ones
• How to integrate drift detection into an orchestrated pipeline with actionable alerting

---

Prerequisites

You should be comfortable with Python data manipulation using pandas, have a working understanding of what a data pipeline is and how its components fit together, and have ideally built at least one ingestion pipeline — if not, building your first data pipeline with Python will close that gap quickly. Familiarity with file storage (local or cloud) and basic JSON handling is assumed. You don't need to know Spark or Kafka for this lesson.

---

Understanding Schema Drift in the Wild

Before writing a line of code, it's worth being precise about what "schema drift" means and why a one-size-fits-all response is a mistake.

Schema drift is any deviation between the structure of a new file and the structure your pipeline expects based on a previously established baseline. But not all deviations are equal. Consider these categories:

Additive changes — a new column appears that didn't exist before. If your downstream system can handle nullable fields gracefully, this might be totally fine. The new column carries additional data you didn't have before, and your existing logic still works for everything it touched before.

Subtractive changes — a column you depended on has disappeared. This is almost always a problem. Any transformation, join, aggregation, or reporting that referenced that column will now fail or silently produce wrong results.

Rename changes — a column that existed under one name now appears under a different name. From the pipeline's perspective, this looks simultaneously like a deletion and an addition. It's among the trickiest changes to automatically detect and handle.

Type changes — a column that was int64 is now object (string), or what was a date is now a free-text timestamp. These are insidious because the column is still there; your pipeline won't throw a KeyError, but your downstream aggregations will silently compute garbage.

Order changes — column order has shifted. This is mostly harmless if you're accessing by column name, but it will silently destroy pipelines that consume positional CSV formats without headers.

For deeper context on how these changes manifest across a pipeline's lifetime, the article on schema evolution strategies for production data pipelines is excellent reading alongside this lesson.

---

Inferring and Capturing Schemas at Ingest Time

The foundation of drift detection is a reliable, deterministic schema capture step that runs every time a file is ingested. The captured schema becomes your fingerprint for that file's structure.

For CSV and JSON files, pandas gives you a reasonable starting point. For production-grade work, you'll want to augment the inferred types with nullability, cardinality signals, and format hints. Here's a schema capture function that does that:

import pandas as pd
import json
import hashlib
from datetime import datetime, timezone
from pathlib import Path


def infer_schema(filepath: str, file_format: str = "csv", sample_rows: int = 1000) -> dict:
    """
    Infer schema from a file and return a structured schema document.
    
    Args:
        filepath: Path to the source file.
        file_format: 'csv', 'json', or 'parquet'
        sample_rows: Number of rows to sample for type inference.
    
    Returns:
        A schema dict with column metadata and file-level stats.
    """
    if file_format == "csv":
        df = pd.read_csv(filepath, nrows=sample_rows)
    elif file_format == "json":
        df = pd.read_json(filepath, lines=True, nrows=sample_rows)
    elif file_format == "parquet":
        df = pd.read_parquet(filepath).head(sample_rows)
    else:
        raise ValueError(f"Unsupported format: {file_format}")

    columns = {}
    for col in df.columns:
        series = df[col]
        columns[col] = {
            "dtype": str(series.dtype),
            "nullable": bool(series.isna().any()),
            "null_rate": round(float(series.isna().mean()), 4),
            "unique_count": int(series.nunique()),
            "sample_values": [str(v) for v in series.dropna().head(3).tolist()]
        }

    schema = {
        "captured_at": datetime.now(timezone.utc).isoformat(),
        "source_file": str(filepath),
        "row_count_sample": len(df),
        "column_count": len(df.columns),
        "columns": columns,
        "column_order": list(df.columns)
    }

    return schema


def schema_fingerprint(schema: dict) -> str:
    """
    Generate a stable hash of the schema's structural elements (names + types).
    Ignores runtime metadata like capture time and sample values.
    """
    structural = {
        col: meta["dtype"] 
        for col, meta in schema["columns"].items()
    }
    canonical = json.dumps(structural, sort_keys=True)
    return hashlib.sha256(canonical.encode()).hexdigest()[:16]

Notice a few intentional design decisions here. We sample rather than read the entire file — for large files, reading a thousand rows is enough to establish column names and types, and reading everything just to capture the schema is wasteful. We store sample values so that when you're reviewing a schema change later, you can immediately see what the data in a new column actually looks like. And the fingerprint function only hashes column names and types, ignoring timestamps and row counts, so the same structural schema always produces the same fingerprint even if the file changes.

---

Building a Schema Registry

Detecting drift requires comparing the current file's schema against something. That "something" is your schema registry — a persistent store of previously approved schemas indexed by source identifier (the logical name of the feed, not the filename, since filenames usually change daily).

For a production system, a schema registry might live in PostgreSQL, DynamoDB, or a dedicated tool like Confluent Schema Registry or AWS Glue Data Catalog. For this lesson, we'll implement a JSON-file-based registry that you can drop into any project without infrastructure dependencies — the same logic applies if you swap the storage backend.

import os

REGISTRY_DIR = Path("schema_registry")


def get_registry_path(feed_name: str) -> Path:
    """Return the path to the registry file for a named feed."""
    return REGISTRY_DIR / f"{feed_name}.json"


def load_registered_schema(feed_name: str) -> dict | None:
    """
    Load the most recently registered (approved) schema for a feed.
    Returns None if no schema has been registered yet.
    """
    registry_path = get_registry_path(feed_name)
    if not registry_path.exists():
        return None
    with open(registry_path, "r") as f:
        return json.load(f)


def register_schema(feed_name: str, schema: dict, reason: str = "initial_registration") -> None:
    """
    Persist a schema as the approved baseline for a feed.
    
    Args:
        feed_name: Logical name for the data feed (e.g., 'partner_sales_daily')
        schema: The schema dict returned by infer_schema()
        reason: Human-readable reason for registration (for audit trail)
    """
    REGISTRY_DIR.mkdir(parents=True, exist_ok=True)
    registry_path = get_registry_path(feed_name)

    # Keep history alongside the current schema
    history_entry = {
        "registered_at": datetime.now(timezone.utc).isoformat(),
        "reason": reason,
        "fingerprint": schema_fingerprint(schema),
        "schema": schema
    }

    # Load existing history if any
    if registry_path.exists():
        with open(registry_path, "r") as f:
            existing = json.load(f)
        history = existing.get("history", [])
    else:
        history = []

    history.append(history_entry)

    registry_doc = {
        "feed_name": feed_name,
        "current_schema": schema,
        "current_fingerprint": schema_fingerprint(schema),
        "last_registered_at": history_entry["registered_at"],
        "history": history
    }

    with open(registry_path, "w") as f:
        json.dump(registry_doc, f, indent=2)

    print(f"[registry] Schema registered for '{feed_name}' "
          f"(fingerprint: {schema_fingerprint(schema)})")

The history list is worth calling out explicitly. You're not just storing the current schema — you're keeping a chain of every previously approved schema with timestamps and reasons. When you're debugging a production incident six weeks from now and trying to figure out when a column went missing, that history is invaluable. This integrates naturally with logging and observability practices for data pipelines.

---

Implementing Drift Classification

Now for the core logic: given a registered schema and a newly inferred schema, what changed, and how serious is it?

from dataclasses import dataclass, field
from enum import Enum


class DriftSeverity(Enum):
    NONE = "none"
    ADDITIVE = "additive"       # New columns appeared — often safe
    SUBTRACTIVE = "subtractive" # Columns disappeared — usually breaking
    TYPE_CHANGE = "type_change" # Column type changed — potentially silent corruption
    RENAME = "rename"           # Likely a rename — requires human judgment
    REORDER = "reorder"         # Column order changed — low risk if named access
    BREAKING = "breaking"       # Multiple destructive changes — halt


@dataclass
class DriftReport:
    feed_name: str
    old_fingerprint: str
    new_fingerprint: str
    is_drifted: bool
    severity: DriftSeverity
    added_columns: list = field(default_factory=list)
    removed_columns: list = field(default_factory=list)
    type_changed_columns: dict = field(default_factory=dict)
    reordered: bool = False
    summary: str = ""
    recommended_action: str = ""


def classify_drift(
    feed_name: str,
    registered_schema: dict,
    new_schema: dict
) -> DriftReport:
    """
    Compare two schemas and return a structured drift report.
    """
    old_cols = registered_schema["columns"]
    new_cols = new_schema["columns"]
    old_fp = schema_fingerprint(registered_schema)
    new_fp = schema_fingerprint(new_schema)

    if old_fp == new_fp:
        return DriftReport(
            feed_name=feed_name,
            old_fingerprint=old_fp,
            new_fingerprint=new_fp,
            is_drifted=False,
            severity=DriftSeverity.NONE,
            summary="No schema drift detected.",
            recommended_action="proceed"
        )

    added = [c for c in new_cols if c not in old_cols]
    removed = [c for c in old_cols if c not in new_cols]
    type_changes = {
        col: {"old": old_cols[col]["dtype"], "new": new_cols[col]["dtype"]}
        for col in old_cols
        if col in new_cols and old_cols[col]["dtype"] != new_cols[col]["dtype"]
    }
    old_order = registered_schema.get("column_order", list(old_cols.keys()))
    new_order = new_schema.get("column_order", list(new_cols.keys()))
    shared_in_old_order = [c for c in old_order if c in new_cols]
    shared_in_new_order = [c for c in new_order if c in old_cols]
    reordered = shared_in_old_order != shared_in_new_order

    # Heuristic: if removed and added counts match and are small, suspect rename
    possible_rename = (
        len(removed) == len(added) 
        and 1 <= len(removed) <= 3 
        and not type_changes
    )

    # Determine severity
    if removed and type_changes:
        severity = DriftSeverity.BREAKING
        action = "halt"
        summary = (
            f"Breaking drift: {len(removed)} column(s) removed AND "
            f"{len(type_changes)} type change(s). Manual review required."
        )
    elif removed:
        if possible_rename:
            severity = DriftSeverity.RENAME
            action = "alert_and_pause"
            summary = (
                f"Possible rename detected: removed {removed}, added {added}. "
                "Cannot auto-resolve — human confirmation needed."
            )
        else:
            severity = DriftSeverity.SUBTRACTIVE
            action = "halt"
            summary = f"Subtractive drift: columns {removed} no longer present in source."
    elif type_changes:
        severity = DriftSeverity.TYPE_CHANGE
        action = "halt"
        summary = (
            f"Type change drift: {list(type_changes.keys())} changed types. "
            "Downstream queries may silently fail."
        )
    elif added:
        severity = DriftSeverity.ADDITIVE
        action = "auto_adapt"
        summary = (
            f"Additive drift: {len(added)} new column(s) detected: {added}. "
            "Safe to proceed with schema update."
        )
    elif reordered:
        severity = DriftSeverity.REORDER
        action = "auto_adapt"
        summary = "Column order changed. No structural change to column names or types."
    else:
        severity = DriftSeverity.NONE
        action = "proceed"
        summary = "Minor metadata change detected, no structural drift."

    return DriftReport(
        feed_name=feed_name,
        old_fingerprint=old_fp,
        new_fingerprint=new_fp,
        is_drifted=True,
        severity=severity,
        added_columns=added,
        removed_columns=removed,
        type_changed_columns=type_changes,
        reordered=reordered,
        summary=summary,
        recommended_action=action
    )

The severity classification here is opinionated, and intentionally so. You might have a feed where downstream consumers are known to handle nullable columns gracefully, in which case additive drift is completely safe. You might have another feed where even a column reorder breaks a legacy consumer that reads positionally. These defaults are sane starting points — your job is to tune them per feed.

---

Building the Adaptive Ingestion Pipeline

Now we assemble all the pieces into a complete ingestion function that handles the full lifecycle: infer, compare, classify, respond.

import shutil
from typing import Callable


QUARANTINE_DIR = Path("quarantine")
PROCESSED_DIR = Path("processed")


def adaptive_ingest(
    filepath: str,
    feed_name: str,
    file_format: str = "csv",
    load_fn: Callable[[pd.DataFrame, str], None] = None,
    alert_fn: Callable[[DriftReport], None] = None,
    sample_rows: int = 5000
) -> dict:
    """
    Full adaptive ingestion with schema inference, drift detection, and routing.

    Args:
        filepath: Path to the source file.
        feed_name: Logical name of the data feed.
        file_format: File format ('csv', 'json', 'parquet').
        load_fn: Function that accepts (DataFrame, feed_name) and loads data.
        alert_fn: Function that accepts a DriftReport and sends an alert.
        sample_rows: Rows to sample for schema inference.

    Returns:
        A result dict summarizing what happened.
    """
    filepath = Path(filepath)
    result = {
        "feed_name": feed_name,
        "file": str(filepath),
        "status": None,
        "action_taken": None,
        "drift_severity": None,
        "timestamp": datetime.now(timezone.utc).isoformat()
    }

    # Step 1: Infer schema from new file
    print(f"[ingest] Inferring schema from {filepath.name}...")
    new_schema = infer_schema(str(filepath), file_format=file_format, sample_rows=sample_rows)

    # Step 2: Load registered schema (if any)
    registry_doc = load_registered_schema(feed_name)

    if registry_doc is None:
        # First time we've seen this feed — register and proceed
        print(f"[ingest] No existing schema for '{feed_name}'. Registering as baseline.")
        register_schema(feed_name, new_schema, reason="initial_registration")
        drift_report = None
        action = "proceed_first_run"
    else:
        # Step 3: Classify drift
        registered_schema = registry_doc["current_schema"]
        drift_report = classify_drift(feed_name, registered_schema, new_schema)
        action = drift_report.recommended_action
        result["drift_severity"] = drift_report.severity.value

        print(f"[ingest] Drift check: {drift_report.severity.value} — {drift_report.summary}")

    # Step 4: Route based on drift classification
    if action in ("proceed", "proceed_first_run"):
        # No drift or first run — load normally
        df = _read_file(filepath, file_format)
        if load_fn:
            load_fn(df, feed_name)
        result["status"] = "success"
        result["action_taken"] = "loaded_normally"

    elif action == "auto_adapt":
        # Safe drift (additive or reorder) — update registry and load
        print(f"[ingest] Auto-adapting schema for '{feed_name}'...")
        register_schema(
            feed_name, 
            new_schema, 
            reason=f"auto_adapt_{drift_report.severity.value}"
        )
        df = _read_file(filepath, file_format)
        if load_fn:
            load_fn(df, feed_name)
        result["status"] = "success"
        result["action_taken"] = "auto_adapted_and_loaded"
        if alert_fn and drift_report:
            alert_fn(drift_report)  # Notify even on safe changes

    elif action == "alert_and_pause":
        # Possible rename — quarantine and alert, do not load
        _quarantine_file(filepath, feed_name, drift_report)
        if alert_fn and drift_report:
            alert_fn(drift_report)
        result["status"] = "paused"
        result["action_taken"] = "quarantined_pending_review"

    elif action == "halt":
        # Dangerous drift — quarantine and alert
        _quarantine_file(filepath, feed_name, drift_report)
        if alert_fn and drift_report:
            alert_fn(drift_report)
        result["status"] = "halted"
        result["action_taken"] = "quarantined_schema_violation"
        raise SchemaViolationError(
            f"Breaking schema drift detected for '{feed_name}': {drift_report.summary}"
        )

    return result


def _read_file(filepath: Path, file_format: str) -> pd.DataFrame:
    """Read a file into a DataFrame based on format."""
    if file_format == "csv":
        return pd.read_csv(filepath)
    elif file_format == "json":
        return pd.read_json(filepath, lines=True)
    elif file_format == "parquet":
        return pd.read_parquet(filepath)
    raise ValueError(f"Unsupported format: {file_format}")


def _quarantine_file(filepath: Path, feed_name: str, drift_report: DriftReport) -> None:
    """Move a file to the quarantine directory with a drift report artifact."""
    QUARANTINE_DIR.mkdir(parents=True, exist_ok=True)
    ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S")
    dest = QUARANTINE_DIR / f"{feed_name}__{ts}__{filepath.name}"
    shutil.move(str(filepath), str(dest))
    
    # Write drift report alongside quarantined file
    report_path = dest.with_suffix(".drift_report.json")
    with open(report_path, "w") as f:
        json.dump({
            "feed_name": drift_report.feed_name,
            "severity": drift_report.severity.value,
            "summary": drift_report.summary,
            "added_columns": drift_report.added_columns,
            "removed_columns": drift_report.removed_columns,
            "type_changes": drift_report.type_changed_columns,
            "old_fingerprint": drift_report.old_fingerprint,
            "new_fingerprint": drift_report.new_fingerprint
        }, f, indent=2)
    
    print(f"[ingest] File quarantined: {dest.name}")


class SchemaViolationError(Exception):
    """Raised when a file cannot be ingested due to breaking schema drift."""
    pass

The quarantine pattern here deserves emphasis. When a file is dangerous, you don't delete it and you don't load it — you park it in a holding area with a full drift report beside it. This gives your team a path to recovery: once you've updated your pipeline logic or confirmed the schema change is intentional, you can re-ingest from quarantine. This integrates cleanly with the dead letter queue pattern for handling problematic records.

---

Putting It Together: A Realistic End-to-End Example

Let's walk through a concrete scenario. Your company receives a daily partner export called partner_sales_daily — a CSV with columns like order_id, customer_id, product_sku, sale_amount, sale_date, store_code.

# Simulate your load function (in production this writes to a database)
def load_to_warehouse(df: pd.DataFrame, feed_name: str) -> None:
    print(f"[warehouse] Loading {len(df)} rows for feed '{feed_name}'...")
    # In production: df.to_sql(...) or use your cloud connector
    PROCESSED_DIR.mkdir(parents=True, exist_ok=True)
    output_path = PROCESSED_DIR / f"{feed_name}_{datetime.now().strftime('%Y%m%d')}.parquet"
    df.to_parquet(output_path, index=False)
    print(f"[warehouse] Loaded to {output_path}")


# Simulate your alert function (in production: Slack, PagerDuty, email)
def send_drift_alert(report: DriftReport) -> None:
    print(f"\n{'='*60}")
    print(f"SCHEMA DRIFT ALERT — {report.feed_name}")
    print(f"Severity: {report.severity.value.upper()}")
    print(f"Summary: {report.summary}")
    if report.added_columns:
        print(f"Added columns: {report.added_columns}")
    if report.removed_columns:
        print(f"Removed columns: {report.removed_columns}")
    if report.type_changed_columns:
        print(f"Type changes: {report.type_changed_columns}")
    print(f"Recommended action: {report.recommended_action}")
    print(f"{'='*60}\n")


# --- Day 1: First file arrives, no registered schema ---
result = adaptive_ingest(
    filepath="data/partner_sales_20240101.csv",
    feed_name="partner_sales_daily",
    file_format="csv",
    load_fn=load_to_warehouse,
    alert_fn=send_drift_alert
)
# Output: Registers schema, loads file, status = success

# --- Day 15: Partner adds two new analytics columns ---
# File now has: order_id, customer_id, product_sku, sale_amount, 
#               sale_date, store_code, promo_code, channel_source
result = adaptive_ingest(
    filepath="data/partner_sales_20240115.csv",
    feed_name="partner_sales_daily",
    file_format="csv",
    load_fn=load_to_warehouse,
    alert_fn=send_drift_alert
)
# Output: ADDITIVE drift detected, auto-adapts registry, loads file
# Alert sent (informational), status = success

# --- Day 30: Partner renames customer_id to cust_id ---
# File now has: order_id, cust_id, product_sku, sale_amount,
#               sale_date, store_code, promo_code, channel_source
result = adaptive_ingest(
    filepath="data/partner_sales_20240130.csv",
    feed_name="partner_sales_daily",
    file_format="csv",
    load_fn=load_to_warehouse,
    alert_fn=send_drift_alert
)
# Output: RENAME drift detected (1 removed, 1 added, same count)
# File quarantined with drift report, alert sent, status = paused

# --- Day 45: Partner drops region, changes sale_amount to string ---
# This would trigger BREAKING drift and raise SchemaViolationError

---

Integrating Drift Detection with Orchestration

A standalone ingestion function is useful, but the real value comes when it's embedded in an orchestrated pipeline where failures surface correctly, alerting is reliable, and quarantined files can be reprocessed. If you're running Airflow, here's how the ingestion task fits in:

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from datetime import timedelta


def ingest_partner_sales(**context):
    execution_date = context["ds"]  # e.g., '2024-01-15'
    filepath = f"/data/landing/partner_sales_{execution_date.replace('-', '')}.csv"
    
    try:
        result = adaptive_ingest(
            filepath=filepath,
            feed_name="partner_sales_daily",
            file_format="csv",
            load_fn=load_to_warehouse,
            alert_fn=send_drift_alert
        )
        context["ti"].xcom_push(key="ingest_result", value=result)
    except SchemaViolationError as e:
        # Airflow will mark this task as failed, triggering retry and alerting
        raise


with DAG(
    dag_id="partner_sales_ingestion",
    schedule_interval="0 6 * * *",
    start_date=days_ago(1),
    catchup=False,
    default_args={
        "retries": 1,
        "retry_delay": timedelta(minutes=15),
        "email_on_failure": True,
    }
) as dag:
    
    ingest_task = PythonOperator(
        task_id="ingest_partner_sales",
        python_callable=ingest_partner_sales,
    )

When SchemaViolationError propagates out of the task, Airflow marks it as failed, retries once (in case the file is a re-upload that actually fixed the issue), and then alerts on failure. The file is in quarantine. Your team gets a notification with the drift report. Everyone knows exactly what happened and what to do next.

---

Handling Reprocessing After Schema Issues

Once a human reviews a quarantined file and approves the new schema (or updates downstream dependencies to handle the change), you need a clean reprocessing path. Here's a utility that handles that:

def reprocess_quarantined_file(
    quarantined_filepath: str,
    feed_name: str,
    file_format: str = "csv",
    approve_new_schema: bool = False,
    load_fn: Callable = None,
    alert_fn: Callable = None
) -> dict:
    """
    Reprocess a quarantined file after human review.

    If approve_new_schema=True, the new schema is registered as the 
    approved baseline before loading. This is the human-in-the-loop 
    approval step.
    """
    quarantined_path = Path(quarantined_filepath)
    
    if not quarantined_path.exists():
        raise FileNotFoundError(f"Quarantined file not found: {quarantined_path}")

    if approve_new_schema:
        # Infer the schema from the quarantined file and register it
        new_schema = infer_schema(str(quarantined_path), file_format=file_format)
        register_schema(
            feed_name, 
            new_schema, 
            reason="manual_approval_after_review"
        )
        print(f"[reprocess] New schema approved and registered for '{feed_name}'.")

    # Now run ingest with the updated registry — drift should resolve
    return adaptive_ingest(
        filepath=str(quarantined_path),
        feed_name=feed_name,
        file_format=file_format,
        load_fn=load_fn,
        alert_fn=alert_fn
    )

This pattern ensures that schema approval is explicit and audited. You can't accidentally load a file with a breaking schema change — a human has to call reprocess_quarantined_file with approve_new_schema=True, and that action is logged in the registry history. This feeds naturally into data quality validation and monitoring processes where schema approval should be gated by downstream impact analysis.

---

Hands-On Exercise

You'll build a complete drift-detecting ingestion system for a realistic scenario: ingesting monthly customer churn exports from a CRM system that's been known to change its export format after major releases.

Setup: Create a directory structure:

project/
├── data/
│   ├── churn_export_2024_01.csv
│   ├── churn_export_2024_02.csv  (new column: engagement_score)
│   └── churn_export_2024_03.csv  (column churned renamed to is_churned)
├── schema_registry/
├── quarantine/
└── processed/

Create the first CSV manually with columns: customer_id, account_age_days, last_login_date, plan_tier, monthly_spend, churned.

Task 1: Run adaptive_ingest on churn_export_2024_01.csv with feed name crm_churn_monthly. Inspect the JSON file created in schema_registry/. What information does it store? What's the fingerprint?

Task 2: Create churn_export_2024_02.csv with the same schema plus a new engagement_score column (float). Run the ingestion. Confirm it detects additive drift, auto-adapts, and updates the registry. Open the registry file and confirm the history now has two entries.

Task 3: Create churn_export_2024_03.csv with churned renamed to is_churned. Run the ingestion and confirm that the pipeline correctly identifies this as a possible rename, quarantines the file, and writes a .drift_report.json alongside it. Read the drift report and verify its contents.

Task 4: Simulate approval of the new schema by calling reprocess_quarantined_file with approve_new_schema=True. Confirm the file loads successfully and the registry now shows three history entries.

Stretch goal: Modify the classify_drift function to accept a per-feed configuration dict that specifies which columns are "critical" (e.g., customer_id, churned). If any critical column is removed or has its type changed, escalate to BREAKING severity regardless of other conditions.

---

Common Mistakes & Troubleshooting

Mistake 1: Using filenames as the registry key instead of feed names. If your partner sends files named export_20240101.csv, export_20240102.csv, etc., using the filename as the registry key means you're comparing each day's file against nothing — you never actually detect drift. Always use a stable logical feed name that represents the data source, not a specific file instance.

Mistake 2: Trusting pandas dtype inference completely for schema registration. Pandas will infer int64 for a column of integers, but if the next file has one NULL value in that column, it'll infer float64 (because NaN is a float in Python). This creates false positives for type-change drift. Consider normalizing types during registration — mapping float64 to numeric, int64 to integer, object to string — so minor representation changes don't trigger alerts.

Mistake 3: Sampling too few rows. If you only sample 100 rows from a 10-million-row file, you might not encounter a column that only appears in certain record types (think JSON with sparse keys). Sample at least 1,000-5,000 rows, and for JSON files with known sparse structures, consider reading all unique keys explicitly.

Mistake 4: Treating all additive drift as automatically safe. An additive change is usually safe, but not always. If the new column has a 100% null rate in the first file and your downstream schema defines it as NOT NULL, auto-adapting will cause downstream failures. Add a nullability check to your auto-adapt logic: if a new column has more than a configurable null-rate threshold (say, 90%), alert a human before registering the new schema.

Mistake 5: Not handling partial files or truncated uploads. A file that arrives mid-upload will infer a schema from an incomplete dataset. Its row count will be suspiciously low, and columns might be missing or malformed. Add a file completeness check before schema inference — this might mean checking for a companion .done sentinel file, validating a row count against an expected range, or checking file size against a minimum threshold.

Mistake 6: Running schema inference and drift detection inside a transaction. If your ingestion is long-running and something fails halfway through loading, you want the schema change to roll back too, so the registry doesn't reflect a schema you didn't actually successfully load. Only call register_schema after the load completes successfully, not before.

---

Summary & Next Steps

You've built a file-based ingestion system that actively understands its own data — not just reading files blindly, but capturing their structure, comparing it against an approved baseline, classifying any deviation by severity, and routing accordingly. The key design principles at work here:

• Schema inference happens every run, not just at setup time. The pipeline never assumes today's file looks like yesterday's.
• Classification separates signal from noise. Additive drift is treated differently from type changes or column removal. Not all change is dangerous.
• Quarantine instead of delete. Files that fail schema validation are preserved for human review and reprocessing — no data is lost.
• The registry is an audit trail. Every schema state transition is logged with a timestamp and reason. This turns debugging from archaeology into reading a changelog.

From here, the natural directions to go deeper are:

1. Downstream impact analysis — When drift is detected, automatically query your lineage graph to identify which downstream pipelines, models, or dashboards depend on the changed columns. See understanding pipeline dependencies and lineage tracking for the foundations.

2. Connecting to production storage — Replace the JSON registry with a proper database or cloud metadata store, and read files from S3 or GCS rather than local disk. Connecting to databases, flat files, and cloud storage covers the mechanics.

3. Data contracts — The schema registry you've built here is a lightweight implementation of a data contract. For formal contract enforcement, testing, and versioning between teams, pipeline testing with data contracts takes this concept further.

4. Incremental processing — Once your ingestion is schema-safe, you'll want to process only new files, not re-process everything. Incremental loading patterns with timestamps, CDC, and watermarks covers exactly that.

The foundation you've built here — a schema-aware, drift-detecting ingestion layer — is one of the most practically valuable additions you can make to any file-based data pipeline. It transforms a fragile system that breaks silently into a resilient one that fails loudly, in the right places, with the right information.