Web data normalization is the process of transforming inconsistent raw JSON from different scraping sources into a single, validated schema with predictable field names, data types, and value formats. For teams collecting data from Google Maps, Amazon, LinkedIn, or social platforms through APIs like CoreClaw's scraper store, each source returns its own field names, nesting depth, and value conventions — business_name vs name, phone_number vs phone vs tel, prices as strings with currency symbols vs raw floats. A reusable Python normalization pipeline with field mapping, type coercion, validation rules, and deduplication handles this at scale without writing a custom transformer for every new source.
TL;DR
Build a DataNormalizer class that accepts a field-mapping config (source field → target field), a type-coercion config (target field → expected Python type), and a validation config (required fields, value ranges). Feed raw scraped JSON through the pipeline to get clean, consistent records ready for storage, analytics, or AI-agent context. Use the CoreClaw console to configure and test your data sources, then normalize the structured output downstream in your own pipeline.
Why Raw Scraped Data Is Inconsistent
Even when a web data API returns structured JSON, the schema varies across sources for three reasons:
Source-specific field names. A Google Maps business record uses
displayed_name; an Amazon product usesproduct_title; a LinkedIn company page usesname. All three map to the same concept — a human-readable entity name — but the keys differ.Inconsistent value formats. Phone numbers may include country codes, extensions, or formatting characters. Prices may be strings like
"$29.99", integers in cents like2999, or floats like29.99. Dates may be ISO-8601, Unix timestamps, or human-readable strings.Variable nesting and optional fields. One source nests address fields under
location.address; another flattens them at the top level. Optional fields may benull, missing entirely, or an empty string.
Without normalization, every downstream consumer — your database, analytics dashboard, CRM, or AI agent — needs to handle every variant. That creates brittle code that breaks when a source changes its response shape.
The Normalization Pipeline Architecture
A robust normalization pipeline has five stages:
| Stage | Input | Output | Key Operations |
|---|---|---|---|
| Field mapping | Raw JSON with source-specific keys | Flat dict with target field names | Key rename, nested-field extraction |
| Type coercion | String-typed values | Correctly typed values | Parse float, int, datetime, bool |
| Validation | Coerced record | Clean record or error list | Required fields, range checks, regex |
| Deduplication | List of validated records | Deduplicated list | Hash-based, fuzzy match, or key-based |
| Enrichment | Deduplicated record | Enriched record | Derived fields, source provenance |
Each stage should be independently testable and configurable. Hard-coding transformations inside one monolithic function makes the pipeline impossible to maintain when you add a new source.
Runnable Python Pipeline
import os
import json
import re
import hashlib
from datetime import datetime, timezone
from typing import Any, Callable
# Configuration is loaded from environment or a config file.
# Endpoint and API key values are conceptual — copy current
# values from your CoreClaw console or product documentation.
CORECLAW_ENDPOINT = os.environ.get("CORECLAW_ENDPOINT", "")
CORECLAW_API_KEY = os.environ.get("CORECLAW_API_KEY", "")
class DataNormalizer:
"""Reusable normalization pipeline for multi-source web data."""
def __init__(
self,
field_map: dict[str, str],
type_map: dict[str, Callable],
required_fields: list[str],
dedup_key: str | None = None,
):
self.field_map = field_map
self.type_map = type_map
self.required_fields = required_fields
self.dedup_key = dedup_key
# --- Stage 1: Field mapping ---
def _map_fields(self, raw: dict) -> dict:
mapped = {}
for source_key, target_key in self.field_map.items():
value = self._extract_nested(raw, source_key)
if value is not None:
mapped[target_key] = value
return mapped
@staticmethod
def _extract_nested(record: dict, dotted_key: str) -> Any:
"""Support dot-notation for nested fields: 'location.address.city'."""
keys = dotted_key.split(".")
value = record
for k in keys:
if isinstance(value, dict) and k in value:
value = value[k]
else:
return None
return value
# --- Stage 2: Type coercion ---
@staticmethod
def _coerce(value: Any, coercer: Callable) -> Any:
try:
return coercer(value)
except (ValueError, TypeError, AttributeError):
return None
def _coerce_fields(self, record: dict) -> dict:
coerced = {}
for key, value in record.items():
if key in self.type_map and value is not None:
coerced[key] = self._coerce(value, self.type_map[key])
else:
coerced[key] = value
return coerced
# --- Stage 3: Validation ---
def _validate(self, record: dict) -> tuple[dict | None, list[str]]:
errors = []
for field in self.required_fields:
if field not in record or record[field] in (None, "", []):
errors.append(f"Missing required field: {field}")
if errors:
return None, errors
return record, []
# --- Stage 4: Deduplication ---
@staticmethod
def _record_hash(record: dict, key: str) -> str:
"""Deterministic hash for deduplication based on a key field."""
raw = str(record.get(key, ""))
return hashlib.sha256(raw.encode()).hexdigest()
def _deduplicate(self, records: list[dict]) -> list[dict]:
if not self.dedup_key:
return records
seen: set[str] = set()
unique = []
for r in records:
h = self._record_hash(r, self.dedup_key)
if h not in seen:
seen.add(h)
unique.append(r)
return unique
# --- Stage 5: Enrichment ---
@staticmethod
def _enrich(record: dict) -> dict:
record["_normalized_at"] = datetime.now(timezone.utc).isoformat()
record["_pipeline_version"] = "1.0"
return record
# --- Full pipeline ---
def normalize(self, raw_records: list[dict]) -> list[dict]:
clean = []
errors_log = []
for raw in raw_records:
mapped = self._map_fields(raw)
coerced = self._coerce_fields(mapped)
validated, errs = self._validate(coerced)
if validated:
clean.append(self._enrich(validated))
else:
errors_log.extend(errs)
deduped = self._deduplicate(clean)
if errors_log:
print(f"[normalizer] {len(errors_log)} validation issues skipped")
return deduped
# --- Coercion helpers ---
def to_float(value: Any) -> float:
"""Parse price strings like '$29.99' or '2999' (cents) into float."""
if isinstance(value, (int, float)):
return float(value)
cleaned = re.sub(r"[^\d.]", "", str(value))
return float(cleaned) if cleaned else 0.0
def to_phone(value: Any) -> str:
"""Strip formatting, keep digits and leading +."""
digits = re.sub(r"[^\d+]", "", str(value))
return digits if digits else ""
def to_int(value: Any) -> int:
return int(float(str(value).replace(",", "")))
# --- Example: normalizing Google Maps business records ---
maps_field_map = {
"displayed_name": "name",
"phone": "phone",
"location.address.postal_code": "postal_code",
"rating": "rating",
"user_rating_count": "review_count",
"location.address.country": "country",
}
maps_type_map = {
"rating": to_float,
"review_count": to_int,
"phone": to_phone,
}
maps_required = ["name", "phone"]
maps_dedup_key = "phone"
normalizer = DataNormalizer(
field_map=maps_field_map,
type_map=maps_type_map,
required_fields=maps_required,
dedup_key=maps_dedup_key,
)
# Sample raw records (representative — not live API output)
raw_businesses = [
{
"displayed_name": "Sunset Bakery",
"phone": "+1 (415) 555-0123",
"location": {"address": {"postal_code": "94102", "country": "US"}},
"rating": "4.7",
"user_rating_count": "312",
},
{
"displayed_name": "Sunset Bakery", # duplicate
"phone": "+1 (415) 555-0123",
"location": {"address": {"postal_code": "94102", "country": "US"}},
"rating": 4.7,
"user_rating_count": 312,
},
{
"displayed_name": "Blue Bottle Coffee",
"phone": "+1 (415) 555-9876",
"location": {"address": {"postal_code": "94110", "country": "US"}},
"rating": "4.5",
"user_rating_count": "1,204",
},
]
clean_records = normalizer.normalize(raw_businesses)
print(json.dumps(clean_records, indent=2))
Expected Output
[
{
"name": "Sunset Bakery",
"phone": "+14155550123",
"postal_code": "94102",
"country": "US",
"rating": 4.7,
"review_count": 312,
"_normalized_at": "2026-09-12T09:00:00+00:00",
"_pipeline_version": "1.0"
},
{
"name": "Blue Bottle Coffee",
"phone": "+14155559876",
"postal_code": "94110",
"country": "US",
"rating": 4.5,
"review_count": 1204,
"_normalized_at": "2026-09-12T09:00:00+00:00",
"_pipeline_version": "1.0"
}
]
Notice: the duplicate "Sunset Bakery" record (second entry with the same phone) was removed by the deduplication stage, and string-typed ratings and review counts were coerced to float and int.
Adapting the Pipeline for Multiple Sources
The same DataNormalizer class works for Amazon product data, LinkedIn company records, or YouTube channel stats — you only change the configuration:
# Amazon product normalization config
amazon_field_map = {
"product_title": "name",
"price.value": "price",
"price.currency": "currency",
"rating": "rating",
"review_count": "review_count",
"asin": "product_id",
}
amazon_type_map = {
"price": to_float,
"rating": to_float,
"review_count": to_int,
}
amazon_required = ["name", "product_id"]
amazon_dedup_key = "product_id"
amazon_normalizer = DataNormalizer(
field_map=amazon_field_map,
type_map=amazon_type_map,
required_fields=amazon_required,
dedup_key=amazon_dedup_key,
)
This approach means adding a new data source is a configuration task, not a code change. You define the field map, type map, and validation rules, then reuse the same pipeline.
Business Use Cases
| Use Case | Sources | Normalized Output | Downstream Consumer |
|---|---|---|---|
| B2B lead enrichment | Google Maps, LinkedIn | Unified company records with name, phone, address, industry | CRM import (HubSpot, Salesforce) |
| Price intelligence | Amazon, Walmart, eBay | Product records with consistent price-as-float, currency, product_id | Price monitoring dashboard |
| Creator analytics | YouTube, Instagram, TikTok | Creator records with engagement_rate, follower_count, content_count | Marketing analytics pipeline |
| Competitive intelligence | Google SERP, Google Maps | Keyword + business ranking records | SEO reporting tool |
| AI agent context | All of the above | Structured, validated JSON with provenance | RAG / vector database |
Each use case benefits from normalization because downstream systems — CRMs, databases, AI models — expect consistent schemas. Without normalization, you spend more time writing adapters than building features.
DIY Normalization vs CoreClaw Structured Data vs Custom Pipeline
| Dimension | DIY (raw requests + custom code) | CoreClaw structured data + custom normalizer | Enterprise data platform |
|---|---|---|---|
| Field consistency | Raw HTML/JSON varies widely | API returns structured JSON; you normalize residual differences | Platform handles normalization end-to-end |
| Setup model | Build parser + normalizer from scratch | Configure field maps; reuse pipeline class | Vendor manages schema and delivery |
| Maintenance burden | High — parser breaks on layout changes | Medium — update field map when API schema changes | Low — vendor handles it |
| Data coverage | Limited to what you can parse | Multiple sources through one API (see pricing for current options) | Varies by vendor |
| Integration path | Custom HTTP + parsing | HTTP API + env-var-configured Python | Vendor SDK or managed pipeline |
| Cost model | Infrastructure + dev time | Pay-per-result; no proxy or infra costs | Enterprise licensing |
The middle option — CoreClaw structured data plus a lightweight custom normalizer — is the sweet spot for most teams. You get consistent JSON from the API, and your normalizer handles the last mile of field mapping and type coercion.
Limitations and Compliance
- Schema drift. Even structured APIs occasionally change field names or add nested objects. Your normalizer should log unknown fields rather than silently dropping them, so you can update the field map when schemas change.
- Type coercion edge cases. Phone numbers, currencies, and dates have regional variations. Test coercers against representative records from each region you operate in.
- Deduplication is not identity resolution. Hashing on a phone number catches exact duplicates. It does not merge records that refer to the same business with slightly different phone formats. For identity resolution, you need fuzzy matching (e.g., Levenshtein distance or TF-IDF similarity) — a separate concern from normalization.
- Compliance. Only collect and normalize publicly available data. Respect target-site terms of service, robots directives where relevant, and applicable privacy regulations (GDPR, CCPA). The normalizer does not make data collection lawful — it only structures data you have already lawfully obtained.
- Provenance tracking. Always store the source URL and collection timestamp alongside normalized records. This supports auditability and reproducibility, especially when normalized data feeds AI-agent context.
FAQ
Do I need a normalizer if I use a structured web data API?
Yes, if you combine data from multiple sources or feed it into a system that expects a specific schema. A structured API reduces the parsing burden, but field names, value formats, and nesting still vary across sources. A normalizer handles that last mile.
How do I handle fields that exist in one source but not another?
Mark them as optional in your field map. The normalizer skips missing fields rather than erroring. For required fields, the validation stage catches records that lack them and logs an error.
What happens when a source changes its response schema?
Your normalizer will start dropping fields or producing None values. To detect this early, log unknown source fields (keys in the raw record that are not in your field map). Review these logs periodically and update the field map.
Can this pipeline run inside n8n, Zapier, or Make.com?
Yes. The normalizer is a plain Python class. You can run it as a Code node in n8n, a Python step in a Zapier Code action, or an HTTP webhook handler invoked from any no-code platform. The output is clean JSON that flows into the next node.
How do I normalize data for AI-agent context?
Run the normalizer, then add a _source_url and _collected_at field to each record. Package records as a JSON array or JSONL file and load them into a vector database or pass them as structured context to an LLM. The consistent schema makes retrieval and grounding more reliable.
Should I store raw records or only normalized records?
Store both. Raw records are your source of truth — if the normalizer has a bug, you can re-run it on the raw data. Store normalized records in your analytics or application database for fast querying.
How many sources can one normalizer handle?
There is no hard limit, but in practice, 3-5 sources per normalizer instance keeps the configuration manageable. For more sources, group them by domain (e.g., one normalizer for e-commerce sources, another for social media) and merge outputs downstream.
Summary
A web data normalization pipeline turns inconsistent JSON from multiple scraping sources into clean, validated, deduplicated records with a single consistent schema. The pipeline has five stages — field mapping, type coercion, validation, deduplication, and enrichment — and each stage is configurable without code changes. By combining CoreClaw's structured web data API with a lightweight Python normalizer, you get production-ready data without maintaining parsers for every source. Start by configuring your field maps for one source, validate the output, then add more sources as needed. Explore the CoreClaw console to test data sources and review pricing options for your volume.
Related Reading
- CoreClaw Workers platform — deploy custom scrapers without DevOps
- CoreClaw scraper store — ready-to-use scrapers for Google Maps, Amazon, LinkedIn, and more
Top comments (0)