Memvalidasi Data yang Diambil dan Mendeteksi Perubahan Skema

Data yang diambil melalui scraping jarang menimbulkan kegagalan yang mencolok. Masalahnya berkembang secara perlahan: perubahan desain memindahkan kolom harga ke elemen yang berbeda, parser mengembalikan nilai None, dan sepuluh ribu baris data dengan kolom harga kosong dimuat ke dalam gudang data tanpa satu pun pengecualian. Panduan ini membangun lapisan-lapisan yang mendeteksi pergeseran skema sebelum hal itu menyebabkan laporan yang buruk: skema rekaman pydantic v2, pemeriksaan kualitas per batch, detektor pergeseran dengan baseline bergulir, langkah karantina, uji kontrak selektor harian pada URL canary, serta versi parser sehingga Anda dapat mengisi data secara retroaktif setelah memperbaiki masalah.

Mengapa Data yang Diekstraksi Membusuk Tanpa Disadari

Tiga penyebab pergeseran skema menjadi akar hampir semua insiden. Perombakan desain situs mengubah DOM, dan parser yang toleran yang ditulis dengan .get() dan find() mengembalikan nilai kosong alih-alih memicu kesalahan. Tata letak A/B lebih buruk karena hanya sebagian kecil permintaan yang melihat templat baru, sehingga tingkat pengisian data turun dari 99 persen menjadi 70 persen dan tidak ada yang menyadarinya selama seminggu. Perubahan lokal dan geografis mengubah format tanpa mengubah struktur: 1,299.00 menjadi 1.299,00, format tanggal berubah dari ISO menjadi hari-terlebih-dahulu, dan nilainya diparsing secara salah alih-alih tidak diparsing sama sekali.

Sifat yang sama adalah bahwa parser tidak mengalami crash. Toleransi tingkat rekaman adalah pengaturan default yang tepat, tetapi itu berarti keakuratan harus ditegakkan di tempat lain. Tempat tersebut adalah validasi data tingkat batch dalam pipa pengikisan web, dan pembuatannya tidak mahal.

Menentukan skema rekaman dengan pydantic v2

Mulailah dengan menuliskan seperti apa bentuk rekaman yang valid. Bidang yang wajib diisi, tipe data, rentang nilai, dan format akan menjadi model pydantic, dan setiap kamus (dict) yang telah diparsing akan melalui model_validate sebelum diteruskan ke tahap selanjutnya.

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

Ada dua pilihan penting di sini. mode="before" pada validator harga memungkinkan skema melakukan normalisasi terakhir sehingga parser dapat tetap sederhana. Dan e.errors() mengembalikan kesalahan terstruktur dengan tuple loc, yang memungkinkan Anda menghitung kegagalan per bidang, bukan per catatan. Sebuah batch di mana 40 persen baris gagal pada price dan tidak ada yang lain merupakan masalah yang berbeda dari batch di mana setiap baris gagal pada setiap bidang.

Pemeriksaan per-batch

Validasi menyingkirkan baris-baris yang tidak valid. Pemeriksaan batch menganalisis baris-baris yang lolos dan yang tidak lolos secara bersamaan, serta menilai apakah batch tersebut secara keseluruhan masuk akal. Empat jenis pemeriksaan ini dapat mendeteksi sebagian besar penyimpangan skema.

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

Simpan batch_metrics untuk setiap eksekusi. Ini adalah bahan mentah untuk detektor penyimpangan, dan juga hal pertama yang akan Anda periksa ketika seseorang bertanya mengapa angka hari Selasa terlihat aneh.

Detektor penyimpangan dengan garis dasar bergulir

Ambang batas tetap tidak akurat karena sumber-sumbernya berbeda. Bidang yang terisi 60 persen di satu lokasi dianggap normal; di lokasi lain, hal itu bisa menjadi krisis. Sebaiknya bandingkan dengan riwayat terkini sumber itu sendiri: garis dasar bergulir dari N batch terakhir yang baik.

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)

Baris yang penting adalah accept: hanya batch yang bersih yang dimasukkan ke dalam garis dasar. Jika Anda membiarkan batch yang mengalami drift masuk, garis dasar akan mengikuti drift tersebut dan detektor akan menjadi diam tepat ketika seharusnya bersuara keras. Gunakan median daripada rata-rata karena alasan yang sama; satu hari yang tidak biasa tidak boleh mengubah referensi.

Karantina, bukan memproses

Ketika detektor terpicu, jangan memproses batch tersebut, dan jangan pula membuangnya. Simpan batch tersebut ke lokasi karantina bersama metrik dan peringatannya, kirimkan peringatan tersebut, lalu biarkan manusia atau aturan otomatis yang memutuskan. Keputusan biasanya salah satu dari tiga hal berikut: alarm palsu (naikkan status batch dan terima ke dalam baseline), penyimpangan nyata (perbaiki parser, jalankan ulang dari konten mentah yang tersimpan), atau gangguan sumber (buang batch tersebut, biarkan baseline tetap tidak berubah).

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

Berikan peringatan mengenai apa yang berubah, bukan sekadar “validasi gagal”. Pesan yang menyatakan price fill rate 0.98 -> 0.61 on source X dapat ditindaklanjuti; sedangkan kegagalan umum tidak.

Uji kontrak selektor pada URL canary

Pemeriksaan batch memberi tahu Anda bahwa ada sesuatu yang menyimpang setelah hal itu benar-benar terjadi. Uji kontrak selektor memberi tahu Anda lebih awal, seringkali sebelum proses harian dijalankan. Pilih sekumpulan URL canary tetap per sumber, yang nilai yang diharapkannya sudah Anda ketahui, ambil data tersebut melalui POST /v1/extract setiap hari, jalankan parser, dan pastikan bahwa setiap bidang yang tercantum dalam kontrak ada dan masuk akal. Jika kontrak gagal, parser perlu diperiksa sebelum proses eksekusi besar dimulai.

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)

Lima hingga sepuluh canary per sumber sudah cukup. Tetapkan negara proxy jika situs tersebut dilokalkan, karena canary yang diambil dari wilayah yang berbeda akan menguji halaman yang berbeda pula. Pilih URL yang mencakup tata letak yang Anda perhatikan: satu yang tersedia, satu yang habis, satu yang sedang diskon, dan satu tanpa peringkat.

Berikan versi pada parser dan simpan halaman mentah

Setiap baris harus berisi parser_version, dan versinya harus berubah setiap kali logika parser berubah. Hal ini memberikan dua manfaat. Pertama, Anda dapat memfilter metrik apa pun berdasarkan versi dan langsung melihat apakah peringatan penyimpangan bertepatan dengan proses deploy. Kedua, saat Anda memperbaiki parser, Anda tahu persis baris mana yang perlu dihasilkan ulang.

Pembuatan ulang hanya mungkin dilakukan jika Anda menyimpan data masukan. Simpan HTML mentah atau markdown dari setiap pengambilan data, dalam bentuk terkompresi, dengan kunci source_url dan scraped_at. Biaya penyimpanan ini kecil dibandingkan dengan biaya pengambilan data ulang, dan pengambilan data ulang seringkali bahkan bukan pilihan karena halaman tersebut telah berubah lagi. Ketika pergeseran skema telah dikonfirmasi, alur kerjanya adalah: perbaiki parser, tingkatkan versi, parse ulang konten mentah yang disimpan untuk jendela yang terpengaruh, validasi melalui skema yang sama, dan muat untuk menggantikan batch yang dikarantina. Tanpa penyimpanan data mentah, celah tersebut akan permanen.

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

Kesalahan umum

Melakukan validasi per baris dan bukan per batch

Skema baris dapat mendeteksi baris yang tidak valid; namun, skema tersebut tidak dapat mendeteksi batch di mana setiap barisnya valid tetapi setengah dari harga hilang. Selalu hitung tingkat pengisian dan jumlah pada tingkat batch.

Membiarkan batch yang bermasalah masuk ke baseline

Jika baseline bergulir mencakup batch yang mengalami penyimpangan, batch berikutnya yang mengalami penyimpangan akan terlihat normal. Hanya terima batch yang lolos ke riwayat, dan gunakan median.

Membuat parser terlalu toleran

.get("price") yang mengembalikan None tidak masalah, tetapi parser yang mengembalikan 0.0 atau string kosong saat gagal akan menggagalkan pemeriksaan tingkat pengisian. Kembalikan None untuk data yang hilang, dan biarkan skema menolaknya.

Tidak mengunci kondisi pengambilan data

Pengujian canary dan proses produksi harus menggunakan pengaturan rendering dan proxy yang sama. Pengujian kontrak yang lulus dengan render_js: true dan proxy AS tidak membuktikan apa pun mengenai proses yang tidak menggunakan keduanya.

Langkah selanjutnya

Semua proses pengambilan data dalam panduan ini dilakukan melalui Geonode Scraper API, yang mengembalikan markdown atau HTML yang bersih dari satu panggilan POST /v1/extract dengan rendering JavaScript dan penargetan proxy tingkat negara saat Anda membutuhkannya. Menjaga stabilitas lapisan pengambilan data menghilangkan satu variabel dari penyelidikan penyimpangan skema, karena yang berubah adalah halamannya, bukan cara Anda mendapatkannya. Tingkat gratis mencakup 1.5K permintaan per bulan, cukup untuk menjalankan kontrak canary terhadap beberapa sumber setiap hari saat Anda membangun pipa validasi.

Pertanyaan yang Sering Diajukan

Apa yang dimaksud dengan “schema drift” dalam web scraping?

Schema drift adalah perubahan bertahap atau mendadak pada format data yang dihasilkan oleh parser Anda, yang disebabkan oleh perubahan pada situs sumber, bukan pada kode Anda. Perombakan desain memindahkan harga ke elemen baru, uji A/B menampilkan tata letak yang berbeda untuk beberapa permintaan, atau pergantian lokalisasi mengubah format angka. Parser tetap berjalan, tetapi kolom-kolom menjadi kosong atau salah.

Mengapa data yang di-scrape membusuk secara diam-diam alih-alih gagal secara jelas?

Sebagian besar parser dirancang agar toleran, sehingga elemen yang hilang akan mengembalikan None alih-alih memicu pengecualian. Hal ini masuk akal per catatan tetapi berakibat fatal per batch, karena sepuluh ribu catatan dengan harga null akan dimuat tanpa kesalahan. Anda baru mendeteksinya ketika laporan hilir terlihat salah, seringkali berminggu-minggu kemudian. Validasi tingkat batch dan pemeriksaan pergeseran skema menutup celah tersebut.

Bagaimana cara memvalidasi data yang di-scrape dengan pydantic?

Tentukan model per jenis catatan dengan kolom wajib, tipe data, rentang numerik, dan batasan regex, lalu panggil model_validate pada setiap kamus yang telah diparsing di dalam blok try yang menangkap ValidationError. Hitung jumlah kegagalan per kolom di seluruh batch. Tingkat penolakan yang tinggi pada satu bidang merupakan sinyal penyimpangan skema; tingkat penolakan yang tinggi di semua bidang biasanya berarti proses pengambilan data itu sendiri gagal.

Ambang batas apa yang harus memicu peringatan penyimpangan skema?

Bandingkan tingkat pengisian setiap bidang dalam batch saat ini dengan baseline bergulir dari batch-batch baik terbaru. Penurunan sepuluh poin persentase atau lebih pada bidang yang wajib diisi merupakan sinyal yang kuat; gunakan ambang batas lima poin persentase untuk bidang bernilai tinggi seperti harga. Periksa juga jumlah catatan terhadap perkiraan dalam rentang plus atau minus 20 persen dan pantau tingkat duplikat. Sesuaikan ambang batas per sumber setelah satu minggu pengumpulan data.

Apakah saya harus menyimpan HTML mentah atau markdown dari setiap proses pengikisan?

Ya, jika kapasitas penyimpanan memungkinkan. HTML mentah atau markdown adalah satu-satunya cara yang memungkinkan Anda mengisi data yang terlewat setelah memperbaiki parser, karena halaman seperti apa adanya pada saat pengikisan akan hilang jika tidak disimpan. HTML terkompresi berukuran kecil dibandingkan dengan biaya pengikisan ulang, dan juga memberi Anda bukti yang diperlukan untuk mendiagnosis mengapa suatu bidang mengalami penyimpangan.

Apakah validasi dapat menjamin bahwa data yang saya kumpulkan sudah benar?

Tidak. Validasi dan deteksi pergeseran mendeteksi perubahan bentuk dan distribusi; keduanya tidak dapat memverifikasi bahwa nilai yang diparsing sesuai dengan apa yang dilihat manusia di halaman tersebut. Sebuah bidang dapat tetap terisi meskipun maknanya berubah. Padukan pemeriksaan otomatis dengan pengambilan sampel manual secara berkala, dan ingatlah bahwa pengikisan data harus mematuhi ketentuan setiap situs dan hukum yang berlaku, terlepas dari kualitas data.