Data Compression
  1. Home
  2. Data Conversion & Migration Pipelines for Cloud-Native Geospatial Storage
  3. Building Batch Conversion Pipelines with Python
  4. Memory-Safe Chunked Writes for Large Shapefiles
Section: Data Conversion & Migration Pipelines for Cloud-Native Geospatial Storage 15 min read

Memory-Safe Chunked Writes for Large Shapefiles

Read a bounded batch, convert it, write it as one row group, release it, repeat — with the schema fixed before the first byte is read. Done this way, converting a 40 GB shapefile needs a few hundred megabytes of memory and resumes from a checkpoint after a failure. Done the obvious way, it needs a machine three times the size of the input and starts over every time anything goes wrong. This page extends building batch conversion pipelines with Python.

Quick Reference

Decision Value Reason Primary Use Case
Batch size 50k–200k features Row group of 100–200 MB compressed Polygon layers
Schema source DBF field descriptors A late widening cannot fail the job Any large conversion
Writer lifetime One open writer, many batches One file, many row groups Single-output conversions
Checkpoint Feature offset after each batch Resume instead of restart Multi-hour runs
Memory budget Batch × expansion × 2 Read plus conversion held together Sizing the machine

Where the Memory Goes

A shapefile on disk is compact: fixed-width DBF records and packed coordinate arrays. In memory it is not. Every geometry becomes an object with a header, a reference to a coordinate buffer, and bookkeeping; every text attribute becomes a string object with its own header; the dataframe adds an index and per-column overhead. Two to three times the on-disk size is the normal expansion.

The failure is usually worse than that, because the obvious code path builds a list of records and then constructs a frame from it. For a moment both exist, and peak memory is roughly double the steady-state figure — which is exactly when the machine dies, at the very end of a long read, having done all the work and produced nothing.

Chunked writing removes both problems. Only one batch is resident at a time, the peak is bounded by the batch size rather than by the input size, and the transient doubling applies to a batch rather than to the whole layer.

Peak memory: read-then-write versus stream-and-flush Two memory traces over the duration of a conversion. The read-then-write trace climbs steadily as the whole layer accumulates, spikes sharply at the moment the dataframe is constructed from the record list because both exist simultaneously, and only then begins writing. The streaming trace is a low sawtooth: memory rises as a batch is read and converted, falls as the row group is flushed and the batch released, and repeats, never exceeding a few hundred megabytes regardless of input size. Resident memory converting a 40 GB shapefile read the whole layer, then write peak 118 GB at the frame construction list and frame both resident stream, flush, release — peak 340 MB time through the conversion → memory → The sawtooth is flat in the input size: the same code converts a 400 GB layer on the same machine.

Implementation

python
# Requires: pyogrio>=0.9, pyarrow>=16, shapely>=2.0, psutil>=6.0  (Python 3.10+)
from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path
from typing import Iterator

import psutil
import pyarrow as pa
import pyarrow.parquet as pq
import pyogrio


@dataclass(frozen=True)
class ConversionResult:
    features: int
    row_groups: int
    peak_rss_mb: float
    resumed_from: int


DBF_TO_ARROW = {
    "int32": pa.int32(), "int64": pa.int64(), "float64": pa.float64(),
    "object": pa.string(), "datetime64[ms]": pa.timestamp("ms"), "bool": pa.bool_(),
}


def schema_from_source(source: Path) -> pa.Schema:
    """Derive the full schema before reading any features.

    A Parquet file commits to one schema at the first row group, so a column
    that widens at feature 30 million would otherwise fail the job after hours.
    """
    info = pyogrio.read_info(source)
    fields = []
    for name, dtype in zip(info["fields"], info["dtypes"]):
        arrow_type = DBF_TO_ARROW.get(str(dtype))
        if arrow_type is None:
            raise ValueError(f"no Arrow type mapped for DBF field {name} ({dtype})")
        # Every attribute column is nullable: a DBF cannot distinguish an empty
        # string from a null, so the target schema must accommodate both.
        fields.append(pa.field(name, arrow_type, nullable=True))
    fields.append(pa.field("geometry", pa.binary(), nullable=True))
    return pa.schema(fields)


def batches(source: Path, schema: pa.Schema, size: int, start: int) -> Iterator[pa.Table]:
    """Yield bounded batches; only one is ever resident."""
    info = pyogrio.read_info(source)
    total = int(info["features"])
    offset = start
    while offset < total:
        frame = pyogrio.read_dataframe(
            source, skip_features=offset, max_features=size, use_arrow=True
        )
        if frame.empty:
            break
        table = pa.Table.from_pandas(frame.drop(columns="geometry"), preserve_index=False)
        table = table.append_column(
            "geometry", pa.array(frame.geometry.to_wkb(), type=pa.binary())
        )
        yield table.cast(schema)
        offset += len(frame)


def convert(
    source: Path,
    target: Path,
    *,
    batch_size: int = 100_000,
    memory_budget_mb: float = 4096,
    checkpoint: Path | None = None,
) -> ConversionResult:
    """Stream a very large shapefile into GeoParquet in bounded memory."""
    if not source.exists():
        raise FileNotFoundError(source)

    resume_from = 0
    if checkpoint and checkpoint.exists():
        resume_from = int(json.loads(checkpoint.read_text())["offset"])

    schema = schema_from_source(source)
    process = psutil.Process()
    peak = 0.0
    written = groups = 0

    writer = pq.ParquetWriter(target, schema, compression="zstd", compression_level=3)
    try:
        for table in batches(source, schema, batch_size, resume_from):
            writer.write_table(table, row_group_size=len(table))
            written += len(table)
            groups += 1

            rss = process.memory_info().rss / 1e6
            peak = max(peak, rss)
            if rss > memory_budget_mb:
                raise MemoryError(
                    f"resident memory {rss:.0f} MB exceeded the {memory_budget_mb:.0f} MB "
                    f"budget at feature {resume_from + written} — reduce batch_size"
                )
            if checkpoint:
                checkpoint.write_text(json.dumps({"offset": resume_from + written}))
            del table                       # release before the next read
    finally:
        writer.close()

    return ConversionResult(written, groups, peak, resume_from)

Validation

Assert the peak rather than hoping for it. A conversion that fits today will not fit after the source grows, and the failure mode is an OOM kill with no traceback.

bash
# Watch resident memory through a real run; the trace should be a flat sawtooth
/usr/bin/time -v python3 convert.py parcels.shp parcels.parquet 2>&1 \
  | grep -E 'Maximum resident set size|Elapsed'
# Maximum resident set size (kbytes): 348212     ← ~340 MB, independent of input size
# Elapsed (wall clock) time: 1:47:22
python
# Requires: pyarrow>=16 — the output should be many row groups, not one
import pyarrow.parquet as pq
meta = pq.read_metadata("parcels.parquet")
sizes = [meta.row_group(i).total_byte_size / 1e6 for i in range(meta.num_row_groups)]
print(f"{meta.num_row_groups} row groups, "
      f"{min(sizes):.0f}–{max(sizes):.0f} MB (median {sorted(sizes)[len(sizes)//2]:.0f})")

Expected results: peak resident memory a small multiple of one batch and flat in input size; row group sizes clustered in the 100–200 MB band with only the final group smaller; and a resumed run producing byte-identical output to an uninterrupted one.

What a checkpoint buys on a multi-hour conversion Two timelines for the same conversion interrupted at seventy per cent. Without a checkpoint the run restarts from feature zero and the completed work is discarded, so total elapsed time is the failed run plus a full second run. With a checkpoint recorded after each batch, the second run resumes from the last recorded feature offset and completes the remaining thirty per cent, so total elapsed time is the failed run plus a short remainder. A 2-hour conversion interrupted at 70% No checkpoint run 1 — 84 min, then killed run 2 — full 120 min from feature 0 204 min Checkpointed run 1 — 84 min, offset 24.1 M recorded run 2 — 36 min 120 min The checkpoint is one small JSON write per batch — the cheapest insurance in the pipeline. Checkpoint ordering, and which way to be wrong Two orderings around a crash. Writing the checkpoint before flushing the row group means a crash between the two leaves a checkpoint claiming progress that was never written, so the resumed run skips those features and they are lost silently. Writing the checkpoint after the flush means the same crash causes the resumed run to redo one batch, producing duplicated work but no loss. When in doubt, be wrong in the direction of duplication. Checkpoint before flush — data can be lost write checkpoint flush row group crash here between the two writes resumed run skips features that were never written — silent loss Checkpoint after flush — work can be repeated flush row group write checkpoint crash here between the two writes resumed run redoes one batch — duplicated work, nothing lost

Edge Cases and Caveats

A source whose feature count is wrong. Some shapefiles report a header count that disagrees with the actual record count, so an offset-based loop either stops early or reads past the end. Cross-check the header count against the .shx index size, and treat an empty batch as the true end of the layer rather than trusting the header.

Batch boundaries that split a logical group. If downstream processing assumes all features of one administrative area sit in one row group, an arbitrary feature-count batch will violate it. Either sort the source so groups are contiguous and batch on group boundaries, or drop the assumption — the quadtree leaf capacity approach makes the grouping explicit instead.

Checkpoints that do not match the output. A checkpoint written before the row group is flushed will, after a crash, cause the resumed run to skip features that were never written. Write the checkpoint strictly after the writer confirms the batch, as the implementation does, and prefer duplicated work over silent loss if the ordering is ever in doubt.

Frequently Asked Questions

Why does reading a 40 GB shapefile need far more than 40 GB of RAM?

Because the in-memory representation is much larger than the on-disk one. Each geometry becomes a Python or GEOS object with per-object overhead, attribute strings become Python strings with their own headers, and the dataframe holds index structures alongside the values. A two to three times expansion is typical, and building a dataframe from a list of records transiently holds both the list and the frame, which can double it again at the worst moment.

How large should each write batch be?

Large enough that the resulting row group lands in the useful range — typically 100 to 200 MB compressed — and small enough that a batch plus its Arrow conversion fits comfortably in the memory budget. For polygon data averaging a few kilobytes per feature that is usually 50,000 to 200,000 features. Derive it from measured bytes per feature rather than picking a round number.

Why must the schema be fixed before the first batch is written?

Because a Parquet file has one schema for all its row groups, and the writer commits to it when the first row group is written. If batch 400 contains a value that widens a column — a null in a column previously all-integer, or a string longer than expected — the writer cannot retroactively change the schema and the job fails after hours of work. Deriving the schema from the DBF field descriptors up front removes the possibility.

← Back to Building Batch Conversion Pipelines with Python