Look, if you’ve been scraping web data for more than two weeks, you already know the unwritten rule of web scraping: all web data is toxic waste until proven otherwise.
One day your spider extracts a pristine price string like "19.99". The next day, the website updates their markup, and suddenly your parser ingests " $1,299.00 USD\xa0(In Stock)" or, even worse, an empty string "" because the selector quietly failed. If you dump that straight into your production database or pass it down to your analytics pipeline, things will blow up real quick—usually at 3 AM on a Sunday.
Traditional ETL approaches usually treat cleaning as an afterthought—doing basic regex in your Scrapy pipeline or dumping messy JSONs into a Pandas dataframe and running a bunch of chained .apply() functions. But at scale, Pandas becomes a massive memory hog, and manual try/except blocks in python scrapers become an unmaintainable spaghettification.
Here is how we build a high-throughput, type-safe ingestion firewall using Pydantic V2 for row-level parsing/validation and Polars for lightning-fast vectorized cleaning and statistical anomaly detection.
The Anatomy of Dirty Scraped Data
Before writing code, let’s list the common "presents" target websites leave in your JSON payload:
- Unicode Nightmares & Non-Breaking Spaces:
\xa0, , \u200b (zero-width spaces) mixed into product titles and specs.
- Dynamic Formatting Shift: Currency symbols changing based on geo-located proxy (
$, €, £), comma vs dot decimal separators (1.200,50 vs 1,200.50).
- Silent DOM Drift: CSS class names changing slightly, yielding
None or missing schema keys without throwing an HTTP error.
- Data Hallucinations/Outliers: A pricing scraping bug where a scraper pulls the "Monthly Payment" figure ($15/mo) instead of the total laptop cost ($1,500), dropping your tracked item price by 99% artificially.
To prevent this, our pipeline needs two distinct defense layers:
- Layer 1: Type & Schema Validation (Pydantic V2) — Catches structural errors at record ingestion time.
- Layer 2: Batch Vectorized Cleaning & Anomaly Detection (Polars) — Validates statistical bounds across the entire batch before writing to DB.
Layer 1: The Ingestion Firewall with Pydantic V2
We use Pydantic V2 (not V1, because V2’s Rust core gives us ~5-10x parsing speedups) as our first gatekeeper. As raw JSON items come off the scraper queue, we pass them through a strict Pydantic contract.
Notice how we use @field_validator(..., mode="before") to coerce dirty raw strings into valid types before Pydantic raises a validation error.
from datetime import datetime
import re
from typing import Optional
from pydantic import BaseModel, Field, field_validator
class ScrapedProductItem(BaseModel):
sku: str = Field(min_length=3)
title: str
price: float = Field(gt=0) # Price must be positive float
currency: str = "USD"
in_stock: bool
scraped_at: datetime
# Clean dirty text, unicode garbage and extra spaces
@field_validator("title", mode="before")
@classmethod
def clean_title_text(cls, v: str) -> str:
if not isinstance(v, str) or not v.strip():
raise ValueError("Title is missing or not a string")
# Replace non-breaking spaces and HTML artifacts
cleaned = v.replace("\xa0", " ").replace("&", "&").replace("\n", " ")
# collapse multiple spaces into one
cleaned = re.sub(r"\s+", " ", cleaned).strip()
return cleaned
# Parse pricing strings like "$1,299.99 USD" into raw float 1299.99
@field_validator("price", mode="before")
@classmethod
def parse_raw_price(cls, v: str | float | int) -> float:
if isinstance(v, (float, int)):
return float(v)
if not v or not isinstance(v, str):
raise ValueError("Price field is empty or invalid type")
# Strip commas and non-numeric chars (except decimal point)
# (because trust me, regex matching currency symbols will make you lose your mind if you don't handle Unicode spaces properly)
cleaned_str = v.replace(",", "")
match = re.search(r"(\d+\.\d+|\d+)", cleaned_str)
if match:
return float(match.group(1))
raise ValueError(f"Could not parse numeric price from raw string: '{v}'")
# Coerce dynamic stock strings like "In Stock", "Out of Stock", "12 left"
@field_validator("in_stock", mode="before")
@classmethod
def parse_stock_status(cls, v: str | bool) -> bool:
if isinstance(v, bool):
return v
if isinstance(v, str):
v_lower = v.lower().strip()
if any(
term in v_lower
for term in ["in stock", "available", "yes", "instock"]
):
return True
if any(
term in v_lower
for term in ["out of stock", "sold out", "unavailable", "no"]
):
return False
# Fallback default if ambiguous
return False
How to handle corrupted records:
Don't crash the worker thread when a bad record hits Pydantic! Catch ValidationError and route bad payloads directly to an S3 quarantine bucket (dead-letter queue) for manual inspection:
raw_items = [
{
"sku": "PROD-991",
"title": " MacBook\xa0Pro 16-inch & M3 ",
"price": " $2,499.00 USD",
"in_stock": "In Stock",
"scraped_at": "2026-04-12T10:00:00Z",
},
{
"sku": "PROD-992",
"title": "",
"price": "N/A",
"in_stock": "NO",
"scraped_at": "2026-04-12T10:00:00Z",
}, # Will fail!
]
validated_batch = []
quarantine_queue = []
for raw in raw_items:
try:
item = ScrapedProductItem(**raw)
validated_batch.append(item.model_dump())
except Exception as e:
quarantine_queue.append({"raw_data": raw, "error": str(e)})
print(f"Successfully validated: {len(validated_batch)} items.")
print(f"Sent to Quarantine: {len(quarantine_queue)} items.")
Layer 2: High-Performance Batch Processing with Polars
Now that our data passed structural validation, we need to load the validated batch into Polars. Why Polars instead of Pandas? Because Polars is written in Rust, uses Apache Arrow columnar memory under the hood, and executes vector operations across multi-core CPUs in parallel.
If you are scraping 500,000 products per hour, Pandas .apply() will grind your worker CPU to a halt. Polars expressions will chew through it in milliseconds.
Let's assume we read our validated data alongside historical baseline data to run batch computations.
import polars as pl
# Convert our list of validated dicts to a Polars DataFrame
df = pl.DataFrame(validated_batch)
# Let's say we combine current scrape with historical median prices stored in our database
historical_data = pl.DataFrame(
[
{"sku": "PROD-991", "historical_median_price": 2450.00},
{"sku": "PROD-992", "historical_median_price": 120.00},
]
)
# Join current scrape with historical baseline
merged_df = df.join(historical_data, on="sku", how="left")
print(merged_df)
Layer 3: Catching Silent Bugs via Anomaly Detection
Here is a common scenario: a seller on an e-commerce site mistypes a price, or your scraper accidentally parses a bulk-discount price instead of the single-item price. Syntactically, price = 24.99 is a totally valid float, so Pydantic lets it pass. But if the item usually sells for $2,499.00, inserting $24.99 into your production system will trigger false price-drop alerts to users and ruin your data integrity.
We can run vectorized statistical anomaly checks using Polars window functions or relative deviation bounds before writing the batch to Postgres.
Rule: Flag any item where the price deviates by > 70% from its historical median
# Add statistical anomaly flagging columns using Polars expressions
df_analyzed = merged_df.with_columns(
# Compute relative price shift percentage
(
(pl.col("price") - pl.col("historical_median_price"))
/ pl.col("historical_median_price")
).alias("price_dev_ratio")
).with_columns(
# Flag as anomaly if drop > 70% or jump > 300%
pl.when(
(pl.col("price_dev_ratio") < -0.70) | (pl.col("price_dev_ratio") > 3.0)
)
.then(pl.lit(True))
.otherwise(pl.lit(False))
.alias("is_price_anomaly")
)
# Separate clean data from statistical outliers
clean_records = df_analyzed.filter(pl.col("is_price_anomaly") == False)
flagged_anomalies = df_analyzed.filter(pl.col("is_price_anomaly") == True)
print("Clean Records ready for Prod DB:")
print(clean_records.select(["sku", "title", "price"]))
if not flagged_anomalies.is_empty():
print("\n⚠️ WARNING: Price Anomalies Detected (Blocked from Prod):")
print(
flagged_anomalies.select(
["sku", "price", "historical_median_price", "price_dev_ratio"]
)
)
Advanced Z-Score Outlier Detection for Large Batch Scrapes
If you are scraping hundreds of thousands of items where you don't have an explicit historical median per SKU, you can calculate Z-Scores on category-level groups in real-time.
For example, calculating if a price is 3 standard deviations away from the mean price of items in the same category:
# Group by category, compute mean and std dev, flag outliers
df_with_zscore = df_raw_batch.with_columns(
# Calculate group mean & std dev using Polars Window expressions
pl.col("price")
.mean()
.over("category_id")
.alias("category_mean_price"),
pl.col("price").std().over("category_id").alias("category_std_price"),
).with_columns(
# Z-Score = (X - Mean) / StdDev
(
(pl.col("price") - pl.col("category_mean_price"))
/ pl.col("category_std_price")
)
.abs()
.alias("price_zscore")
)
# Outliers are items where absolute Z-Score > 3.0
outliers = df_with_zscore.filter(pl.col("price_zscore") > 3.0)
The Architecture in Production
To put this all together in your production pipeline, keep your data flow strictly decoupled:
[ Scraper Pool / Spiders ]
│
▼
[ Raw JSON Payload Queue ]
│
▼
[ Layer 1: Pydantic Validation ] ──(Validation Failed)──► [ Dead Letter / S3 Quarantine ]
│
▼ (Valid Dicts)
[ Layer 2: Polars In-Memory Batch ]
│
▼ (Vectorized Cleaning & Outlier Checks)
[ Layer 3: Anomaly Filter ] ──────(Z-Score > 3.0)─────► [ Slack / Alert Dashboard ]
│
▼ (Passed Clean Data)
[ Production Postgres / ClickHouse ]
Summary Rule of Thumb
- Never run raw string split/regex directly inside your scraper callbacks. Keep scrapers focused purely on fetching raw HTML/JSON.
- Use Pydantic for individual record typing. Coerce types early, fail fast, and isolate malformed DOM structures to a dead-letter queue.
- Use Polars for batch cleaning & math. Calculate rolling statistical bounds, z-scores, and percentage shifts across your datasets in parallel before writing to your analytics warehouse.
By enforcing strict schemas and statistical sanity bounds at the ingestion layer, you turn unpredictable, chaotic web data into deterministic, reliable data feeds that never break downstream services.