スクレイピングしたデータの検証とスキーマのドリフトの検出

スクレイピングされたデータが顕著なエラーを引き起こすことはめったにありません。問題は徐々に進行します。例えば、再設計によって価格が別の要素に移動すると、パーサーは「None」というエラーを返すものの、価格が空の1万行のデータが、例外を1つも発生させることなくデータウェアハウスに読み込まれてしまいます。 このガイドでは、スキーマのドリフトがレポートの不具合につながる前にそれを捕捉するための仕組みを構築します。具体的には、pydantic v2のレコードスキーマ、バッチごとの品質チェック、ローリングベースラインを用いたドリフト検出器、隔離ステップ、カナリアURLでの毎日のセレクタ契約テスト、そして問題を修正した後にデータを遡及的に反映できるようにするためのパーサーのバージョン管理です。

抽出されたデータが知らぬ間に劣化してしまう理由

スキーマドリフトの3つの要因が、ほぼすべてのインシデントの原因となっています。サイトの再設計によりDOMが変更されると、.get()やfind()で記述された許容性の高いパーサーは、例外を発生させる代わりに空の値を返してしまいます。 A/Bテストによるレイアウト変更はさらに深刻です。リクエストのごく一部しか新しいテンプレートを表示しないため、充填率が99％から70％に低下しても、1週間は誰にも気づかれません。 ロケールや地域の変更は、構造を変えずにフォーマットを変更します。1,299.00が1.299,00になり、日付がISO形式から日先行形式に反転し、値は完全に解析されないのではなく、誤って解析されてしまいます。

これらに共通するのは、パーサーがクラッシュしないという点です。レコードレベルの許容範囲は適切なデフォルトですが、それは正しさを別の場所で強制しなければならないことを意味します。その「別の場所」とは、ウェブスクレイピングパイプラインにおけるバッチレベルのデータ検証であり、その実装コストは低いです。

pydantic v2 を使用したレコードスキーマの定義

まず、有効なレコードがどのようなものかを定義します。必須フィールド、型、範囲、フォーマットは pydantic モデルとなり、解析されたすべての辞書は、他の処理に進む前に model_validate を通過します。

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

ここで重要な選択が2つあります。価格バリデータに mode="before" を指定することで、スキーマが最後の正規化処理を行うため、パーサーをシンプルに保つことができます。 また、e.errors()はlocタプルを含む構造化されたエラーを返すため、レコード単位ではなくフィールド単位で失敗件数をカウントできるようになります。 priceでのみ行の40パーセントが失敗し、それ以外は問題がないバッチと、すべての行がすべてのフィールドで失敗するバッチでは、問題の性質が異なります。

バッチごとのチェック

検証では行が拒否されます。バッチチェックでは、検証に合格した行と不合格となった行をまとめて検討し、バッチ全体として妥当かどうかを判断します。4つのチェックにより、スキーマのドリフトの大部分を検出できます。

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

実行のたびに batch_metrics を保持してください。これはドリフト検出器の原料となるものであり、火曜日の数値がなぜ不自然に見えるのかと誰かに尋ねられた際に、真っ先に確認すべきデータでもあります。

ローリングベースラインを用いたドリフト検出器

固定しきい値では、ソースごとに状況が異なるため機能しなくなります。あるサイトでは60％満たされている状態が正常でも、別のサイトでは危機的な状況となる場合があります。その代わりに、ソース自身の最近の履歴、つまり直近のN個の正常なバッチに基づくローリングベースラインと比較してください。

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)

重要な点は accept です。基準線には、正常なバッチのみが反映されます。ドリフトしたバッチを基準線に含めてしまうと、基準線もドリフトに追随してしまい、本来は信号が強くなるべき時に検出器が静かになってしまいます。同じ理由から、平均値ではなく中央値を使用してください。1日の異常値だけで基準が変動してはなりません。

ロードの代わりに隔離

検出器がアラートを発生させた場合、そのバッチをロードせず、破棄もしないでください。メトリクスとアラートを添付した状態で隔離場所に書き出し、アラートを送信し、人間または自動化されたルールに判断を委ねてください。 判断は通常、次の3つのいずれかになります：誤警報（バッチを昇格させ、ベースラインに受け入れ）、実際のドリフト（パーサーを修正し、保存された生データから再実行）、またはソースの停止（バッチを破棄し、ベースラインは変更しない）。

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

「検証に失敗した」という事実ではなく、何が変更されたかについてアラートを発行してください。price fill rate 0.98 -> 0.61 on source X というメッセージは対応が必要な情報ですが、一般的な失敗メッセージはそうではありません。

カナリアURLに対するセレクタ契約テスト

バッチチェックでは、何かがずれてからその事実が判明します。一方、セレクタ契約テストでは、その事実をより早く、多くの場合、毎日の実行前に把握することができます。 ソースごとに、期待値が分かっているカナリアURLの固定セットを選び、毎日POST /v1/extractを通じてそれらを取得し、パーサーを実行して、契約対象の各フィールドが存在し、妥当であることを確認します。契約に違反した場合、大規模な実行が始まる前にパーサーの対応が必要です。

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)

ソースごとに5～10個のカナリアURLがあれば十分です。サイトが地域ごとにローカライズされている場合は、プロキシの国を固定してください。異なる地域から取得したカナリアURLでは、異なるページがテストされてしまうためです。重視するレイアウトを網羅するURLを選択してください。例えば、在庫あり、在庫切れ、セール中、評価なしの各状態を1つずつです。

パーサーのバージョンを管理し、生のページデータを保存する

すべての行には「parser_version」を含める必要があり、パーサーのロジックが変更されるたびにバージョンを変更する必要があります。これにより、2つの利点があります。まず、任意のメトリクスをバージョンごとに切り分けて分析することで、ドリフトアラートがデプロイと一致しているかどうかを即座に確認できます。 次に、パーサーを修正した際、どの行を再生成すべきかが正確に把握できます。

再生成は、入力データを保持している場合にのみ可能です。 各フェッチから取得した生のHTMLまたはMarkdownを、圧縮して、source_urlおよびscraped_atをキーとして保存してください。保存コストは再スクレイピングのコストに比べれば微々たるものであり、ページが再び変更されているため、再スクレイピングが選択肢にすらならないこともよくあります。 スキーマのドリフトが確認された場合のワークフローは次の通りです。パーサーを修正し、バージョンを上げ、影響を受けた期間分の保存済み生データを再解析し、同じスキーマで検証した上で、隔離されたバッチと置き換えるように読み込みます。生データの保存がなければ、そのギャップは永久的なものになってしまいます。

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

よくある間違い

レコード単位での検証を行い、バッチ単位での検証を怠る

レコードスキーマは形式不備の行を検出できますが、すべての行が正しい形式であるにもかかわらず、価格の半数が欠落しているようなバッチを検出することはできません。充填率や件数は、常にバッチ単位で計算してください。

不正なバッチをベースラインに含めてしまう

ローリングベースラインにドリフトしたバッチが含まれていると、次にドリフトしたバッチは正常に見えてしまいます。履歴に記録されたバッチのみを受け入れ、中央値を使用してください。

パーサーの許容度が高すぎる

.get("price") に対して None を返すのは問題ありませんが、失敗時に 0.0 や空文字列を返すパーサーでは、フィルレートのチェックが機能しなくなります。データが欠落している場合は None を返し、スキーマで拒否させるようにしてください。

フェッチ条件を固定していない

カナリアテストと本番環境の実行では、同じレンダリング設定とプロキシ設定を使用する必要があります。render_js: true と米国のプロキシで合格した契約テストは、そのどちらも使用しない実行環境については何も証明しません。

次の手順

このガイドで紹介するすべてのフェッチ処理は、 Geonode Scraper APIを経由します。このサービスは、必要に応じてJavaScriptによるレンダリングや国単位のプロキシターゲティングを行い、単一のPOST /v1/extract呼び出しからクリーンなMarkdownまたはHTMLを返します。 フェッチ層を安定させておくことで、スキーマのドリフト調査における変動要因が1つ排除されます。なぜなら、変化するのはページそのものであり、その取得方法ではないからです。無料プランでは月間1.5Kリクエストまで利用可能であり、検証パイプラインを構築している間、毎日複数のソースに対してカナリアテストを実施するには十分な量です。

よくある質問

ウェブスクレイピングにおける「スキーマドリフト」とは何ですか？

スキーマドリフトとは、コードの変更ではなく、ソースサイトの変更によって引き起こされる、パーサーが生成するデータの形式における漸進的または急激な変化のことです。 サイトのリデザインにより価格が新しい要素に移動したり、A/Bテストによって一部のリクエストで異なるレイアウトが表示されたり、ロケールの切り替えで数値の書式が変更されたりすることがあります。パーサーは動作し続けますが、フィールドが空になったり、誤った値になったりします。

スクレイピングされたデータは、なぜ「目立つエラー」を出さずに「静かに」劣化してしまうのでしょうか？

ほとんどのパーサーは許容性が高く設計されているため、要素が欠落しても例外を発生させるのではなく None を返します。これはレコード単位では理にかなっていますが、バッチ単位では致命的です。なぜなら、価格が null のレコードが1万件あっても、エラーなしで読み込まれてしまうからです。この問題は、下流のレポートに不備が見られた時点で初めて検出されますが、多くの場合、数週間後になってしまいます。バッチレベルの検証とドリフトチェックを行うことで、このギャップを埋めることができます。

pydantic を使ってスクレイピングしたデータを検証するにはどうすればよいですか？

レコードタイプごとに、必須フィールド、型、数値範囲、正規表現による制約を定義したモデルを作成し、ValidationError をキャッチする try ブロック内で、解析された各辞書に対して model_validate を呼び出します。バッチ全体でフィールドごとの失敗件数をカウントします。 あるフィールドで拒否率が高い場合はスキーマドリフトの兆候であり、すべてのフィールドで拒否率が高い場合は、通常、データ取得自体が失敗したことを意味します。

スキーマドリフトのアラートをトリガーする閾値はどの程度に設定すべきか？

現在のバッチにおける各フィールドの入力率を、直近の正常なバッチのローリングベースラインと比較します。 必須フィールドで10パーセントポイント以上の低下が見られた場合は、明確なシグナルとみなします。価格などの重要度の高いフィールドについては、5パーセントポイント以上を閾値として設定してください。また、レコード数が予想値から±20パーセントの範囲内にあるかを確認し、重複率にも注意を払ってください。1週間分のデータが蓄積された後、ソースごとに閾値を調整してください。

スクレイピングのたびに生のHTMLやMarkdownを保存すべきですか？

ストレージに余裕があれば、はい。生のHTMLやMarkdownは、パーサーを修正した後にデータを遡及的に補充できる唯一の手段です。そうでなければ、スクレイピング時のページの状態は失われてしまうからです。 圧縮されたHTMLは、再スクレイピングにかかるコストに比べて容量が小さく、また、フィールドの値がずれた原因を診断するために必要な証拠も提供してくれます。

バリデーションによって、スクレイピングしたデータの正確性が保証されますか？

いいえ。バリデーションやドリフト検出は、構造や分布の変化を検知するものであり、解析された値が人間がページ上で目にする内容と一致しているかどうかを検証することはできません。フィールドは、その意味が変化しても、値が入力されたままの状態になる可能性があります。自動チェックと定期的な手動サンプリングを組み合わせて実施し、データの品質にかかわらず、スクレイピングは各サイトの利用規約および適用される法律を遵守しなければならないことを忘れないでください。