Compliant Web Scraping

Writing scraped data to Parquet with PyArrow #

Once a crawl produces more than a few thousand rows, appending them to CSV or JSON files becomes a liability: no column types, no compression, and full-file reads for every downstream query. Apache Parquet solves this with a columnar, self-describing, compressed on-disk format that analytical engines read selectively. This page shows how to persist scraped records to Parquet using PyArrow — defining an explicit schema, tuning row groups and compression, writing partitioned datasets, and reading the data back. It sits within the Structured Data Sinks and Warehousing technique in the broader Pipeline Storage, Deduplication & Monitoring section, and assumes the records arriving here have already been validated upstream.

Problem Framing #

A scraper emits a stream of loosely typed dictionaries — strings, nested lists, occasional nulls. If you dump those straight into Parquet without a declared schema, PyArrow infers types per batch, and the first batch that infers price as int64 will refuse to concatenate with a later batch that saw a decimal. Schema drift is the single most common failure in a Parquet write path.

On-disk size for one million scraped recordsColumnar encoding does most of the work; the codec only trims the rest.On-disk size for one million scraped recordsCSV, uncompressed1.4 GBCSV with gzip310 MBParquet with snappy180 MBParquet with zstd120 MB
Columnar encoding does most of the work; the codec only trims the rest.

The second recurring problem is file layout. Writing one Parquet file per HTTP response produces millions of tiny files that destroy read performance and overwhelm object-store listing APIs. You want a deliberate strategy: batch records in memory, flush at a target row-group size, and partition by a low-cardinality column such as crawl date or source domain so downstream queries can prune whole directories.

Step-by-Step Implementation #

  1. Declare an explicit pyarrow.schema. Pin every column’s type up front so batches are always compatible. Nullability and timestamp units matter — declare them once.
  2. Accumulate records into a RecordBatch or Table. Convert your list of dicts with pa.Table.from_pylist(records, schema=schema) so the declared schema is enforced rather than inferred.
  3. Choose a compression codec. snappy is fast and CPU-cheap; zstd compresses 20–40% smaller at a modest CPU cost and is the better default for cold warehouse storage.
  4. Write with a controlled row-group size so readers can skip irrelevant groups via column statistics.
  5. Partition by a pruning key using pyarrow.dataset.write_dataset so each partition value becomes its own directory.
Writing a partitioned dataset safelyPublishing after close is what stops readers seeing a half-written file.Writing a partitioned dataset safely1Build an explicitarrow schema2Batch records intorecord batches3Write partitionedby crawl date4Publish only afterthe write closes
Publishing after close is what stops readers seeing a half-written file.
import pyarrow as pa
import pyarrow.dataset as ds

# 1. Explicit schema — the contract for every batch written to this sink.
SCHEMA = pa.schema([
    pa.field("record_id", pa.string(), nullable=False),
    pa.field("source_domain", pa.string(), nullable=False),
    pa.field("title", pa.string()),
    pa.field("price", pa.decimal128(12, 2)),      # never float for money
    pa.field("currency", pa.string()),
    pa.field("scraped_at", pa.timestamp("us", tz="UTC")),
    pa.field("crawl_date", pa.date32()),           # partition key
])

def write_batch(records: list[dict], root: str) -> None:
    """Append one batch of scraped rows to a partitioned Parquet dataset.

    Compliance: `records` must already be schema-validated and stripped of
    fields with no lawful basis for retention before reaching this sink.
    """
    table = pa.Table.from_pylist(records, schema=SCHEMA)
    ds.write_dataset(
        table,
        base_dir=root,
        format="parquet",
        partitioning=ds.partitioning(
            pa.schema([("crawl_date", pa.date32())]), flavor="hive"
        ),
        existing_data_behavior="overwrite_or_ignore",
        max_rows_per_group=128 * 1024,   # ~128k rows per row group
        file_options=ds.ParquetFileFormat().make_write_options(
            compression="zstd", compression_level=3
        ),
        basename_template="part-{i}.parquet",
    )

The existing_data_behavior="overwrite_or_ignore" flag is the key decision point for appending versus rewriting. Parquet files are immutable — you cannot open a file and append rows to it. To add data you write new files into the dataset directory. Use overwrite_or_ignore to drop new files alongside existing ones (append semantics at the dataset level), or delete_matching to replace an entire partition atomically when you re-crawl a source. For deterministic filenames that avoid collisions across workers, give each writer a unique basename_template such as f"part-{worker_id}-.parquet".

Verification & Testing #

Confirm the round trip and inspect the physical layout before trusting the sink in production:

Columnar-write mistakes worth guarding againstEvery one of these passes local testing and fails at production scale.Columnar-write mistakes worth guarding against✗Schema inferred per batch instead of declared once✗One file written per record, producing millions of tiny files✗Timestamps written without a declared time zone✗Partition column also duplicated inside the row data✗Readers pointed at the directory while it is still being written
Every one of these passes local testing and fails at production scale.
import pyarrow.parquet as pq
import pyarrow.dataset as ds

# Read back only two columns and one partition — column + partition pruning.
dataset = ds.dataset("s3://crawl-warehouse/products/", format="parquet")
table = dataset.to_table(
    columns=["record_id", "price"],
    filter=(ds.field("crawl_date") == "2026-07-05"),
)
assert table.num_rows > 0

# Inspect a single file's metadata: row groups, compression, statistics.
meta = pq.read_metadata("part-0-0.parquet")
print(meta.num_row_groups, meta.row_group(0).column(0).compression)
assert meta.schema.to_arrow_schema().equals(SCHEMA, check_metadata=False)

If equals fails, a batch drifted from the contract — catch it here rather than at query time. You can also run parquet-tools inspect part-0-0.parquet from the shell to see per-column min/max statistics, which prove that predicate pushdown will be able to skip row groups.

Compliance & Operational Guardrails #

  • Do not persist fields you have no lawful basis to keep. Parquet’s compression makes it tempting to hoard columns “just in case”, but every retained personal-data field must map to a documented purpose. Apply schema validation with Pydantic before the write step so unauthorised fields never reach disk.
  • Partition on a key you can delete by. If subjects can request erasure, a crawl_date or source_domain partition lets you rewrite a narrow slice instead of the whole table; align this with your data retention and GDPR deletion workflows.
  • Store money as decimal128, timestamps as UTC. Float prices and naive local timestamps corrupt downstream aggregates and audit trails.
  • Keep row groups between 64k and 512k rows. Too small and you lose compression and statistics benefit; too large and readers cannot skip granularly.

Declare the Schema; Never Infer It #

Inference is the source of nearly every columnar-format problem in a scraping pipeline. Each batch is inferred independently, so a column that is all nulls in one batch becomes null type, an integer column becomes int64 in one file and double in another, and the dataset stops being readable as a whole — the failure appearing at read time, days later, in someone else’s query.

import pyarrow as pa
import pyarrow.parquet as pq

SCHEMA = pa.schema([
    pa.field("natural_key",    pa.string(),                 nullable=False),
    pa.field("source_url",     pa.string(),                 nullable=False),
    pa.field("fetched_at",     pa.timestamp("us", tz="UTC"), nullable=False),
    pa.field("title",          pa.string()),
    pa.field("price_minor",    pa.int64()),
    pa.field("currency",       pa.string()),
    pa.field("rating",         pa.float32()),
    pa.field("crawl_date",     pa.date32(),                 nullable=False),
    pa.field("schema_version", pa.string(),                 nullable=False),
])

def write_batches(records, out_dir: str, batch_size: int = 50_000) -> None:
    writer = pq.ParquetWriter(out_dir, SCHEMA, compression="zstd",
                              use_dictionary=True, write_statistics=True)
    try:
        buffer = []
        for record in records:
            buffer.append(record)
            if len(buffer) >= batch_size:
                writer.write_table(pa.Table.from_pylist(buffer, schema=SCHEMA))
                buffer.clear()
        if buffer:
            writer.write_table(pa.Table.from_pylist(buffer, schema=SCHEMA))
    finally:
        writer.close()          # the footer is written HERE — an unclosed file is unreadable

Passing schema=SCHEMA to from_pylist makes a type mismatch raise immediately, at the record that caused it, rather than producing a file that fails to merge much later. Timestamps carry an explicit time zone for the same reason: a naive timestamp is interpreted differently by different readers, and the resulting off-by-hours errors are subtle enough to survive review.

The finally: writer.close() is not defensive style — Parquet writes its schema and statistics footer on close, so a process that dies mid-write leaves a file that no reader can open. Combined with the publish-after-close pattern, this is what keeps a crashed run from corrupting a dataset.

Row Groups and File Sizes #

Two sizing decisions determine whether the dataset performs well. Row group size controls the granularity at which a reader can skip data using statistics: too small and the metadata overhead dominates, too large and predicate pushdown stops helping. Between 64 MB and 256 MB of uncompressed data per row group suits most analytical workloads.

File size matters more, because the classic failure is the small-file problem: a pipeline writing one file per crawl batch produces hundreds of thousands of tiny files, and listing the directory becomes slower than reading the data. Target files in the hundreds of megabytes, buffer across batches to reach that size, and run a periodic compaction that rewrites a partition’s small files into a few large ones. Partition by crawl date so compaction operates on closed partitions only, never on the one currently being written — the pattern described in partitioning scraped datasets by crawl date.

Common Mistakes #

  1. Letting PyArrow infer the schema per batch. Inference makes the first batch define the types; the next incompatible batch crashes the write. Always pass an explicit schema=.
  2. Writing one file per record or per request. The resulting small-file explosion cripples read performance. Buffer to a target row-group size before flushing.
  3. Using floating-point for prices or quantities. Binary floats cannot represent decimal cents exactly; use pa.decimal128.

Frequently Asked Questions #

Should I use snappy or zstd compression for scraped data? #

Use zstd (level 3) as the default for warehouse storage: it typically yields 20–40% smaller files than snappy with acceptable CPU cost, which matters when you are paying for object storage and cross-region transfer. Reach for snappy only when write latency is critical and you are re-reading the data within seconds, such as a short-lived intermediate stage between pipeline steps.

How do I append to an existing Parquet dataset without rewriting it? #

You never modify an existing Parquet file — you add new files to the dataset directory. Call pyarrow.dataset.write_dataset with existing_data_behavior="overwrite_or_ignore" and a unique basename_template per worker so new batches land as additional part-*.parquet files. Readers see the union of all files automatically. To replace a re-crawled partition atomically, use delete_matching scoped to that partition’s directory.