Ausgewertete Daten validieren und Schema-Abweichungen erkennen

Ausgekrabbelte Daten scheitern selten mit großem Aufsehen. Sie verrotten einfach: Bei einer Neugestaltung wird ein Preis in ein anderes Element verschoben, der Parser gibt „None“ zurück, und zehntausend Zeilen mit leeren Preisen werden ohne eine einzige Ausnahme in das Data Warehouse geladen. Dieser Leitfaden beschreibt die Ebenen, die Schema-Drift abfangen, bevor sie zu einem fehlerhaften Bericht führt: ein „pydantic v2“-Datensatzschema, Qualitätsprüfungen pro Batch, einen Drift-Detektor mit rollierender Basislinie, einen Quarantäneschritt, einen täglichen Selektor-Vertragstest auf Canary-URLs sowie eine Parser-Versionierung, damit Sie die Daten nachträglich einpflegen können, sobald Sie das Problem behoben haben.

Warum extrahierte Daten unbemerkt veralten

Drei Ursachen für Schema-Drift sind für fast jeden Vorfall verantwortlich. Bei der Neugestaltung von Websites ändert sich das DOM, und ein toleranter Parser, der mit .get() und find() geschrieben wurde, gibt leere Werte zurück, anstatt einen Fehler auszulösen. A/B-Layouts sind noch problematischer, da nur ein Bruchteil der Anfragen die neue Vorlage sieht, sodass Ihre Füllrate von 99 Prozent auf 70 Prozent sinkt und dies eine Woche lang niemand bemerkt. Änderungen an Ländereinstellungen und geografischen Angaben verändern Formate, ohne die Struktur zu verändern: Aus 1,299.00 wird 1.299,00, ein Datum wechselt von ISO-Format zu „Tag zuerst“, und der Wert wird falsch geparst, anstatt gar nicht geparst zu werden.

Gemeinsam ist allen Fällen, dass der Parser nicht abstürzt. Toleranz auf Datensatzebene ist die richtige Standardeinstellung, bedeutet jedoch, dass die Korrektheit an anderer Stelle durchgesetzt werden muss. Diese andere Stelle ist die Datenvalidierung auf Batch-Ebene in Web-Scraping-Pipelines, und deren Implementierung ist kostengünstig.

Definieren eines Datensatzschemas mit pydantic v2

Zunächst wird festgehalten, wie ein gültiger Datensatz aussieht. Pflichtfelder, Typen, Bereiche und Formate werden zu einem pydantic-Modell, und jedes geparste Dict durchläuft model_validate, bevor es an eine andere Stelle weitergeleitet wird.

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

Zwei Entscheidungen sind hier von Bedeutung. mode="before" im Preisvalidator ermöglicht es, die letzte Normalisierung durch das Schema vornehmen zu lassen, sodass der Parser einfach gehalten werden kann. Und e.errors() gibt strukturierte Fehler in Form eines loc-Tupels zurück, wodurch Sie Fehler pro Feld statt pro Datensatz zählen können. Ein Stapel, bei dem 40 Prozent der Zeilen bei price fehlschlagen und sonst nichts, stellt ein anderes Problem dar als einer, bei dem jede Zeile in jedem Feld fehlschlägt.

Chargenbezogene Prüfungen

Bei der Validierung werden Zeilen zurückgewiesen. Bei chargenbezogenen Prüfungen werden die Zeilen, die die Prüfung bestanden haben, und diejenigen, die sie nicht bestanden haben, gemeinsam betrachtet, um zu beurteilen, ob die Charge als Ganzes plausibel ist. Vier Prüfungen decken den Großteil der Schemaabweichungen auf.

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

Speichern Sie batch_metrics bei jedem Durchlauf. Es ist das Ausgangsmaterial für den Drift-Detektor und gleichzeitig das Erste, was Sie sich ansehen, wenn jemand fragt, warum die Zahlen vom Dienstag seltsam aussehen.

Ein Driftdetektor mit gleitender Basislinie

Feste Schwellenwerte versagen, weil sich die Quellen unterscheiden. Ein Feld, das an einem Standort zu 60 Prozent gefüllt ist, gilt dort als normal; an einem anderen ist es ein Krisenfall. Vergleichen Sie stattdessen mit der jüngsten Historie der Quelle selbst: einer gleitenden Basislinie der letzten N einwandfreien Chargen.

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)

Die wichtige Zeile ist accept: Nur fehlerfreie Chargen fließen in die Basislinie ein. Wenn Sie eine Chargen mit Abweichung zulassen, folgt die Basislinie der Abweichung, und der Detektor verstummt genau dann, wenn er eigentlich Alarm schlagen sollte. Verwenden Sie aus demselben Grund den Median statt des Mittelwerts; ein einzelner Ausreißertag sollte die Referenz nicht verschieben.

Quarantäne statt Einlesen

Wenn der Detektor einen Alarm auslöst, sollten Sie den Batch weder einlesen noch verwerfen. Speichern Sie ihn stattdessen zusammen mit seinen Metriken und Alarmen in einem Quarantänebereich, senden Sie den Alarm ab und lassen Sie einen Mitarbeiter oder eine automatisierte Regel entscheiden. Die Entscheidung fällt in der Regel unter eine von drei Möglichkeiten: Fehlalarm (den Batch freigeben und in die Baseline aufnehmen), tatsächliche Abweichung (den Parser korrigieren, die Verarbeitung anhand des gespeicherten Rohdatensatzes erneut durchführen) oder Ausfall der Quelle (den Batch verwerfen, die Baseline unverändert lassen).

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

Warnung darüber, was sich geändert hat, nicht über „Validierung fehlgeschlagen“. Eine Meldung mit dem Text „price fill rate 0.98 -> 0.61 on source X“ ist umsetzbar; ein allgemeiner Fehler ist es nicht.

Der Selektor-Vertragstest für Canary-URLs

Bei Batch-Prüfungen erfahren Sie erst dann, dass etwas abweicht, wenn die Abweichung bereits aufgetreten ist. Ein Selektor-Vertragstest informiert Sie früher darüber, oft noch vor dem täglichen Durchlauf. Wählen Sie pro Quelle einen festen Satz von Canary-URLs aus, deren erwartete Werte Ihnen bekannt sind, rufen Sie diese täglich über POST /v1/extract ab, führen Sie den Parser aus und stellen Sie sicher, dass jedes vertragsgebundene Feld vorhanden und plausibel ist. Wenn der Vertrag nicht erfüllt wird, muss der Parser überprüft werden, bevor der Hauptlauf beginnt.

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)

Fünf bis zehn Canary-URLs pro Quelle reichen aus. Legen Sie das Proxy-Land fest, wenn die Website lokalisiert ist, da ein aus einer anderen Region abgerufener Canary eine andere Seite testet. Wählen Sie URLs aus, die die für Sie relevanten Layouts abdecken: eine mit vorrätigem Artikel, eine mit nicht vorrätigem Artikel, eine im Sonderangebot und eine ohne Bewertung.

Den Parser versionieren und die Rohdaten der Seite speichern

Jede Zeile sollte den Wert „parser_version“ enthalten, und die Version sollte sich ändern, sobald sich die Logik des Parsers ändert. Das bietet Ihnen zwei Vorteile. Erstens können Sie jede Metrik nach Version aufschlüsseln und sofort erkennen, ob eine Drift-Warnung mit einer Bereitstellung zusammenfällt. Zweitens wissen Sie bei der Behebung eines Parserfehlers genau, welche Zeilen neu generiert werden müssen.

Eine Neugenerierung ist nur möglich, wenn Sie die Eingabedaten aufbewahrt haben. Speichern Sie den Roh-HTML- oder Markdown-Code jedes Abrufs, komprimiert und nach „source_url“ und „scraped_at“ indiziert. Die Speicherkosten sind im Vergleich zu den Kosten für ein erneutes Scraping gering, und ein erneutes Scraping ist oft gar keine Option, da sich die Seite bereits wieder geändert hat. Wenn die Schemaabweichung bestätigt ist, sieht der Arbeitsablauf wie folgt aus: Den Parser korrigieren, die Version erhöhen, den gespeicherten Rohinhalt für das betroffene Zeitfenster erneut parsen, ihn anhand desselben Schemas validieren und laden, um die unter Quarantäne gestellten Stapel zu ersetzen. Ohne Rohdatenspeicherung bleibt die Lücke dauerhaft bestehen.

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

Häufige Fehler

Validierung pro Datensatz und niemals pro Stapel

Ein Datensatzschema erkennt fehlerhafte Zeilen; es kann jedoch keinen Stapel erkennen, in dem zwar alle Zeilen korrekt formatiert sind, aber die Hälfte der Preise fehlt. Berechnen Sie Füllraten und Stückzahlen stets auf Stapelebene.

Fehlerhafte Batches in die Baseline aufnehmen

Wenn die rollierende Baseline einen abweichenden Batch enthält, erscheint der nächste abweichende Batch normal. Nehmen Sie nur Batches in den Verlauf auf, die die Prüfung bestanden haben, und verwenden Sie einen Median.

Den Parser zu tolerant gestalten

.get("price"), das None zurückgibt, ist in Ordnung, aber ein Parser, der bei einem Fehler 0.0 oder eine leere Zeichenfolge zurückgibt, macht die Füllgradprüfung zunichte. Geben Sie bei fehlenden Werten None zurück und lassen Sie das Schema dies ablehnen.

Die Abrufbedingungen werden nicht festgeleg

Canary-Tests und Produktionsläufe müssen dieselben Rendering- und Proxy-Einstellungen verwenden. Ein Vertragstest, der mit render_js: true und einem US-Proxy bestanden wird, sagt nichts über einen Lauf aus, der weder das eine noch das andere verwendet.

Nächster Schritt

Alle Abrufe in dieser Anleitung laufen über die Geonode Scraper API, das sauberes Markdown oder HTML aus einem einzigen POST /v1/extract-Aufruf mit JavaScript-Rendering und Proxy-Targeting auf Länderebene liefert, wenn Sie dies benötigen. Durch die Stabilisierung der Abrufebene wird eine Variable bei der Untersuchung von Schema-Abweichungen eliminiert, da sich nur die Seite geändert hat, nicht aber die Art und Weise, wie Sie sie abgerufen haben. Die kostenlose Stufe umfasst 1.500 Anfragen pro Monat – genug, um täglich Canary-Tests für mehrere Quellen durchzuführen, während Sie die Validierungspipeline aufbauen.

FAQ

Was versteht man unter „Schema Drift“ beim Web-Scraping?

Unter „Schema Drift“ versteht man die allmähliche oder plötzliche Veränderung der Struktur der Daten, die Ihr Parser erzeugt, die nicht durch Ihren Code, sondern durch Änderungen an der Quellseite verursacht wird. Durch eine Neugestaltung wird ein Preis in ein neues Element verschoben, ein A/B-Test rendert für einige Anfragen ein anderes Layout, oder ein Wechsel der Ländereinstellung verändert das Zahlenformat. Der Parser läuft weiter, aber Felder bleiben leer oder sind fehlerhaft.

Warum verfallen gescrapte Daten still und leise, anstatt lautstark zu versagen?

Die meisten Parser sind so programmiert, dass sie Fehler tolerieren; ein fehlendes Element gibt daher „None“ zurück, anstatt einen Fehler auszulösen. Das ist auf Datensatzebene sinnvoll, auf Batch-Ebene jedoch katastrophal, da zehntausend Datensätze mit einem Nullwert als Preis ohne Fehler geladen werden. Sie bemerken dies erst, wenn ein nachgelagerter Bericht falsch aussieht – oft erst Wochen später. Validierungen auf Batch-Ebene und Drift-Prüfungen schließen diese Lücke.

Wie validiere ich gescrapte Daten mit pydantic?

Definieren Sie ein Modell pro Datensatztyp mit erforderlichen Feldern, Typen, numerischen Bereichen und Regex-Einschränkungen und rufen Sie dann model_validate für jedes geparste Dict innerhalb eines try-Blocks auf, der ValidationError abfängt. Zählen Sie die Fehler pro Feld über den gesamten Batch hinweg. Eine hohe Ablehnungsrate bei einem Feld ist ein Anzeichen für eine Schemaabweichung; eine durchgehend hohe Rate bedeutet in der Regel, dass der Abruf selbst fehlgeschlagen ist.

Welcher Schwellenwert sollte eine Warnung wegen einer Schemaabweichung auslösen?

Vergleichen Sie die Ausfüllrate jedes Feldes im aktuellen Batch mit einer gleitenden Basislinie aus den letzten fehlerfreien Batches. Ein Rückgang um zehn Prozentpunkte oder mehr bei einem Pflichtfeld ist ein starkes Signal; verwenden Sie fünf Prozentpunkte für hochwertige Felder wie den Preis. Überprüfen Sie außerdem die Datensatzanzahl im Vergleich zur Erwartung innerhalb von plus/minus 20 Prozent und beobachten Sie die Duplikatsrate. Passen Sie die Schwellenwerte pro Quelle nach einer Woche mit Daten an.

Sollte ich Roh-HTML oder Markdown aus jedem Scrape speichern?

Ja, sofern der Speicherplatz dies zulässt. Nur mit Roh-HTML oder Markdown können Sie Daten nachträglich ergänzen, nachdem Sie einen Parser korrigiert haben, da die Seite in ihrem Zustand zum Zeitpunkt des Scrapes sonst verloren ist. Komprimiertes HTML ist im Vergleich zum Aufwand eines erneuten Scrapings gering, und es liefert Ihnen zudem die notwendigen Anhaltspunkte, um zu diagnostizieren, warum ein Feld abgewichen ist.

Kann eine Validierung garantieren, dass meine gescrapten Daten korrekt sind?

Nein. Validierung und Drift-Erkennung erkennen zwar Änderungen an Struktur und Verteilung; sie können jedoch nicht überprüfen, ob ein geparster Wert mit dem übereinstimmt, was ein Mensch auf der Seite sehen würde. Ein Feld kann weiterhin ausgefüllt bleiben, während sich seine Bedeutung ändert. Kombinieren Sie automatisierte Prüfungen mit regelmäßigen manuellen Stichproben und denken Sie daran, dass das Scraping unabhängig von der Datenqualität die Nutzungsbedingungen jeder Website und geltendes Recht beachten muss.