Validate Scraped Data and Detect Schema Drift

Scraped data rarely fails loudly. It rots: a redesign moves a price into a different element, the parser returns None, and ten thousand rows with empty prices load into the warehouse without a single exception. This guide builds the layers that catch schema drift before it becomes a bad report: a pydantic v2 record schema, per-batch quality checks, a drift detector with a rolling baseline, a quarantine step, a daily selector contract test on canary URLs, and parser versioning so you can backfill once you fix the problem.

Why extracted data silently rots

Three sources of schema drift account for almost every incident. Site redesigns change the DOM, and a tolerant parser written with .get() and find() returns empties rather than raising. A/B layouts are worse because only a fraction of requests see the new template, so your fill rate drops from 99 percent to 70 percent and nobody notices for a week. Locale and geo changes alter formats without altering structure: 1,299.00 becomes 1.299,00, a date flips from ISO to day-first, and the value is parsed wrong rather than not at all.

The shared property is that the parser does not crash. Record-level tolerance is the right default, but it means correctness has to be enforced somewhere else. That somewhere is batch-level data validation in web scraping pipelines, and it is cheap to build.

Define a record schema with pydantic v2

Start by writing down what a valid record looks like. Required fields, types, ranges and formats become a pydantic model, and every parsed dict goes through model_validate before it goes anywhere else.

from __future__ import annotations

import re
from datetime import date, datetime
from typing import Optional

from pydantic import BaseModel, Field, ValidationError, field_validator

PRICE_RE = re.compile(r"^\d+(\.\d{1,2})?$")
SKU_RE = re.compile(r"^[A-Z0-9-]{4,32}$")

class ProductRecord(BaseModel):
    source_url: str = Field(min_length=10)
    sku: str
    title: str = Field(min_length=3, max_length=300)
    price: float = Field(gt=0, lt=100_000)
    currency: str = Field(pattern=r"^[A-Z]{3}$")
    in_stock: bool
    rating: Optional[float] = Field(default=None, ge=0, le=5)
    scraped_at: datetime
    listed_on: Optional[date] = None
    parser_version: str

    @field_validator("sku")
    @classmethod
    def sku_shape(cls, v: str) -> str:
        v = v.strip().upper()
        if not SKU_RE.match(v):
            raise ValueError(f"sku does not match pattern: {v!r}")
        return v

    @field_validator("price", mode="before")
    @classmethod
    def normalise_price(cls, v):
        # Accept "1,299.00" or "1299" from the parser; reject anything else.
        if isinstance(v, str):
            cleaned = v.replace(",", "").replace("$", "").strip()
            if not PRICE_RE.match(cleaned):
                raise ValueError(f"price not parseable: {v!r}")
            return float(cleaned)
        return v

    @field_validator("title")
    @classmethod
    def title_not_placeholder(cls, v: str) -> str:
        if v.strip().lower() in {"", "n/a", "untitled", "loading..."}:
            raise ValueError("title looks like a placeholder")
        return v.strip()

def validate_batch(rows: list[dict]) -> tuple[list[ProductRecord], list[dict]]:
    good: list[ProductRecord] = []
    bad: list[dict] = []
    for row in rows:
        try:
            good.append(ProductRecord.model_validate(row))
        except ValidationError as e:
            bad.append({"row": row, "errors": e.errors(include_url=False)})
    return good, bad

Two choices here matter. mode="before" on the price validator lets the schema do the last bit of normalisation so the parser can stay simple. And e.errors() returns structured errors with a loc tuple, which is what lets you count failures per field rather than per record. A batch where 40 percent of rows fail on price and nothing else is a different problem from one where every row fails on every field.

Per-batch checks

Validation rejects rows. Batch checks look at the rows that passed and the ones that did not, together, and ask whether the batch as a whole is plausible. Four checks catch most schema drift.

from collections import Counter
from statistics import median

def batch_metrics(good: list[ProductRecord], bad: list[dict], fields: list[str]) -> dict:
    total = len(good) + len(bad)
    failures: Counter[str] = Counter()
    for b in bad:
        for err in b["errors"]:
            if err["loc"]:
                failures[str(err["loc"][0])] += 1

    fill_rate = {}
    for f in fields:
        present = sum(1 for g in good if getattr(g, f, None) is not None)
        fill_rate[f] = present / total if total else 0.0

    skus = [g.sku for g in good]
    dup_rate = 1 - (len(set(skus)) / len(skus)) if skus else 0.0
    prices = [g.price for g in good]

    return {
        "record_count": total,
        "reject_rate": len(bad) / total if total else 0.0,
        "field_failures": dict(failures),
        "fill_rate": fill_rate,
        "duplicate_rate": dup_rate,
        "price_median": median(prices) if prices else None,
    }

def count_in_range(count: int, expected: int, tolerance: float = 0.20) -> bool:
    return abs(count - expected) <= expected * tolerance

Persist batch_metrics for every run. It is the raw material for the drift detector, and it is also what you will look at first when someone asks why Tuesday's numbers look odd.

A drift detector with a rolling baseline

Fixed thresholds break because sources differ. A field that is 60 percent filled on one site is normal; on another it is a crisis. Compare against the source's own recent history instead: a rolling baseline of the last N good batches.

from dataclasses import dataclass, field

@dataclass
class DriftAlert:
    field_name: str
    baseline: float
    current: float
    drop_points: float

@dataclass
class DriftDetector:
    window: int = 14
    drop_threshold: float = 10.0  # percentage points
    critical_fields: dict[str, float] = field(default_factory=lambda: {"price": 5.0})
    history: list[dict] = field(default_factory=list)

    def baseline_fill(self, field_name: str) -> float | None:
        rates = [m["fill_rate"].get(field_name) for m in self.history[-self.window:]]
        rates = [r for r in rates if r is not None]
        return median(rates) if rates else None

    def check(self, metrics: dict, expected_count: int) -> list[DriftAlert]:
        alerts: list[DriftAlert] = []
        for name, current in metrics["fill_rate"].items():
            base = self.baseline_fill(name)
            if base is None:
                continue
            drop = (base - current) * 100
            limit = self.critical_fields.get(name, self.drop_threshold)
            if drop >= limit:
                alerts.append(DriftAlert(name, base, current, drop))
        if not count_in_range(metrics["record_count"], expected_count):
            alerts.append(DriftAlert("record_count", expected_count, metrics["record_count"], 0.0))
        if metrics["duplicate_rate"] > 0.02:
            alerts.append(DriftAlert("duplicate_rate", 0.0, metrics["duplicate_rate"], 0.0))
        return alerts

    def accept(self, metrics: dict) -> None:
        # Only batches that passed become part of the baseline.
        self.history.append(metrics)

The important line is accept: only clean batches feed the baseline. If you let a drifted batch in, the baseline follows the drift and the detector goes quiet exactly when it should be loud. Use the median rather than the mean for the same reason; one odd day should not move the reference.

Quarantine instead of load

When the detector raises, do not load the batch, and do not discard it either. Write it to a quarantine location with its metrics and alerts attached, send the alert, and let a human or an automated rule decide. The decision is usually one of three: false alarm (promote the batch and accept it into the baseline), real drift (fix the parser, re-run from stored raw content), or source outage (drop the batch, keep the baseline untouched).

import json
from pathlib import Path

def route_batch(batch_id: str, good, bad, metrics, alerts, detector: DriftDetector) -> str:
    if alerts:
        q = Path("quarantine") / batch_id
        q.mkdir(parents=True, exist_ok=True)
        (q / "metrics.json").write_text(json.dumps(metrics, default=str, indent=2))
        (q / "alerts.json").write_text(json.dumps([a.__dict__ for a in alerts], indent=2))
        (q / "rows.jsonl").write_text(
            "\n".join(g.model_dump_json() for g in good)
        )
        notify(f"batch {batch_id} quarantined: " + ", ".join(a.field_name for a in alerts))
        return "quarantined"
    load_to_warehouse(good)
    detector.accept(metrics)
    return "loaded"

def notify(message: str) -> None:
    # Wire to Slack, PagerDuty, email, or a log sink.
    print("ALERT:", message)

def load_to_warehouse(records) -> None:
    pass

Alert on what changed, not on "validation failed". A message that says price fill rate 0.98 -> 0.61 on source X is actionable; a generic failure is not.

The selector contract test on canary URLs

Batch checks tell you something drifted after it drifted. A selector contract test tells you earlier, often before the daily run. Pick a fixed set of canary URLs per source, ones you know the expected values for, fetch them through POST /v1/extract every day, run the parser, and assert that each contracted field is present and plausible. When the contract fails, the parser needs attention before the big run starts.

import os
import httpx

BASE = "https://scraper.geonode.io"
HEADERS = {"X-Api-Key": os.environ["GEONODE_SCRAPER_API_KEY"]}

CANARIES = [
    {"url": "https://example.com/product/123", "expect": {"sku": "EX-123", "currency": "USD"}},
    {"url": "https://example.com/product/456", "expect": {"sku": "EX-456", "currency": "USD"}},
]

CONTRACT_FIELDS = ["sku", "title", "price", "currency", "in_stock"]

def fetch_markdown(url: str) -> str:
    r = httpx.post(
        f"{BASE}/v1/extract",
        headers=HEADERS,
        json={
            "url": url,
            "formats": ["markdown"],
            "render_js": True,
            "processing_mode": "sync",
            "proxy": {"country": "US", "type": "residential"},
        },
        timeout=90.0,
    )
    r.raise_for_status()
    return r.json()["data"]["markdown"]

def parse_product(markdown: str) -> dict:
    # Your parser lives here; it should return a plain dict for the schema.
    raise NotImplementedError

def run_contract(parser_version: str) -> list[str]:
    failures: list[str] = []
    for c in CANARIES:
        md = fetch_markdown(c["url"])
        row = parse_product(md) | {"source_url": c["url"], "parser_version": parser_version}
        try:
            rec = ProductRecord.model_validate(row)
        except ValidationError as e:
            failures.append(f"{c['url']}: {[x['loc'] for x in e.errors()]}")
            continue
        for k, v in c["expect"].items():
            if getattr(rec, k) != v:
                failures.append(f"{c['url']}: {k} expected {v!r} got {getattr(rec, k)!r}")
    return failures

if __name__ == "__main__":
    problems = run_contract(parser_version="2026.10.1")
    if problems:
        notify("selector contract failed:\n" + "\n".join(problems))
        raise SystemExit(1)

Five to ten canaries per source is enough. Pin the proxy country if the site localises, because a canary fetched from a different region tests a different page. Choose URLs that exercise the layouts you care about: one in-stock, one out-of-stock, one on sale, one with no rating.

Version the parser and store the raw page

Every row should carry parser_version, and the version should change whenever the parser logic changes. That gives you two things. First, you can slice any metric by version and see immediately whether a drift alert coincides with a deploy. Second, when you fix a parser you know exactly which rows to regenerate.

Regeneration is only possible if you kept the input. Store the raw HTML or markdown from every fetch, compressed, keyed by source_url and scraped_at. The storage cost is small next to the cost of re-scraping, and re-scraping is often not even an option because the page has changed again. When the schema drift is confirmed, the workflow is: fix the parser, bump the version, re-parse the stored raw content for the affected window, validate it through the same schema, and load it to replace the quarantined batches. Without raw storage, the gap is permanent.

import gzip
import hashlib
from pathlib import Path

def store_raw(source_url: str, scraped_at: datetime, content: str, fmt: str = "md") -> Path:
    key = hashlib.sha256(source_url.encode()).hexdigest()[:16]
    p = Path("raw") / scraped_at.strftime("%Y/%m/%d") / f"{key}.{fmt}.gz"
    p.parent.mkdir(parents=True, exist_ok=True)
    with gzip.open(p, "wt", encoding="utf-8") as f:
        f.write(content)
    return p

Common mistakes

Validating per record and never per batch

A record schema catches malformed rows; it cannot catch a batch where every row is well-formed and half the prices are gone. Always compute fill rates and counts at the batch level.

Letting bad batches into the baseline

If the rolling baseline includes a drifted batch, the next drifted batch looks normal. Only accept batches that passed into the history, and use a median.

Making the parser too tolerant

.get("price") that returns None is fine, but a parser that returns 0.0 or an empty string on failure defeats the fill-rate check. Return None for missing, and let the schema reject it.

Not pinning the fetch conditions

Canary tests and production runs must use the same rendering and proxy settings. A contract test that passes with render_js: true and a US proxy proves nothing about a run that uses neither.

Next step

All of the fetches in this guide go through the Geonode Scraper API, which returns clean markdown or HTML from a single POST /v1/extract call with JavaScript rendering and country-level proxy targeting when you need it. Keeping the fetch layer stable removes one variable from schema drift investigations, because what changed is the page, not how you got it. The free tier covers 1.5K requests a month, enough to run canary contracts against several sources daily while you build out the validation pipeline.

FAQ

What is schema drift in web scraping?

Schema drift is the gradual or sudden change in the shape of the data your parser produces, caused by the source site changing rather than your code. A redesign moves a price into a new element, an A/B test renders a different layout for some requests, or a locale switch changes the number format. The parser keeps running, but fields go empty or wrong.

Why does scraped data rot silently instead of failing loudly?

Most parsers are written to be tolerant, so a missing element returns None instead of raising. That is sensible per record and disastrous per batch, because ten thousand records with a null price load without error. You detect it only when a downstream report looks wrong, often weeks later. Batch-level validation and drift checks close that gap.

How do I validate scraped data with pydantic?

Define a model per record type with required fields, types, numeric ranges and regex constraints, then call model_validate on each parsed dict inside a try block that catches ValidationError. Count failures per field across the batch. A high rejection rate on one field is a schema drift signal; a high rate everywhere usually means the fetch itself failed.

What threshold should trigger a schema drift alert?

Compare each field's fill rate in the current batch against a rolling baseline of recent good batches. A drop of ten percentage points or more on a required field is a strong signal; use five for high-value fields like price. Also check record count against expectation within plus or minus 20 percent and watch duplicate rate. Tune thresholds per source after a week of data.

Should I store raw HTML or markdown from every scrape?

Yes, if storage allows. Raw HTML or markdown is the only thing that lets you backfill after you fix a parser, because the page as it was at scrape time is otherwise gone. Compressed HTML is small relative to the cost of re-scraping, and it also gives you the evidence you need to diagnose why a field drifted.

Can validation guarantee my scraped data is correct?

No. Validation and drift detection catch shape and distribution changes; they cannot verify that a parsed value matches what a human would see on the page. A field can stay filled while its meaning changes. Pair automated checks with periodic manual sampling, and remember that scraping must respect each site's terms and applicable law regardless of data quality.