Structured Data Sinks and Warehousing #

A scraped record that lives only in a Python dict is worthless the moment the process exits. This technique covers the durable other end of the pipeline — picking the right structured sink and writing to it so that replays are safe, batches are efficient, and the schema can change without rewriting history. It belongs to the Pipeline Storage, Deduplication & Monitoring section, and it assumes its inputs have already passed the upstream typing gate: a record should reach a sink only after schema validation with Pydantic has confirmed its shape, because a database constraint is a brutal and expensive place to discover that a price is a string. The problem this solves is the gap between “I extracted some data” and “I have a queryable, deduplicated, legally-bounded dataset that survives a crashed batch.” Anyone running a crawl larger than a single run needs it.

Core Principles & Sink-Selection Foundation #

There is no single correct sink; there is a correct sink per access pattern. The four durable options that cover almost every scraping pipeline map cleanly onto how the data will be read.

The governing principle is to write raw once, immutably, and derive everything else. Land the original payload in object storage so you can always re-parse it when extraction logic improves; write validated, typed records into a relational or columnar curated tier for querying. Two further principles apply to every sink: writes must be idempotent, so a replayed batch changes nothing; and writes should be batched, because per-row round-trips dominate latency at any real volume. Partitioning — by crawl date, source domain, or both — is what keeps both query cost and deletion tractable, since expiring or erasing a partition is far cheaper than scanning a whole table.

Implementation Steps #

1. Land the raw payload immutably in object storage #

Before any parsing, write the original response to S3 under a partitioned, content-addressed key. This is your source of truth and your re-processing safety net.

import boto3, gzip, hashlib, datetime as dt

s3 = boto3.client("s3")

def land_raw(body: bytes, source_url: str, bucket: str) -> str:
    digest = hashlib.sha256(body).hexdigest()
    day = dt.datetime.utcnow().strftime("%Y-%m-%d")
    # Partition by day; content-address by hash so identical bodies collapse.
    key = f"raw/dt={day}/{digest}.html.gz"
    s3.put_object(Bucket=bucket, Key=key, Body=gzip.compress(body),
                  Metadata={"source_url": source_url})
    # Raw HTML may hold PII — a bucket lifecycle rule must expire this prefix.
    return key

2. Upsert typed records into Postgres on a natural key #

For the curated operational tier, upsert on a business key so a re-crawl updates in place rather than duplicating. The detailed patterns — conflict targets, partial indexes, and stale-write guards — are expanded in upserting scraped records into Postgres.

import psycopg2.extras

UPSERT = """
    INSERT INTO listings (listing_id, title, price, source_url, crawled_at)
    VALUES %s
    ON CONFLICT (listing_id) DO UPDATE SET
        title = EXCLUDED.title,
        price = EXCLUDED.price,
        crawled_at = EXCLUDED.crawled_at
    WHERE listings.crawled_at < EXCLUDED.crawled_at
"""

def upsert_listings(conn, rows: list[dict]) -> None:
    values = [(r["listing_id"], r["title"], r["price"],
               r["source_url"], r["crawled_at"]) for r in rows]
    with conn.cursor() as cur:
        # Idempotent: replaying the same batch is a no-op on unchanged rows.
        psycopg2.extras.execute_values(cur, UPSERT, values, page_size=500)
    conn.commit()

3. Write columnar snapshots to Parquet, partitioned #

For analytical exports, batch records into partitioned Parquet files. Columnar layout compresses well and reads fast for aggregate queries. The full PyArrow mechanics — writer options, row-group sizing, and dataset partitioning — are in writing scraped data to Parquet with PyArrow.

import pyarrow as pa
import pyarrow.parquet as pq

def write_parquet(rows: list[dict], root: str) -> None:
    table = pa.Table.from_pylist(rows)
    # Hive-style partitioning by crawl date makes retention deletes a directory op.
    pq.write_to_dataset(
        table, root_path=root,
        partition_cols=["crawl_date"],
        existing_data_behavior="overwrite_or_ignore",
    )

4. Load into the warehouse with MERGE, not append #

When pushing curated data to BigQuery or Snowflake, stage the batch and MERGE on the key so the warehouse copy stays idempotent and deduplicated in one statement.

MERGE INTO analytics.listings AS tgt
USING staging.listings_batch AS src
ON tgt.listing_id = src.listing_id
WHEN MATCHED AND src.crawled_at > tgt.crawled_at THEN
    UPDATE SET title = src.title, price = src.price, crawled_at = src.crawled_at
WHEN NOT MATCHED THEN
    INSERT (listing_id, title, price, source_url, crawled_at)
    VALUES (src.listing_id, src.title, src.price, src.source_url, src.crawled_at);

5. Evolve the schema additively #

Schemas drift as target sites change. Add columns as nullable; never repurpose an existing column’s meaning. In Postgres, ALTER TABLE ... ADD COLUMN ... NULL is cheap and non-locking for the common case; in Parquet and warehouses, readers tolerate new columns when old files simply lack them. Keep a versioned schema definition alongside the Pydantic model so a migration and a validation change land together.

Error Handling & Observability #

Distinguish errors you retry from errors you quarantine, and instrument both.

Emit a structured event per batch and export Prometheus metrics so a stalled or failing sink is visible immediately.

{
  "event": "sink_write",
  "ts": "2026-07-05T09:20:41Z",
  "sink": "postgres.listings",
  "operation": "upsert",
  "rows_in": 500,
  "rows_written": 493,
  "rows_skipped_stale": 7,
  "latency_ms": 214,
  "correlation_id": "b41c-77de"
}

Track scraper_sink_rows_written_total{sink,operation}, scraper_sink_write_latency_seconds{sink}, and scraper_sink_errors_total{sink,class}. Alert when the error counter for any sink rises above roughly 1% of writes over five minutes, or when write latency’s 95th percentile exceeds your batch SLA — both usually mean the sink is degraded or a schema change slipped through. Wiring these into dashboards is covered in crawl observability with Prometheus and Grafana.

A note on batching versus latency: larger batches amortise round-trip and transaction overhead, but they also widen the window in which a crash loses uncommitted work and delay the point at which a record becomes queryable. For most scraping pipelines a batch of a few hundred to a few thousand rows, flushed either on size or on a short timer, is the right balance — small enough to bound replay cost, large enough that per-row overhead disappears. Tie the flush to the same correlation ID threaded from the fetcher so a stalled batch is traceable to the crawl segment that produced it.

Compliance Boundaries #

A sink is where retention and lawful-basis obligations become concrete. Partition and tag data so you can prove what you hold and delete it on demand.

• Every table or dataset holding personal data has a documented retention period and lawful basis.
• Raw payload prefixes in object storage have lifecycle expiry rules configured.
• Personal-data tables are partitioned so an erasure request maps to a bounded, indexable set of rows.
• Writes carry a crawled_at and source_url Writes carry a `crawled_at` and `source_url` so every stored row has provenance for lineage and audit.
• Access to PII-bearing sinks is restricted and logged separately from anonymised aggregates.

On the maths: object storage at rest is cheap, but a raw HTML corpus containing PII is a liability regardless of price, so retention is a legal constraint before it is a cost one. The mechanics of enforcing whatever window you set — sweeping expired partitions and satisfying deletion requests — belong to data retention and GDPR deletion workflows.

Common Mistakes #

1. Auto-increment surrogate keys as the identity. Wrong: SERIAL PRIMARY KEY with no natural key, so every re-crawl inserts a fresh row. Right: a natural business key with ON CONFLICT upsert; the surrogate can exist but must not be what defines a duplicate.
2. Row-by-row inserts. Wrong: one INSERT per record in a Python loop. Right: batched execute_values or a staged bulk load — the difference is often two orders of magnitude in throughput.
3. Overwriting a column’s meaning during evolution. Wrong: reusing status to mean something new. Right: add a new nullable column and migrate readers, keeping old data interpretable.
4. Committing partial batches without idempotency. Wrong: a batch that crashes halfway leaves half its rows inserted and re-runs duplicate the rest. Right: idempotent upserts so a full replay is safe regardless of where the prior run died.
5. Landing raw HTML with no lifecycle rule. Wrong: an ever-growing S3 prefix of pages full of personal data. Right: partitioned keys plus an expiry policy from day one.

Frequently Asked Questions #

When should I use Parquet instead of just loading into a warehouse? #

Parquet on object storage is the cheaper choice when you want columnar analytics without paying for always-on warehouse storage, or when you need an open, portable format that many engines can read. A warehouse wins when you need concurrent SQL access, joins across many large tables, and managed governance. Plenty of pipelines write Parquet as the durable analytical tier and load only recent partitions into a warehouse for interactive queries.

How do I make a write idempotent if the source has no stable ID? #

Synthesise one. Build a deterministic key from the fields that define identity — often the source URL plus a stable subset of content — and hash it. That fingerprint becomes your conflict target, so the same logical record maps to the same key on every crawl. The content-hashing approach is detailed in deduplication strategies for scraped data.

Does upserting hurt performance versus plain inserts? #

Marginally, because the database checks the conflict target, but the cost is far smaller than the alternative of inserting duplicates and cleaning up later. Keep the conflict column indexed (a primary key or unique index already is), batch your upserts, and the overhead is negligible relative to network round-trips.

How do I handle a target site adding a new field mid-crawl? #

Add the column as nullable in the sink and to your Pydantic model in the same change, then deploy. Existing rows and older Parquet files simply carry a null for the new column, and readers that do not yet know about it are unaffected. Never block ingestion waiting for a schema migration — additive evolution lets the two proceed independently.

Related guides #

• Pipeline Storage, Deduplication & Monitoring — the parent section framing storage as a compliance surface.
• Writing Scraped Data to Parquet with PyArrow — columnar file mechanics and partitioning.
• Upserting Scraped Records into Postgres — conflict targets and stale-write guards in depth.
• Schema Validation with Pydantic — the gate that must pass before any record reaches a sink.
• Data Retention and GDPR Deletion Workflows — enforcing the retention rules your partitioning enables.