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.

Exactly-Once Writes Without Distributed Transactions #

A crawler cannot have a transaction spanning the fetch and the write, so “exactly once” has to be achieved differently: by making the write idempotent and the progress marker atomic with the data. Together they give a system that produces the same table whether it ran once or was interrupted and re-run five times.

Idempotency comes from a natural key and a merge rather than an append. The progress marker — a watermark recording how far the pipeline has consumed — must be committed in the same transaction as the rows it covers, or a crash between the two produces either duplicated work or silently skipped records.

BEGIN;

-- 1. Merge the batch on its natural key. Re-running with the same batch is a no-op.
INSERT INTO listings_curated AS c (natural_key, title, price_minor, currency,
                                   source_url, fetched_at, content_hash, schema_version)
SELECT natural_key, title, price_minor, currency,
       source_url, fetched_at, content_hash, :schema_version
FROM   listings_staging
ON CONFLICT (natural_key) DO UPDATE
SET    title         = EXCLUDED.title,
       price_minor   = EXCLUDED.price_minor,
       currency      = EXCLUDED.currency,
       source_url    = EXCLUDED.source_url,
       fetched_at    = EXCLUDED.fetched_at,
       content_hash  = EXCLUDED.content_hash,
       schema_version = EXCLUDED.schema_version
WHERE  c.content_hash IS DISTINCT FROM EXCLUDED.content_hash;   -- skip no-op updates

-- 2. Advance the watermark in the SAME transaction as the data it describes.
INSERT INTO ingest_watermark (pipeline, batch_id, committed_at)
VALUES (:pipeline, :batch_id, now())
ON CONFLICT (pipeline) DO UPDATE
SET batch_id = EXCLUDED.batch_id, committed_at = EXCLUDED.committed_at;

COMMIT;

The WHERE c.content_hash IS DISTINCT FROM EXCLUDED.content_hash clause is worth more than it looks. It converts a recrawl that found nothing new into zero row updates — no write amplification, no bloated table, no misleading updated_at timestamps, and no change events emitted to downstream consumers who would otherwise see every row “change” on every crawl.

For object-storage sinks the equivalent pattern is write-then-publish: write the new files under a temporary prefix, verify them, then atomically move or register them into the dataset. Readers never observe a partial write because the partial state was never at the published path.

Choosing the Natural Key #

The natural key is the most consequential schema decision at this stage, and it is hard to change later. Three properties matter: it must be stable across recrawls, unique within the dataset, and derivable from the record alone without consulting other rows.

The canonical URL is the usual starting point, but on its own it is often wrong: a listing page containing twenty records maps twenty records to one URL. The workable form is the canonical URL plus a record identifier drawn from the page — a product code, a listing reference, an anchor identifier. Where the page provides no such identifier, a hash of the identifying fields is the fallback, with the caveat that any change to those fields creates a new row rather than updating the old one.

What does not work is a key derived from volatile data: a position on the page, a timestamp, or a field that the site edits. Each produces a table that grows on every crawl while appearing to deduplicate correctly, and the divergence between row count and reality is usually discovered months later. The content-hashing approach covers how to build a stable hash when no natural identifier exists.

Two Sinks, One Source of Truth #

Most pipelines end up writing to at least two places: a transactional store that owns correctness, and an analytical store that answers questions quickly. The failure is treating both as authoritative, at which point they diverge and nobody can say which is right.

Nominate one source of truth — almost always the transactional store, because it is the one that supports targeted updates and deletions — and make every other sink a derived, regenerable projection of it. The projection is rebuilt from the source rather than written to in parallel, which means a bug in the write path corrupts one store instead of two, and a deletion applied to the source propagates by regeneration rather than by a second deletion someone has to remember.

Record the source’s watermark on each regenerated projection, so it is always possible to say which state of the truth a given projection reflects. Without it, a stale projection and a current one are indistinguishable, and a consumer querying the stale one has no way to know.

Where the analytical store is too large to regenerate wholesale, regenerate by partition: rebuild only the crawl dates affected by a change. That keeps the cost proportional to the change rather than to the dataset, and it is the main practical reason to partition by crawl date in the first place.

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.

Can an append-only sink be used for scraped records? #

Only as a projection, never as the source of truth. Append-only storage cannot satisfy a targeted deletion, which makes it unable to support an erasure request or an incident cleanup on its own. Where an immutable log is genuinely wanted — for reproducibility, or because a downstream consumer needs an event stream — keep it derived from a mutable source, give it a bounded retention, and make sure the deletion procedure includes regenerating or truncating it. An immutable store with an indefinite retention is a commitment, not an architecture choice.

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.