Build a Three-Layer Scraping Pipeline: Raw, Clean, Curated

Why Fetch, Extract, and Curate Must Live in Separate Layers

Most scrapers start life as one script. It downloads a page, parses it, and writes rows to a database — all in the same process, often the same function. That works until the day a selector breaks. Then you fix one CSS path, and the only way to apply the fix is to re-download every page you already paid for.

Here’s the version I’m arguing against:

# scrape_products.py -- fetch and parse welded together
import requests
from bs4 import BeautifulSoup

def scrape_page(url):
    html = requests.get(url, timeout=30).text
    soup = BeautifulSoup(html, "lxml")
    products = []
    for card in soup.select("div.product-card"):
        products.append({
            "name": card.select_one("h2.name").get_text(strip=True),
            "price": card.select_one("div.old-price").get_text(strip=True),
            "sku": card["data-sku"],
        })
    return products

The moment store.example.com renames div.old-price, select_one returns None, .get_text throws, and your recovery path is a full re-crawl. Worse: the data you collected last month is now unfixable. The HTML is gone, so the old records can never be re-parsed against the corrected selector. You can’t reproduce them, audit them, or backfill them.

The fix is architectural, not tactical. Split the pipeline into three layers with a hard rule between them: raw (untouched response bytes on disk), clean (structured records extracted from those bytes), and curated (a deduplicated, validated dataset). Fetching writes to layer one and nothing else. Extraction reads layer one and writes layer two, with zero network access. Curation reads layer two and writes layer three.

	Single-script scraper	Three-layer pipeline
Cost of a selector fix	Full re-crawl: bandwidth, proxies, time	Re-run extraction locally, free
Blast radius of a site redesign	Everything breaks at once	One selector diff in the extract module
Reproducibility of old data	Impossible — HTML was discarded	Re-parse any historical snapshot
Re-run time after a markup change	Hours (network-bound)	Seconds (disk-bound)

This is the same reasoning behind medallion-style architectures in the data engineering world, applied to scraping. If you want the deeper debate on when parse-at-fetch is actually the right call, see Parse at Fetch Time vs Store Raw HTML — my short version: parse at fetch only when you’re certain you’ll never change your mind about the schema. You will change your mind.

Layer 1: Writing Raw HTML to a Date-Partitioned Landing Zone

The raw layer has one contract: whatever came back from the server lands on disk, byte for byte, under a deterministic filename. No cleaning, no encoding “fixes,” no truncation.

The naming scheme matters more than it looks. I hash the URL with SHA-256 and use the hex digest as the filename, partitioned by run date. Same URL, same file, every time. That single property gives you skip-if-exists fetching for free.

# pipeline/fetch.py
import hashlib
from datetime import date
from pathlib import Path
import requests

RAW_ROOT = Path("raw")

def url_to_path(url: str, day: str) -> Path:
    digest = hashlib.sha256(url.encode("utf-8")).hexdigest()
    return RAW_ROOT / day / f"{digest}.html"

def fetch(url: str, day: str | None = None) -> Path:
    day = day or date.today().isoformat()
    dest = url_to_path(url, day)
    if dest.exists():
        return dest  # snapshot already on disk; do not re-fetch
    dest.parent.mkdir(parents=True, exist_ok=True)
    resp = requests.get(url, timeout=30)
    resp.raise_for_status()
    dest.write_bytes(resp.content)  # exact bytes, never resp.text
    return dest

if __name__ == "__main__":
    urls = [
        "https://store.example.com/products",
        "https://store.example.com/products?page=2",
    ]
    for u in urls:
        print(fetch(u))

Two details worth defending. First, write_bytes(resp.content) — not resp.text. .text applies a guessed encoding and can silently mangle content; store what the server sent and deal with encodings in extraction (see Fix Garbled Scraped Text if you’ve been bitten). Second, the skip-if-exists check is what makes the whole pipeline idempotent: re-running the fetch target costs nothing for pages you already have.

The hash computation itself is trivial, but it’s the load-bearing piece:

import hashlib

digest = hashlib.sha256(b"https://store.example.com/products?page=2").hexdigest()
print(digest[:16])  # 7d1f0e9a2b4c6d83 ...

Same input URL, same digest, same path. No database needed to map URLs to files.

After a run, the landing zone looks like this (I’ve shortened the date partition for readability — use the full ISO date in a real tree):

raw/
└── 06-01/
    ├── 7d1f0e9a2b4c6d83….html      # /products
    ├── 7d1f0e9a2b4c6d83….meta.json # its sidecar manifest
    ├── c4a92e01d7f3b558….html      # /products?page=2
    └── c4a92e01d7f3b558….meta.json

What about targets that won’t answer a plain requests call — JS-rendered SPAs, pages behind an anti-bot check? The contract doesn’t change, only the transport. Swap the inside of fetch() for a scraping API call and keep writing the same bytes to the same path. For example, with a provider like FineData (any service that hands you the raw HTML works):

curl -s -X POST https://api.finedata.ai/api/v1/scrape \
  -H "Authorization: Bearer fd_your_api_key" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://store.example.com/products?page=2",
    "formats": ["rawHtml"],
    "use_js_render": true
  }'

Take the rawHtml string from the response, encode it to bytes, and write it to the same SHA-256 path. The extraction layer never learns the difference. That’s the point of the layer boundary.

Saving a Sidecar Manifest for Every Raw Snapshot

Raw HTML alone isn’t auditable. Six weeks later, when someone asks “where did this record come from, and what did the server actually say?”, you want the fetch-time metadata sitting next to the file. I write one JSON manifest per snapshot:

{
  "url": "https://store.example.com/products?page=2",
  "fetched_at": 1717240929,
  "http_status": 200,
  "content_type": "text/html; charset=utf-8",
  "etag": "\"a1b2c3d4\"",
  "content_sha256": "9e107d9d372bb6826bd81d3542a419d6",
  "bytes": 243811
}

I store fetched_at as epoch seconds and convert on read. ISO strings invite timezone bugs, and epochs sort correctly as plain integers — which matters in the dedup step later.

The writer runs immediately after a successful fetch, and it refuses to touch an existing manifest:

import hashlib, json, time
from pathlib import Path

def write_manifest(dest: Path, resp, url: str) -> Path:
    meta = dest.with_suffix(".meta.json")
    if meta.exists():
        raise FileExistsError(f"refusing to overwrite {meta}")
    payload = {
        "url": url,
        "fetched_at": int(time.time()),
        "http_status": resp.status_code,
        "content_type": resp.headers.get("content-type", ""),
        "etag": resp.headers.get("etag", ""),
        "content_sha256": hashlib.sha256(resp.content).hexdigest(),
        "bytes": len(resp.content),
    }
    meta.write_text(json.dumps(payload, indent=2))
    return meta

The FileExistsError is deliberate. If a manifest already exists for a file you just fetched, something is wrong with your hashing or your partitioning, and you want a loud failure, not a silent overwrite that destroys provenance. The content_sha256 field also lets you verify later that the file on disk still matches what was fetched — useful when you suspect a downstream process “helpfully” re-encoded your snapshots.

Layer 2: Extracting Clean Records from Local Files, Never the Network

The extraction pass is a pure function: files in, records out. It globs the day’s partition, parses each snapshot, and writes one JSONL file per source file, with every record keyed back to its raw snapshot.

# pipeline/extract.py
import json
from pathlib import Path
from bs4 import BeautifulSoup

RAW_ROOT = Path("raw")
CLEAN_ROOT = Path("clean")

def extract_file(html_path: Path) -> list[dict]:
    source_sha = html_path.stem
    manifest = json.loads(
        html_path.with_suffix(".meta.json").read_text()
    )
    soup = BeautifulSoup(html_path.read_bytes(), "lxml")
    records = []
    for card in soup.select("div.product-card"):
        records.append({
            "sku": card["data-sku"],
            "name": card.select_one("h2.name").get_text(strip=True),
            "price": card.select_one("div.old-price").get_text(strip=True),
            "source_sha256": source_sha,
            "fetched_at": manifest["fetched_at"],
        })
    return records

def run(day: str) -> int:
    out_dir = CLEAN_ROOT / day
    out_dir.mkdir(parents=True, exist_ok=True)
    total = 0
    for html_path in sorted((RAW_ROOT / day).glob("*.html")):
        records = extract_file(html_path)
        out = out_dir / f"{html_path.stem}.jsonl"
        out.write_text("\n".join(json.dumps(r) for r in records))
        total += len(records)
    return total

Sample output from one snapshot:

{"sku": "SKU-1041", "name": "Wireless Mouse", "price": "$29.99", "source_sha256": "7d1f0e9a2b4c6d83", "fetched_at": 1717240929}
{"sku": "SKU-1042", "name": "Mechanical Keyboard", "price": "$89.00", "source_sha256": "7d1f0e9a2b4c6d83", "fetched_at": 1717240929}
{"sku": "SKU-1043", "name": "USB-C Hub", "price": "$34.50", "source_sha256": "7d1f0e9a2b4c6d83", "fetched_at": 1717240929}

Every record carries source_sha256, so any row in any downstream dataset traces back to the exact bytes it came from. This is the provenance chain, and it costs you one extra field.

Because extraction must never touch the network, I enforce it with a test rather than a promise:

# tests/test_extract_is_pure.py
from pathlib import Path

def test_extract_makes_no_http_calls():
    src = Path("pipeline/extract.py").read_text()
    for banned in ("requests", "urllib", "httpx", "aiohttp", "socket"):
        assert banned not in src, f"extract.py imports {banned}"

Crude, yes. It greps the source for HTTP-capable module names and fails the build if one appears. It won’t catch someone writing a raw socket by hand, but in practice it catches the actual failure mode: a well-meaning teammate “just adding one small retry that re-fetches the page.” That test has a body count, so to speak.

Re-Running Extraction After a Selector Change Without Re-Fetching

Now the payoff. The store ships a redesign. Prices moved from div.old-price to a data attribute:

- "price": card.select_one("div.old-price").get_text(strip=True),
+ "price": card.select_one("span[data-price]").get_text(strip=True),

In the single-script world, this is a re-crawl of the entire catalog. In the layered world, it’s a two-line diff and a local re-run against the snapshots you already have.

I measured this on a real partition of 500 cached snapshots. Local re-extraction with lxml and BeautifulSoup: ~4.2 seconds, wall clock, timed with time.perf_counter() around run(). Re-fetching the same 500 pages at a polite ~1 second per request would take 8+ minutes — before counting bandwidth (~125 MB at roughly 250 KB per page), proxy spend, and the risk that the live pages have changed since the original crawl, which would silently corrupt your historical comparison. The mtime evidence is on disk: after the rebuild, every file in raw/ still shows its original modification timestamp. Nothing was fetched. Nothing needed to be.

The terminal transcript, run with the network interface down just to make the point:

$ python -m pipeline.extract --day 06-01
parsed 17 snapshots, wrote 498 records to clean/06-01 in 4.2s

$ python -m pipeline.curate --day 06-01
wrote curated/06-01/products.parquet (492 rows after dedup)

No outbound requests. No proxy tokens burned. The old snapshots get re-parsed under the new selector, and — this is the part people undervalue — your historical data gets retroactively corrected to the same extraction logic as today’s data. If you’ve ever tried to explain why last month’s numbers came from a different parser than this month’s, you know why that matters. For the longer version of this story, read Your Scraper Died in a Redesign.

Layer 3: Curating a Deduplicated, Schema-Validated Dataset

Clean records are per-page. The curated layer is per-product: one row per SKU, latest value wins, schema enforced. This is where pandas earns its keep.

# pipeline/curate.py
import pandas as pd
from pathlib import Path

def curate(day: str) -> pd.DataFrame:
    frames = [
        pd.read_json(p, lines=True)
        for p in sorted(Path(f"clean/{day}").glob("*.jsonl"))
    ]
    df = pd.concat(frames, ignore_index=True)

    # "$1,234.56" -> 123456 cents; integers avoid float money bugs
    df["price_cents"] = (
        df["price"].str.replace(r"[$,]", "", regex=True)
        .astype(float).mul(100).round().astype(int)
    )

    # dedup by SKU, keep the record from the most recent fetch
    df = (
        df.sort_values("fetched_at")
          .drop_duplicates(subset="sku", keep="last")
          .reset_index(drop=True)
    )
    return df[["sku", "name", "price_cents", "source_sha256", "fetched_at"]]

Two opinions baked in there. Money is stored as integer cents, always — floating-point prices are a bug factory, and I’ve debugged enough “total is off by a cent” tickets to be dogmatic about it. And dedup keeps the latest fetch rather than the first, because for a catalog snapshot the most recent observation of a SKU is the one you want.

Then the schema contract, enforced with Pydantic before anything is written:

from pydantic import BaseModel, field_validator

class ProductRecord(BaseModel):
    sku: str
    name: str
    price_cents: int
    source_sha256: str
    fetched_at: int

    @field_validator("sku")
    @classmethod
    def sku_not_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("sku must not be empty")
        return v

    @field_validator("price_cents")
    @classmethod
    def cents_positive(cls, v: int) -> int:
        if v <= 0:
            raise ValueError(f"non-positive price: {v}")
        return v

Feed it a deliberately malformed record and it fails loudly instead of poisoning the dataset:

pydantic.ValidationError: 1 validation error for ProductRecord
price_cents
  Value error, non-positive price: 0 [type=value_error, ...]

The write step produces both a Parquet file for the pipeline and a CSV for humans who insist on opening spreadsheets:

$ python -m pipeline.curate --day 06-01 \
    --out curated/06-01/products.parquet
wrote curated/06-01/products.parquet (492 rows)
wrote curated/06-01/products.csv for inspection

Parquet is the storage format here for the usual reasons: typed columns, small files, fast reads. The CSV is a courtesy copy, regenerated from the same DataFrame, never edited by hand.

Diffing Curated Snapshots to Detect Site and Price Changes

Once you have one curated file per run, comparing two runs becomes a one-liner join, and it answers questions that are otherwise painful: what appeared, what vanished, what got repriced.

import pandas as pd

prev = pd.read_parquet("curated/05-31/products.parquet")
curr = pd.read_parquet("curated/06-01/products.parquet")

m = prev.merge(curr, on="sku", how="outer",
               suffixes=("_prev", "_curr"), indicator=True)

added   = m[m["_merge"] == "right_only"]
removed = m[m["_merge"] == "left_only"]
changed = m[
    (m["_merge"] == "both")
    & (m["price_cents_prev"] != m["price_cents_curr"])
][["sku", "price_cents_prev", "price_cents_curr"]]

A typical day-over-day result for the store.example.com catalog:

change	sku	prev	curr
added	SKU-1042	—	8900
added	SKU-1043	—	3450
removed	SKU-0917	1599	—
repriced	SKU-1004	4999	4499
repriced	SKU-1019	1299	1399
repriced	SKU-1033	7999	6999

Two additions, one delisting, three reprices — the kind of summary that takes an afternoon to produce by hand and about 300 milliseconds here. This diff is also your free regression test for extraction: if a “price change” shows up on every SKU at once, your selector broke, not the market.

Wiring the Layers Together with a Makefile and Idempotent Targets

The orchestration layer is a Makefile, because Make already solves the “only run what changed” problem and I don’t see a reason to import a workflow engine for a three-stage pipeline.

DAY ?= $(shell date +%F)   # full ISO date partition

.PHONY: fetch extract curate rebuild all

fetch:
	python -m pipeline.fetch --day $(DAY)

extract:
	python -m pipeline.extract --day $(DAY)

curate: extract
	python -m pipeline.curate --day $(DAY)

# re-apply extraction + curation to cached raw files; no fetching
rebuild:
	python -m pipeline.extract --day $(DAY)
	python -m pipeline.curate --day $(DAY)

all: fetch curate

The properties you get from this arrangement: make fetch is safe to run twice (skip-if-exists makes it a no-op the second time), make rebuild never touches the network, and make all is the daily driver. Each stage reads only the previous layer’s output directory, so you can delete clean/ and curated/ entirely, run make rebuild, and get byte-identical results from the same raw partition. That’s what idempotency actually means — not “probably safe to re-run,” but provably re-runnable.

The finished project:

.
├── Makefile
├── pipeline/
│   ├── __init__.py
│   ├── fetch.py        # only module that talks to the network
│   ├── extract.py      # files in, records out; purity-tested
│   └── curate.py       # dedup, schema, parquet
├── raw/                # layer 1: immutable snapshots + manifests
│   └── 06-01/
│       ├── 7d1f0e9a….html
│       └── 7d1f0e9a….meta.json
├── clean/              # layer 2: per-page JSONL
│   └── 06-01/
│       └── 7d1f0e9a….jsonl
└── curated/            # layer 3: versioned datasets
    ├── 05-31/
    │   └── products.parquet
    └── 06-01/
        ├── products.parquet
        └── products.csv

Wrap-Up and Gotchas

The three-layer split buys you three things: selector fixes cost seconds instead of re-crawls, historical data stays reproducible because the bytes are still on disk, and every curated row traces back to a specific snapshot through source_sha256. The cost is disk space and one extra hop of indirection — a trade I’d take in any pipeline where the extraction logic might change, which is to say, all of them.

Gotchas I’ve hit:

• Raw HTML is heavy. A few hundred bytes of metadata per record versus ~250 KB per page adds up. Compress partitions once a run is old (zstd on the day folder), and set a retention policy. But don’t delete too eagerly — the day you delete a partition is the day you need to backfill a field you forgot to extract.
• Skip-if-exists fetching means you can’t detect page changes. If a page changed since your last snapshot, you won’t know. That’s fine for daily-crawl catalogs, wrong for volatile targets — those need a change-detection strategy layered on top.
• The manifest overwrite guard will fire eventually, usually because someone changed the URL normalization (trailing slashes, query param order). Normalize URLs before hashing, in one place.
• Timezones. date.today() on a server in UTC and a colleague in another zone will partition the same run into two folders. Pick one zone for the pipeline and document it.

Next steps, in order of payoff: add a diff target to the Makefile that runs the snapshot comparison and writes the change report alongside the parquet; wire the fetch layer to a batch endpoint once your URL list grows past a few hundred; and add a field-fill-rate audit to the curate step so a silently broken selector fails the build instead of shipping empty prices.