Data Scraping in Production: Build a Pipeline That Lasts

Every data scraping project has the same arc. Week one, the extraction works, and everyone is pleased. Week six, someone asks why a competitor's price has been £0.00 for eleven days, and nobody noticed because nothing ever threw an error.

That is the real shape of data scraping in production. Getting the data is roughly 20% of the work. The other 80% is everything that stands between a raw JSON blob and a table your business can actually trust: validation, deduplication, schema drift, freshness, cost control, and knowing within an hour — not eleven days — when something has quietly stopped working.

If you are still deciding what data scraping is and where it fits, start with our overview of what data scraping is and how it works. This article assumes you are already extracting data and now have to keep it correct.

Table of Contents

• The five ways a data scraping pipeline actually fails
• Prerequisites
• 1. Extraction: make the contract explicit
• 2. Validation: reject at the boundary
• 3. Storage: separate "changed" from "re-scraped"
• 4. Running at volume without a loop
• 5. Monitoring: the four checks that matter
• 6. Scheduling without building an orchestrator
• Troubleshooting reference
• Frequently Asked Questions
• Wrapping up
• Related reading

The five ways a data scraping pipeline actually fails

Not one of these raises an exception. That is precisely why they are dangerous.

Failure	What it looks like	Why it survives undetected
Schema drift	A field silently becomes null for one source	Other fields still populate; row count unchanged
Silent truncation	20 rows collected instead of 400	The run reports success
Duplicate accumulation	The same record inserted on every retry	Counts go up, which looks like growth
Staleness	A source stopped updating three weeks ago	The data is valid, just old
Type corruption	"1.2M" and "$39.99" land in numeric columns	Postgres coerces or your ORM casts silently

A pipeline that only catches crashes catches none of these. What follows is built around detecting them.

Prerequisites

python -m venv venv && source venv/bin/activate
pip install mrscraper-sdk pydantic pandas sqlalchemy psycopg2-binary python-dotenv

Python 3.9+, and a Postgres instance. Get an API token from the MrScraper Dashboard (Profile → API Tokens → New Token) into .env as MRSCRAPER_API_TOKEN.

1. Extraction: make the contract explicit

The extraction step should produce data already shaped like your schema. Most pipeline pain originates here — a scraper that returns "whatever was on the page" pushes normalisation downstream, where it becomes ten conditionals in a transform script.

Encode the contract in the prompt itself:

import asyncio
import os

from dotenv import load_dotenv
from mrscraper import MrScraper

load_dotenv()
client = MrScraper(token=os.getenv("MRSCRAPER_API_TOKEN"))

EXTRACTION_CONTRACT = """
Extract every product on this page in JSON format:
- sku (string, the manufacturer or seller SKU, null if absent)
- name (string)
- price (NUMBER, not a string — strip currency symbols and thousands separators)
- currency (ISO 4217 code, e.g. USD, GBP, EUR)
- in_stock (BOOLEAN — true unless the page explicitly says out of stock)
- rating (number out of 5, or null)
- review_count (integer, expand abbreviations so 1.2K becomes 1200)
- product_url (absolute URL)
Return null for any field not present. Never invent a value.
"""

async def extract(url: str) -> dict:
    run = await client.create_scraper(
        url=url,
        message=EXTRACTION_CONTRACT,
        agent="listing",
        proxy_country="US",
    )
    if run["status"] != "Finished":          # Finished | Processing | Failed
        raise RuntimeError(run.get("error") or f"status={run['status']}")
    return run

Three clauses in that prompt do disproportionate work:

• price (NUMBER, not a string — strip currency symbols) — type coercion at the source. "$39.99" never enters the pipeline, so no downstream regex has to remove it.
• review_count (expand abbreviations so 1.2K becomes 1200) — the same principle for abbreviated counts, which are the most common cause of silently corrupted numeric columns.
• Never invent a value — the guardrail that matters most. You want an explicit null you can detect, not a plausible-looking fabrication you cannot.

---

2. Validation: reject at the boundary

Never let unvalidated scraped data touch your database. A Pydantic model turns a vague dict into an enforced contract, and gives you a rejection reason you can log:

from datetime import datetime, timezone
from decimal import Decimal
from typing import Optional

from pydantic import BaseModel, Field, HttpUrl, ValidationError, field_validator


class Product(BaseModel):
    sku: Optional[str] = None
    name: str = Field(min_length=1, max_length=500)
    price: Decimal = Field(gt=0, lt=1_000_000)
    currency: str = Field(pattern=r"^[A-Z]{3}$")
    in_stock: bool
    rating: Optional[float] = Field(default=None, ge=0, le=5)
    review_count: Optional[int] = Field(default=None, ge=0)
    product_url: HttpUrl

    @field_validator("name")
    @classmethod
    def not_a_placeholder(cls, v: str) -> str:
        junk = {"n/a", "null", "none", "-", "loading...", "undefined"}
        if v.strip().lower() in junk:
            raise ValueError(f"placeholder value: {v!r}")
        return v.strip()


def validate_rows(raw: list[dict]) -> tuple[list[Product], list[dict]]:
    good, bad = [], []
    for row in raw:
        try:
            good.append(Product(**row))
        except ValidationError as e:
            bad.append({"row": row, "errors": e.errors()})
    return good, bad

TIP: Boundary Validation: The bounds are not decoration. price > 0 catches the failure mode where a page renders £0.00 during a stock outage. price < 1_000_000 catches the one where a scraper grabs a product ID and treats it as a price.

Rejected rows are data, not noise. Write them somewhere queryable:

good, bad = validate_rows(rows)
if bad:
    pd.DataFrame(bad).to_sql("scrape_rejects", engine, if_exists="append", index=False)

reject_rate = len(bad) / max(len(rows), 1)
if reject_rate > 0.05:
    raise RuntimeError(f"reject rate {reject_rate:.1%} — likely schema drift, halting")

That last check is the schema-drift alarm. A source that redesigns its markup rarely fails outright; it starts failing 30% of rows. Halting on a reject-rate spike stops you writing a half-corrupted batch and, more importantly, tells you which source changed.

3. Storage: separate "changed" from "re-scraped"

The most common storage mistake is overwriting rows on each run. Do that and you destroy the thing that makes scraped data valuable — its history — while making it impossible to tell whether a value changed or was merely re-collected.

Use two tables. One append-only observation log, one current-state view:

CREATE TABLE product_observations (
    id           BIGSERIAL PRIMARY KEY,
    source       TEXT        NOT NULL,
    sku          TEXT,
    product_url  TEXT        NOT NULL,
    name         TEXT        NOT NULL,
    price        NUMERIC(12,2) NOT NULL,
    currency     CHAR(3)     NOT NULL,
    in_stock     BOOLEAN     NOT NULL,
    rating       REAL,
    review_count INTEGER,
    scraped_at   TIMESTAMPTZ NOT NULL,
    run_id       UUID        NOT NULL,
    row_hash     TEXT        NOT NULL
);

CREATE INDEX ON product_observations (product_url, scraped_at DESC);
CREATE UNIQUE INDEX ON product_observations (run_id, row_hash);

NOTE: Idempotent Runs: row_hash plus run_id is what makes a run idempotent. Retry a partially-failed job and the unique index rejects rows already written, giving you at-most-once delivery semantics without complex bookkeeping.

import hashlib
import json

def row_hash(p: Product) -> str:
    payload = json.dumps(
        {"url": str(p.product_url), "price": str(p.price), "stock": p.in_stock},
        sort_keys=True,
    )
    return hashlib.sha256(payload.encode()).hexdigest()

Current state becomes a view, never a table you mutate:

CREATE VIEW products_current AS
SELECT DISTINCT ON (product_url) *
FROM product_observations
ORDER BY product_url, scraped_at DESC;

Price changes then fall out of the data for free:

SELECT product_url, name,
       price AS current_price,
       LAG(price) OVER w AS previous_price,
       scraped_at
FROM product_observations
WINDOW w AS (PARTITION BY product_url ORDER BY scraped_at)
QUALIFY price IS DISTINCT FROM LAG(price) OVER w;

4. Running at volume without a loop

A for loop over sources is sequential and gives you no per-source visibility. Submit as one job using MrScraper's Web Scraper API and read the per-URL outcome:

async def run_batch(scraper_id: str, urls: list[str]) -> dict:
    job = await client.bulk_rerun_ai_scraper(scraper_id=scraper_id, urls=urls)

    while job["status"] not in ("Finished", "Failed"):
        await asyncio.sleep(15)
        job = await client.get_result_by_id(job["id"])
        s = job["data"]["summary"]
        print(f"{s['scrapedCount']}/{s['totalUrls']} · {s['failedUrls']} failed")

    return job


job = await run_batch(SCRAPER_ID, source_urls)

rows    = job["data"]["mergedData"]
summary = job["data"]["summary"]
failed  = [d for d in job["data"]["urlDetails"] if d["status"] != "Finished"]

urlDetails is what converts a batch failure into an operational signal. A source failing once is transient; the same source failing three runs in a row is a broken source, and those are different alerts:

for d in failed:
    record_failure(source=d["url"], error=d.get("error"), run_id=job["id"])

# alert only on persistent failure
persistent = sources_failing_consecutively(threshold=3)
if persistent:
    notify(f"{len(persistent)} sources failing 3+ runs: {persistent[:5]}")

5. Monitoring: the four checks that matter

Crash alerts catch almost nothing. These four catch the failures that actually reach your dashboards.

Row count deviation — a source returning far fewer rows than usual is the signature of silent truncation:

WITH per_run AS (
  SELECT source, run_id, COUNT(*) AS rows, MAX(scraped_at) AS at
  FROM product_observations GROUP BY source, run_id
),
baseline AS (
  SELECT source, AVG(rows) AS avg_rows, STDDEV(rows) AS sd
  FROM per_run GROUP BY source
)
SELECT p.source, p.rows, b.avg_rows
FROM per_run p JOIN baseline b USING (source)
WHERE p.at > NOW() - INTERVAL '1 day'
  AND p.rows < b.avg_rows - (2 * COALESCE(b.sd, 0))
  AND p.rows < b.avg_rows * 0.7;

Null-rate spike — the clearest schema-drift signal, and it fires before the reject rate does when a field goes optional:

SELECT source,
       ROUND(AVG(CASE WHEN rating IS NULL THEN 1 ELSE 0 END)::numeric, 3) AS null_rating_rate,
       ROUND(AVG(CASE WHEN sku    IS NULL THEN 1 ELSE 0 END)::numeric, 3) AS null_sku_rate
FROM product_observations
WHERE scraped_at > NOW() - INTERVAL '2 days'
GROUP BY source
HAVING AVG(CASE WHEN sku IS NULL THEN 1 ELSE 0 END) > 0.2;

Freshness — the check that catches the eleven-day-old price:

SELECT source, MAX(scraped_at) AS last_seen,
       NOW() - MAX(scraped_at) AS staleness
FROM product_observations
GROUP BY source
HAVING NOW() - MAX(scraped_at) > INTERVAL '36 hours';

Cost per run — tokenUsage and runtime come back on every run, so track them like any other metric. A job whose cost doubles without its row count moving is a job that started rendering pages it did not need to:

record_run_cost(
    run_id=job["id"],
    tokens=summary["totalTokenUsage"],
    urls=summary["totalUrls"],
    successful=summary["successfulUrls"],
)

6. Scheduling without building an orchestrator

At this point you have extraction, validation, storage and monitoring. The remaining piece is execution, and it is where teams most often over-build — a queue, a worker fleet, retry logic, a dead-letter queue, and something to watch all of it.

Attach a Scheduler inside MrScraper and the execution layer stops being your code. It runs on a cron schedule and delivers results straight to your database, S3 bucket, or webhook.

Scheduler (daily 04:00)
  -> Bulk extraction across all sources
  -> Webhook delivers results to your ingest endpoint
  -> Validate -> reject file + clean rows
  -> Append to product_observations (idempotent on run_id + row_hash)
  -> Monitoring queries run -> alerts

Your service then does one job — validate and persist:

@app.post("/ingest")
async def ingest(payload: dict):
    good, bad = validate_rows(payload["data"]["mergedData"])

    if len(bad) / max(len(good) + len(bad), 1) > 0.05:
        alert("reject rate spike — possible schema drift")

    persist(good, run_id=payload["data"]["id"])
    return {"accepted": len(good), "rejected": len(bad)}

TIP: Automated Webhooks & Workflows: Check out our guide on how to use webhooks with a web scraping API and schedule automated web scraping jobs to connect scraping jobs directly into your serverless ingest endpoint.

Troubleshooting reference

Symptom	Cause	Fix
One field suddenly all null	Source markup changed	Null-rate alert; adjust the extraction contract
Row count halved, no errors	Silent truncation or soft block	Row-count deviation alert; check pagination signals
Duplicates after a retry	Non-idempotent writes	Unique index on (run_id, row_hash)
Numbers stored as text	Type coercion left to the database	Specify numeric types in the prompt, enforce in Pydantic
Prices of 0.00 in the table	Out-of-stock pages rendering empty	price > 0 validator; treat as reject, not data
Dashboard shows stale figures	A source stopped without failing	Freshness query on MAX(scraped_at)
Costs rose, output flat	Unnecessary rendering or over-broad crawl	Track tokenUsage per run; tighten crawl bounds
Reject rate climbing slowly	Gradual schema drift	Alert on reject rate, not just on exceptions

Frequently Asked Questions

What is data scraping?

Automated collection of data from websites or applications, converted into a structured format you can query. In production it is less an extraction technique than a data-engineering discipline — the pipeline around the extraction is what determines whether the data can be trusted.

How is data scraping different from web scraping?

In practice the terms overlap heavily. "Web scraping" usually emphasises the act of collecting from web pages; "data scraping" more often describes the broader workflow including transformation, storage and delivery. The distinction matters less than being clear which part of the problem you are solving.

How do I stop scraped data from silently going wrong?

Validate at the boundary with hard bounds, alert on reject rate and null rate rather than only on exceptions, and monitor freshness explicitly. Silent failures dominate scraping incidents precisely because a stale-but-valid row looks identical to a correct one.

Should I overwrite rows or append them?

Append. An observation log plus a DISTINCT ON current-state view costs almost nothing extra and preserves history, which is where most of the analytical value of scraped data lives. Overwriting throws that away permanently.

How do I make scraping runs idempotent?

Hash the identifying fields of each row and enforce a unique index on (run_id, row_hash). Retries then become safe by construction, which matters because partial failure is the normal outcome at scale, not an edge case.

How much data engineering does this really need?

Less than teams expect, if you start with the right shape. Validation is one Pydantic model, storage is two SQL objects, monitoring is four queries. What consumes time is retrofitting all of it onto a pipeline that has already written six months of unvalidated rows.

Wrapping up

Data scraping stops being about extraction the moment somebody depends on the output.

• Encode types in the extraction contract so bad values never enter the pipeline
• Validate at the boundary and keep rejects queryable — they are your drift detector
• Append observations, derive current state; never overwrite history
• Make runs idempotent with a row hash, because retries are routine
• Alert on row counts, null rates, freshness and cost — not just on crashes

Ready to hand off the extraction and scheduling layers? Try MrScraper free — 1,000 free tokens, all pipeline infrastructure handled, no credit card required.

Related reading

• Data Scraping: What It Is, How It Works, and Why It Matters — The conceptual grounding, if you are earlier in the journey than a production pipeline.
• How to Schedule Automated Web Scraping Jobs — The execution layer in depth, including cron patterns and delivery targets.
• How to Use Webhooks With a Web Scraping API — Wiring scrape results into your ingest endpoint without polling.
• How to Export and Store Scraped Data: CSV, JSON, Database, and API — Choosing a storage target before you have a million rows to migrate.
• How to Scale Web Scraping Without Hitting Rate Limits or Getting Banned — Keeping a scheduled pipeline alive as source count grows.
• How AI-Powered Web Scrapers Adapt When Websites Change Their Layout — Why schema drift hits selector-based pipelines hardest.
• Structured vs Unstructured Data — Useful framing for deciding what shape your target schema should take.
• Data Extraction Guide: Tools, Methods, and Best Practices — A broader survey of extraction approaches feeding the same pipeline.
• MrScraper Documentation — Full API reference for the AI scraper, Python SDK, scheduling and integrations.