#Parallel Shapefile Conversion with Dask

When you need to convert thousands of Shapefiles to GeoParquet, the right pattern is to treat each file as an independent, idempotent task and fan those tasks out across a Dask cluster with a bounded submission window so the scheduler never holds more in-flight work than the workers can drain. In practice that means: build a flat work list of (source, destination) pairs, submit them as futures (or map them over dask.bag partitions), cap per-worker memory, and pull results with as_completed so slow or oversized files apply natural backpressure. This turns a multi-hour serial loop into a job whose wall-clock time is governed by your slowest partition, not the sum of every file. This page builds directly on the serial workflow in Building Batch Conversion Pipelines with Python and shows how to scale it horizontally without melting a worker.

#Quick-Reference: Choosing a Fan-Out Strategy

#How the Fan-Out Works

A Shapefile conversion is embarrassingly parallel: each .shp (plus its .dbf, .shx, and .prj siblings) reads and writes with no reference to any other file. The only shared resources are the object store you read from and write to, and the memory on each worker. Dask’s distributed scheduler is a good fit because it lets you express “run this function over these inputs” while it handles task placement, work stealing, and retries. The trap that catches most teams is submitting all N files as N futures at once: the scheduler materialises the entire graph, holds every result reference, and memory climbs on the client and the workers simultaneously. A bounded window — submit K, and only submit the next one as each completes — keeps the working set flat regardless of whether you convert one thousand or one million files.

Partitioning the work list is the lever that decides efficiency. If files are small and uniform, group them: a dask.bag with a few hundred partitions, each holding dozens of files, amortises the few-millisecond scheduler overhead per task across a meaningful chunk of I/O. If files vary by three orders of magnitude, per-file futures plus resource annotations let you send the giants to a high-memory worker pool while the small files stream through everywhere else. The same schema mapping and null-handling rules you apply serially still apply here — parallelism changes the scheduling, not the per-file semantics, so keep the conversion function pure and let it write exactly one output.

Per-worker memory is the constraint that most often turns a fast job into a flapping one. GeoPandas reads a Shapefile fully into a GeoDataFrame by default, so a 3 GB layer can spike a worker well past its limit during the WKB encode step. The defences are layered: set an explicit memory_limit per worker, keep each task single-threaded so a worker converts one file at a time, and read features in row batches with pyogrio rather than one monolithic read_file. When a worker exceeds its limit, Dask pauses it and may restart it — which is exactly why every task must be idempotent, writing to a temporary path and atomically renaming on success.

#Production Implementation

The function below converts one Shapefile to GeoParquet, and the driver fans the work list out with a bounded submission window using as_completed. Each task writes to a temporary path and renames on success so a retried task is safe.

# Requires: dask[distributed]>=2024.5, geopandas>=0.14, pyogrio>=0.7, pyarrow>=14.0
from __future__ import annotations

import logging
import os
from dataclasses import dataclass
from pathlib import Path

import geopandas as gpd
from dask.distributed import Client, as_completed

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class ConvertResult:
    source: str
    destination: str
    rows: int
    ok: bool
    error: str | None = None


def convert_one(src: str, dst: str, *, compression: str = "zstd") -> ConvertResult:
    """Convert a single Shapefile to GeoParquet. Idempotent: writes to a
    temp path and atomically renames, so a retried task is safe."""
    tmp = f"{dst}.tmp-{os.getpid()}"
    try:
        gdf = gpd.read_file(src, engine="pyogrio")
        gdf.to_parquet(tmp, compression=compression, index=False)
        os.replace(tmp, dst)  # atomic on POSIX same-filesystem writes
        return ConvertResult(src, dst, rows=len(gdf), ok=True)
    except Exception as exc:  # noqa: BLE001 - report, do not crash the worker
        if os.path.exists(tmp):
            os.unlink(tmp)
        logger.exception("Conversion failed for %s", src)
        return ConvertResult(src, dst, rows=0, ok=False, error=str(exc))


def run_batch(
    work: list[tuple[str, str]],
    client: Client,
    *,
    window: int = 64,
) -> list[ConvertResult]:
    """Fan out conversion with a bounded submission window for backpressure.

    Only `window` futures are ever in flight; a new file is submitted each
    time one completes, keeping scheduler and worker memory flat.
    """
    pending = iter(work)

    # Prime the window: at most `window` futures are ever in flight.
    futures: set = set()
    for _ in range(window):
        nxt = next(pending, None)
        if nxt is None:
            break
        futures.add(client.submit(convert_one, nxt[0], nxt[1], retries=2))

    results: list[ConvertResult] = []
    ac = as_completed(futures)
    for fut in ac:
        results.append(fut.result())
        nxt = next(pending, None)
        if nxt is not None:
            ac.add(client.submit(convert_one, nxt[0], nxt[1], retries=2))

    failed = [r for r in results if not r.ok]
    logger.info("Converted %d files, %d failed", len(results), len(failed))
    return results


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    client = Client(n_workers=8, threads_per_worker=1, memory_limit="4GB")
    src_root, dst_root = Path("s3://raw/shp"), Path("s3://lake/geoparquet")
    work = [
        (str(p), str(dst_root / f"{p.stem}.parquet"))
        for p in src_root.glob("*.shp")
    ]
    run_batch(work, client, window=64)

Two configuration choices carry most of the safety. threads_per_worker=1 guarantees a worker converts one file at a time, so its memory ceiling is one GeoDataFrame plus the write buffer, not several. memory_limit="4GB" gives Dask a hard budget to pause and restart against. The retries=2 on each submission covers transient object-store hiccups without hand-rolled retry loops; for a durable failure path that survives beyond the run, route exhausted tasks to a dead-letter queue.

#Validation

After a run, confirm that every input produced an output with a matching feature count. A cheap check reads the GeoParquet metadata footer rather than the whole file:

# Requires: pyarrow>=14.0, pyogrio>=0.7
import pyarrow.parquet as pq
from pyogrio import read_info

def verify(src_shp: str, dst_parquet: str) -> bool:
    src_rows = read_info(src_shp)["features"]
    dst_rows = pq.ParquetFile(dst_parquet).metadata.num_rows
    return src_rows == dst_rows

Expected ranges for a healthy cluster: with threads_per_worker=1 and 8 workers on typical 1–50 MB parcel layers, sustained throughput lands around 40–120 files/second, bounded by object-store read bandwidth rather than CPU. If per-worker memory ever crosses roughly 80% of memory_limit on small files, your window is too large or a giant file slipped into the general pool — check the dashboard’s memory panel. For deeper cost and throughput tracking across runs, wire the results into structured logging and metrics for migration jobs.

#Edge Cases and Caveats

The oversized outliers. A directory of 10,000 files where three are 4 GB national layers will stall a uniform pool: those three tasks pin three workers near the memory limit while everyone else idles. Detect file size up front, and submit the giants with resources={"MEM": 1} against workers launched with matching resource annotations so only the high-memory pool accepts them.

Object-store bandwidth as the real ceiling. Beyond a certain worker count, adding workers stops helping because every task competes for the same S3 or GCS egress. If throughput plateaus while CPU sits idle, you have found the network wall — shard reads across prefixes or provision more bandwidth rather than more workers.

Non-atomic writes to object storage. The os.replace trick is atomic on a POSIX filesystem but object stores emulate rename as copy-then-delete. On S3, prefer writing to a unique key and treating the successful multipart upload completion as the commit point; never let a partially written GeoParquet key shadow a good one.

#Frequently Asked Questions

Should I use dask.bag or the futures API for Shapefile conversion?

Use dask.bag when your file list is large and uniform and you want a concise map-over-partitions expression. Use the futures API when files vary wildly in size, when you need per-task retries, or when you want bounded submission with as_completed for backpressure. Both run on the same distributed scheduler, so you can start with a bag and drop to futures for the awkward remainder of oversized files.

How do I stop one giant Shapefile from killing a Dask worker?

Set an explicit worker memory limit, keep each task single-threaded so it reads one file at a time, and stream features in row batches rather than loading the whole GeoDataFrame at once. Route files above a size threshold to a dedicated high-memory worker pool. A worker that crosses its memory limit is paused and then restarted, so tasks must be idempotent to survive the retry.

Why is my Dask conversion slower than a single-process loop?

Almost always because tasks are too fine-grained or the workers contend on shared object storage bandwidth. Reading a 2 MB Shapefile has fixed scheduler overhead of a few milliseconds; if each task does only that, coordination dominates. Batch small files into partitions of dozens per task, and confirm your object store and network are not the real bottleneck before adding workers.

#Related

• Building Batch Conversion Pipelines with Python — the serial workflow this page scales out
• Automating Shapefile to GeoParquet Conversion — the per-file conversion logic in depth
• CI/CD Validation Hooks for GeoParquet Conversion — gate the outputs this job produces
• Dead-Letter Queues for Failed Migrations — where exhausted-retry tasks should land
• Structured Logging and Metrics for Migration Jobs — observability for parallel runs

← Back to Building Batch Conversion Pipelines with Python