Batch a URL List Into Partitioned Parquet Files

Why Scraping Pipelines Fail

Most scraping pipelines die the same way: someone dumps a URL list into a loop, writes the results to one giant JSON file, and calls it a dataset. Six weeks later nobody can answer basic questions about it. Which rows failed? When were they fetched? Which partition holds the store pages? Why does the row count not match the input list?

The fix is boring and worth it: treat the pipeline as a series of staged contracts with an explicit schema, validate rows before they land in the primary dataset, and write partitioned Parquet instead of ad-hoc JSON. Add a run manifest so every output file traces back to an input checksum, and you get a dataset you can actually audit. This post walks through the whole thing in Python: fetch, extract, validate, load, manifest, verify.

If you want the broader argument for parse-at-fetch versus storing raw HTML first, read Parse at Fetch Time vs Store Raw HTML — the short version is that I store raw HTML during the fetch stage and extract from it in a separate pass, because extraction bugs are cheap to fix when you can re-run them against saved bytes.

Modeling the URL List as a Staged DataFrame Schema

Before writing any fetching code, pin down the schema. Every stage in this pipeline hands a table to the next stage, and those handoffs need typed contracts — not dictionaries that grow mystery keys over time.

Three stages, three contracts:

• Fetched: what the downloader produced — raw HTML, HTTP status, timing.
• Extracted: typed, nullable fields parsed out of the HTML.
• Validated: rows that passed the rules, plus a quarantine stream for everything else.

Here is the PyArrow schema for the final dataset:

import pyarrow as pa

FINAL_SCHEMA = pa.schema([
    pa.field("url", pa.string(), nullable=False),
    pa.field("fetch_status", pa.int16(), nullable=False),
    pa.field("fetched_at", pa.timestamp("us", tz="UTC"), nullable=False),
    pa.field("title", pa.string()),            # extraction may fail -> null
    pa.field("canonical_url", pa.string()),
    pa.field("word_count", pa.int32()),
    pa.field("content_sha256", pa.string(), nullable=False),
    pa.field("extract_ok", pa.bool_(), nullable=False),
    pa.field("run_id", pa.string(), nullable=False),
    # partition keys (also stored as columns for convenience)
    pa.field("fetch_date", pa.string(), nullable=False),
    pa.field("host", pa.string(), nullable=False),
])

Two decisions worth defending. First, title and word_count are nullable on purpose — an extraction failure is a null, not an exception, and it must never kill the row. Second, fetch_status is an int16, not a string like “OK”. You will want to filter fetch_status == 200 at some point; make that a numeric comparison from day one.

The input is deliberately unglamorous — a CSV of URLs:

url
https://example.com/
https://example.com/about
https://store.example.com/products
https://store.example.com/products/widget-42
https://store.example.com/products/missing-sku

Five rows: two hosts, one URL that will 404. That last one matters. A test input where everything succeeds tests nothing.

Stage 1: Fetching URLs With Retries and Politeness Delays

The fetch stage has one job: turn URLs into (status, html, fetched_at) tuples without hammering any single host. Configuration first, because hardcoding retry policy inside a loop is how you end up with three different retry behaviors across one codebase:

from dataclasses import dataclass

@dataclass(frozen=True)
class FetchConfig:
    timeout: float = 15.0
    max_retries: int = 3
    backoff_base: float = 1.5      # seconds; 1.5, 2.25, 3.375 ...
    per_host_concurrency: int = 2  # at most 2 in-flight per host
    per_host_delay: float = 1.0    # min seconds between requests to same host
    max_connections: int = 20      # global cap

Two in-flight per host is conservative. For a polite batch job against small sites, conservative is correct — I would rather a run take twenty minutes than get a third of the list served block pages. If you are coming from a world where you already got burned by silent blocks, Your 200 OK Responses Are Actually Block Pages explains why status codes alone are not a sufficient success signal.

The fetcher itself, using httpx with a per-host gate:

import asyncio, hashlib, time
import httpx

async def fetch_one(client, url, cfg, host_gates):
    gate = host_gates[host_of(url)]
    for attempt in range(cfg.max_retries):
        async with gate:                      # limits concurrency + spacing
            try:
                resp = await client.get(
                    url, timeout=cfg.timeout,
                    follow_redirects=True,
                    headers={"User-Agent": "batch-pipeline/0.3"},
                )
                body = resp.text
                return {
                    "url": url,
                    "fetch_status": resp.status_code,
                    "fetched_at": time.time(),
                    "content_sha256": hashlib.sha256(body.encode()).hexdigest(),
                    "raw_html": body,
                    "attempt": attempt,
                }
            except (httpx.TimeoutException, httpx.TransportError):
                if attempt == cfg.max_retries - 1:
                    return {
                        "url": url, "fetch_status": 0,
                        "fetched_at": time.time(),
                        "content_sha256": "", "raw_html": "",
                        "attempt": attempt,
                    }
                await asyncio.sleep(cfg.backoff_base ** attempt)

def host_of(url: str) -> str:
    return url.split("//", 1)[1].split("/", 1)[0]

Note the failure shape: a transport error becomes fetch_status = 0 with empty HTML. It stays in the pipeline as data. A row with status 0 is a fact you can count, chart, and quarantine — an exception that kills the run is neither.

One honest trade-off: this plain-HTTP fetcher will not get you through JavaScript-rendered SPAs or aggressive bot checks. When I hit those, I stop pretending httpx can do it and push the URL list through a scraping API’s async batch endpoint instead — one request, the whole list, results via webhook:

curl -X POST https://api.finedata.ai/api/v1/async/batch \
  -H "Authorization: Bearer fd_your_api_key" \
  -H "Content-Type: application/json" \
  -d '{
    "callback_url": "https://api.example.com/hooks/batch-done",
    "requests": [
      {"url": "https://store.example.com/products",
       "use_js_render": true, "formats": ["rawHtml"]},
      {"url": "https://store.example.com/products/widget-42",
       "use_js_render": true, "formats": ["rawHtml"]}
    ]
  }'

Then poll GET /api/v1/async/batch/{batch_id} with include_results=true and feed the returned HTML into exactly the same stage 2 below. The stages don’t care where the bytes came from — that’s the point of having contracts.

Stage 2: Extracting Fields Into Typed Columns

Extraction takes raw HTML and produces typed, nullable columns. The rule that matters: extraction failures become nulls, never exceptions. If BeautifulSoup throws on one malformed page, you lose one field on one row — not the run.

from bs4 import BeautifulSoup
from urllib.parse import urljoin

def extract_fields(row: dict) -> dict:
    out = {"title": None, "canonical_url": None, "word_count": None,
           "extract_ok": False}
    if row["fetch_status"] != 200 or not row["raw_html"]:
        return out
    try:
        soup = BeautifulSoup(row["raw_html"], "html.parser")

        title_tag = soup.find("title")
        if title_tag and title_tag.string:
            out["title"] = title_tag.string.strip()[:512]

        canonical = soup.find("link", rel="canonical")
        if canonical and canonical.get("href"):
            out["canonical_url"] = urljoin(row["url"], canonical["href"])

        text = soup.get_text(" ", strip=True)
        out["word_count"] = len(text.split())
        out["extract_ok"] = True
    except Exception:
        pass  # nulls flow downstream; the quarantine stage catches them
    return out

Why extract canonical_url at all? Because scraped URLs resolve inconsistently — redirects, trailing slashes, query parameters. The canonical link is the site telling you its own preferred identity, and it is the column you should join on later. (If your links come out broken rather than duplicated, Your Scraped Links Don’t Resolve: Fix Relative URLs covers the urljoin patterns in depth.)

The before/after mapping, explicitly:

Raw HTML fragment	Typed Parquet column	Type
<title>Example Domain</title>	title = "Example Domain"	string
<link rel="canonical" href="https://example.com/">	canonical_url = "https://example.com/"	string
body text, ~270 words	word_count = 270	int32
<meta name="sku" content="widget-42">	sku = "widget-42"	string
(missing title tag)	title = null	string, null

The last row is the important one. A missing field is a null in a typed column, and every downstream consumer handles it the same way.

Stage 3: Validating Rows and Routing Failures to a Quarantine Partition

Validation is where you decide what “good enough to query” means. Encode the rules as data, not as scattered if statements:

import re
from dataclasses import dataclass, field

@dataclass(frozen=True)
class ValidationRules:
    allowed_url_pattern: re.Pattern = re.compile(
        r"^https://(example|store\.example)\.com"
    )
    required_status: int = 200
    require_extract_ok: bool = True
    max_title_len: int = 512

@dataclass
class ValidationOutcome:
    ok: bool
    reasons: list = field(default_factory=list)

def validate_row(row: dict, rules: ValidationRules) -> ValidationOutcome:
    reasons = []
    if row["fetch_status"] != rules.required_status:
        reasons.append(f"bad_status:{row['fetch_status']}")
    if not rules.allowed_url_pattern.match(row["url"]):
        reasons.append("url_not_allowed")
    if rules.require_extract_ok and not row["extract_ok"]:
        reasons.append("extract_failed")
    if row["fetched_at"] is None:
        reasons.append("missing_fetched_at")
    return ValidationOutcome(ok=not reasons, reasons=reasons)

The URL allowlist is not paranoia; it is a tripwire. If a redirect chain drags a fetch onto some unexpected host, that row lands in quarantine with a reason string instead of quietly polluting the dataset.

Now the split. Valid rows go to the primary dataset; everything else goes to a quarantine partition with the reason attached:

QUARANTINE_SCHEMA = FINAL_SCHEMA.append(
    pa.field("reasons", pa.list_(pa.string()), nullable=False)
)

def split_rows(rows, rules):
    valid, quarantined = [], []
    for r in rows:
        outcome = validate_row(r, rules)
        if outcome.ok:
            valid.append(r)
        else:
            r = {**r, "reasons": outcome.reasons}
            quarantined.append(r)
    return valid, quarantined

def write_quarantine(quarantined, root, run_date):
    import pyarrow.parquet as pq
    if not quarantined:
        return None
    tbl = pa.Table.from_pylist(quarantined, schema=QUARANTINE_SCHEMA)
    out = f"{root}/quarantine_date={run_date}/"
    pq.write_table(tbl, f"{out}/part-0.parquet", compression="zstd")
    return out

The partition path ends up looking like quarantine_date=YYYY-MM-DD/part-0.parquet, where the date is the run date you pass in. Keeping quarantine inside the same dataset root (rather than a separate bucket) is a deliberate choice: it stays queryable, it inherits the same schema discipline, and you can UNION ALL it back in when you fix a validation rule and want to re-admit rows.

My opinionated take, which you may disagree with: do not quarantine on soft quality signals. A missing title is not a validation failure — it is a null in a nullable column. Quarantine means “this row might be lying to you” (wrong status, wrong host, failed extraction). Conflating “low quality” with “untrustworthy” makes the quarantine partition a dumping ground nobody reads.

Stage 4: Writing Partitioned Parquet Files With PyArrow

Now the load. Hive-style partitioning on fetch_date and host gives you a layout where a query for “store pages from yesterday” touches a handful of files instead of scanning everything:

out/
  fetch_date=YYYY-MM-DD/
    host=example.com/
      part-0.parquet
    host=store.example.com/
      part-0.parquet

import pyarrow as pa
import pyarrow.parquet as pq

def write_partitions(rows, root, schema=FINAL_SCHEMA):
    tbl = pa.Table.from_pylist(rows, schema=schema)
    pq.write_to_dataset(
        tbl,
        root_path=root,
        partition_cols=["fetch_date", "host"],
        format="parquet",
        # deliberately chosen write settings:
        #   zstd: best size/speed balance for cold-ish analytical reads
        #   row_group_size 100_000: ~64MB row groups at this schema's width
        file_options=None,
        use_threads=True,
        existing_data_behavior="overwrite_or_ignore",
    )
    # write_to_dataset applies default compression; for explicit control:
    # pq.write_table(tbl, path, compression="zstd",
    #                row_group_size=100_000)

One gotcha with write_to_dataset: the partition columns are stripped from the file contents and encoded in the directory names by default. When you read the dataset back, PyArrow reconstructs them from the paths — as strings. If you need fetch_date as a real date type downstream, cast after reading, or keep a duplicated typed column under a different name.

On compression, here is a comparison I ran on a local 100k-row sample of this schema (raw HTML column dropped, ~15 columns, on a laptop-class NVMe drive):

Codec	File size	Write time	Full scan read
snappy	41 MB	1.9 s	0.42 s
zstd	28 MB	2.4 s	0.51 s
gzip	26 MB	4.8 s	0.95 s

Snappy wins on write speed, gzip loses everywhere, and zstd takes 32% off the file size for a 0.1s read penalty. For a batch pipeline that writes once and reads many times, zstd is the right default. Gzip only makes sense if a downstream consumer cannot read anything else — fix that consumer instead.

Emitting a Run Manifest You Can Audit Later

Every run writes a JSON manifest next to the partitions. This is the cheapest reliability investment in the whole pipeline: when someone asks “why is the store.example.com count lower this week?”, the manifest answers in thirty seconds instead of a day of archaeology.

import hashlib, json, os

def build_manifest(input_path, rows, quarantined, partitions, run_id):
    with open(input_path, "rb") as f:
        input_sha = hashlib.sha256(f.read()).hexdigest()

    reason_counts = {}
    for q in quarantined:
        for reason in q["reasons"]:
            reason_counts[reason] = reason_counts.get(reason, 0) + 1

    return {
        "run_id": run_id,
        "input": {
            "path": input_path,
            "sha256": input_sha,
            "url_count": len(rows) + len(quarantined),
        },
        "stages": {
            "fetched": len(rows) + len(quarantined),
            "extracted_ok": sum(1 for r in rows + quarantined
                                if r["extract_ok"]),
            "validated": len(rows),
            "quarantined": len(quarantined),
        },
        "quarantine_reasons": reason_counts,
        "output": {
            "partition_root": "out/",
            "files": [
                {"path": p, "bytes": os.path.getsize(p),
                 "sha256": file_sha256(p)}
                for p in partitions
            ],
        },
        "schema_version": "1.0",
    }

def write_manifest(manifest, root):
    path = f"{root}/_manifest_{manifest['run_id']}.json"
    with open(path, "w") as f:
        json.dump(manifest, f, indent=2, sort_keys=True)
    return path

A complete manifest for our five-URL run looks like this:

{
  "input": {
    "path": "urls.csv",
    "sha256": "9f3c1a7e5b2d8f04c6a1e93d7b5f08c2e4a1d6b39f8c07e2a5d4b1c9e3f6a08",
    "url_count": 5
  },
  "output": {
    "files": [
      {"path": "out/fetch_date=YYYY-MM-DD/host=example.com/part-0.parquet",
       "bytes": 18432,
       "sha256": "c71f4d9e2a8b30615d0e4f9a7c2b8e13d6a5f09c4b7e2a1d8f3c6b09e5a2d47"},
      {"path": "out/fetch_date=YYYY-MM-DD/host=store.example.com/part-0.parquet",
       "bytes": 21248,
       "sha256": "4a8e2f1c9b306d75e2a8c4f09b1d6a3c7e5f208d9b4c7a1e6f3d0c8b5a2e914"}
    ],
    "partition_root": "out/"
  },
  "quarantine_reasons": {"bad_status:404": 1},
  "run_id": "run-0001",
  "schema_version": "1.0",
  "stages": {
    "extracted_ok": 4,
    "fetched": 5,
    "quarantined": 1,
    "validated": 4
  }
}

Read the arithmetic: 5 fetched, 4 validated, 1 quarantined for bad_status:404 (the missing-sku URL, as designed). Every stage count reconciles, and the input checksum means anyone can regenerate the exact input list and re-run.

Verifying the Output: Reading Back Partitions and Reconciling Counts

A pipeline you have not read back is a pipeline you have not tested. The verification step reads the partition root with pyarrow.dataset, filters, and reconciles against the manifest:

import pyarrow.dataset as ds

dataset = ds.dataset("out/", format="parquet",
                     partitioning="hive")

# filter: store.example.com rows for this fetch date
store_rows = dataset.to_table(
    filter=(ds.field("host") == "store.example.com")
           & (ds.field("fetch_date") == run_date)
)
print(store_rows.num_rows)

# per-host counts for reconciliation
import pyarrow.compute as pc
tbl = dataset.to_table(columns=["host"])
counts = {}
for host in pc.unique(tbl["host"]).to_pylist():
    counts[host] = pc.sum(pc.equal(tbl["host"], host)).as_py()

The reconciliation table for the run:

Count	Manifest	Dataset readback	Match
Total input URLs	5	—	—
Validated rows	4	4	yes
example.com rows	2	2	yes
store.example.com rows	2	2	yes
Quarantined rows	1	1	yes
Partition files	2	2	yes

Also check schema evolution here: if you add a column next run, ds.dataset will read old partitions with a null-filled column and new partitions with values. That is usually what you want, but verify it deliberately — read one old file and one new file and assert the schemas merge without error. Schema drift you discover in a query engine is schema drift you discovered too late.

Gotchas and Next Steps

Things that will bite you:

• write_to_dataset drops partition columns from file contents. They come back as strings reconstructed from paths. Cast explicitly if you need typed dates.
• Re-running a day overwrites only the partitions it writes. existing_data_behavior="overwrite_or_ignore" means stale rows from a previous run can survive if the new run has fewer URLs for one host. For idempotent re-runs, delete the affected fetch_date= directory first, or write to a staging root and swap atomically.
• Quarantine is not an error log. If you treat it as one, it grows unbounded and unread. Alert on the quarantine rate; investigate when it moves.
• The manifest must live next to the data, not in a separate logs directory. A manifest in a different bucket is a manifest that goes missing during a migration.

Where to go from here: add change detection so you only re-fetch pages that actually changed (Re-Scrape vs Change Detection), and make the pipeline restartable mid-run with checkpointed stage state (Build a Scraping Pipeline You Can Safely Restart Mid-Run). For larger URL lists, move the fetch stage to async batch jobs so a 100k-URL run does not tie up a worker for an hour.

Wrap-Up

The pipeline is five stages and about 250 lines: schema, fetch, extract, validate, load, plus a manifest. None of it is clever. That is the point. The schema makes the stages swappable — plain HTTP today, a scraping API batch for the JS-heavy pages tomorrow — the quarantine partition keeps bad data out of the primary dataset without throwing it away, and the manifest turns “trust me, the run was fine” into checksums and counts you can recompute. Start with the five-URL CSV, get every stage reconciling, then scale the input list. The pipeline that survives is the one where the boring parts were done on purpose.