Choose Dask or multiprocessing for Parallel Schema Transforms Jump to heading#

Parallelizing a geospatial schema transform looks trivial until a single malformed partition either crashes the whole run or silently vanishes. The two default answers on the Python stack — the standard-library multiprocessing.Pool and Dask — optimize for different failure and scale regimes, and picking the wrong one costs you either a weekend of cluster plumbing you did not need or a memory wall you cannot climb. This comparison sits inside Batch Schema Processing Pipelines: that parent workflow owns the manifest contract and streaming execution model, and this page owns the concrete engine choice underneath it. It assumes the transform itself is already deterministic and picklable — if it is not, no scheduler will save you.

The short version: single-node multiprocessing wins on simplicity, startup latency, and debuggability for datasets that fit in one machine’s RAM; Dask wins when the working set exceeds memory (out-of-core), when you need a task graph with fan-in/fan-out, or when you will eventually move to a multi-node cluster. Everything below is a reproducible way to prove which regime you are actually in, rather than guessing.

Decision matrix Jump to heading#

Prerequisites checklist Jump to heading#

• pyproj >=3.6 and geopandas >=0.14 installed identically on every worker (version skew silently changes transform output).
• pyarrow >=14 available if partitions are exchanged as Arrow tables to keep type coercion consistent with pyarrow vs pandas for type coercion at scale.
• dask[distributed] >=2024.1 installed for the Dask arm (python -c "import dask; print(dask.__version__)").
• The transform function is pure and picklable — no open file handles, DB connections, or lambdas captured from an outer scope.
• A memory monitor available (psutil >=5.9) to sample peak RSS per worker.
• A writable path for the benchmark log and any partition that exhausts retries.

Step 1: Configure a shared transform contract Jump to heading#

Both engines must run byte-for-byte identical work, or the benchmark is meaningless. Wrap the schema transform in one pure function that takes a partition path and returns a small, serializable result (a row count plus a content hash), not the full GeoDataFrame — returning large frames across the process boundary measures pickle bandwidth, not your transform.

# geopandas >=0.14, pyproj >=3.6, pyarrow >=14 — Python 3.10+
import hashlib
from dataclasses import dataclass
from pathlib import Path

import geopandas as gpd


@dataclass(frozen=True, slots=True)
class PartitionResult:
    path: str
    rows: int
    digest: str          # stable hash of the normalized attribute table
    ok: bool
    error: str | None = None


def transform_partition(path: str) -> PartitionResult:
    """Pure, picklable unit of work run identically by both engines."""
    p = Path(path)
    try:
        gdf = gpd.read_file(p)
        gdf = gdf.rename(columns=str.lower)          # deterministic schema step
        gdf = gdf.to_crs(4326)                        # deterministic transform
        payload = gdf.drop(columns="geometry").to_csv(index=False).encode()
        digest = hashlib.sha256(payload).hexdigest()
        return PartitionResult(str(p), len(gdf), digest, ok=True)
    except Exception as exc:  # noqa: BLE001 — capture, never crash the pool
        return PartitionResult(str(p), 0, "", ok=False, error=repr(exc))

Returning a captured error instead of raising is deliberate: it makes error propagation symmetric across engines so the comparison in Step 4 is fair.

Step 2: Execute both engines under matched partitions Jump to heading#

Run the identical partition list through a fixed-size process pool and a Dask bag with the same worker count. Hold everything else constant — same input order, same max_workers, same machine, back to back.

# dask[distributed] >=2024.1 — Python 3.10+
import multiprocessing as mp
import time

import dask.bag as db
from dask.distributed import Client, LocalCluster


def run_multiprocessing(paths: list[str], workers: int) -> tuple[list, float]:
    t0 = time.perf_counter()
    with mp.get_context("spawn").Pool(processes=workers) as pool:
        # imap preserves input order and streams results as they finish.
        results = list(pool.imap(transform_partition, paths, chunksize=4))
    return results, time.perf_counter() - t0


def run_dask(paths: list[str], workers: int) -> tuple[list, float]:
    t0 = time.perf_counter()
    with LocalCluster(n_workers=workers, threads_per_worker=1,
                      memory_limit="2GiB") as cluster, Client(cluster):
        bag = db.from_sequence(paths, npartitions=workers * 4)
        results = bag.map(transform_partition).compute()
    return results, time.perf_counter() - t0

Use the spawn start method explicitly: fork (the Linux default) copies the parent’s memory and interacts badly with GDAL/PROJ global state, producing the intermittent segfaults teams blame on Dask when the real culprit is a forked pool. The Dask arm caps memory_limit so you can observe spill-to-disk rather than an OOM kill — the behavior that justifies Dask in the first place.

Step 3: Validate memory profile and determinism Jump to heading#

Throughput alone is a trap. Sample peak resident memory while each engine runs, and assert the two engines produce the same content. If the digests diverge, the transform is not actually deterministic and no throughput number matters.

# psutil >=5.9 — Python 3.10+
import psutil


def sample_peak_rss(pid: int, children: bool = True) -> int:
    """Peak RSS in bytes for a process tree; call repeatedly while running."""
    proc = psutil.Process(pid)
    total = proc.memory_info().rss
    if children:
        for child in proc.children(recursive=True):
            try:
                total += child.memory_info().rss
            except psutil.NoSuchProcess:
                continue
    return total


def assert_equivalent(mp_results: list, dask_results: list) -> None:
    """Both engines must agree on content, independent of completion order."""
    mp_map = {r.path: r.digest for r in mp_results if r.ok}
    dk_map = {r.path: r.digest for r in dask_results if r.ok}
    assert mp_map == dk_map, "engines disagree — transform is non-deterministic"

Note the ordering caveat: pool.imap yields in input order, but dask.bag.compute() does not guarantee element order across partitions. Compare by keyed dictionary, never by list position, or you will report a false determinism failure.

Step 4: Log the decision and error propagation behavior Jump to heading#

Record throughput, peak memory, and — critically — how each engine handled the partition that failed. A multiprocessing.Pool re-raises the first worker exception at result-collection time and can leave the pool in an undefined state; Dask isolates the failing key and lets independent branches complete. Because Step 1 captures errors into the return value, both engines here degrade gracefully, and failed partitions become records you can route to a dead-letter queue.

# Python 3.10+
import json
import logging

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
log = logging.getLogger("engine-bench")


def log_decision(engine: str, results: list, seconds: float, peak_bytes: int) -> dict:
    failed = [r.path for r in results if not r.ok]
    record = {
        "engine": engine,
        "partitions": len(results),
        "failed": failed,
        "throughput_per_s": round(len(results) / seconds, 2),
        "wall_seconds": round(seconds, 2),
        "peak_rss_mb": round(peak_bytes / 1e6, 1),
    }
    log.info("%s", json.dumps(record))
    return record

Route every path in failed to the structured queue described in building a dead-letter queue for failed feature transforms, and emit the whole record to your append-only lineage manifests so the engine choice and its failure surface are auditable, not tribal knowledge.

Benchmark-style numbers (with caveats) Jump to heading#

Indicative figures for ~10,000 mid-sized shapefiles (each 5–50 MB) on a 16-core / 64 GB node, transform as in Step 1. These are illustrative, not guarantees — they move with disk I/O, GDAL build, partition size, and PROJ grid caching. Measure your own workload with the harness above.

The pattern is consistent: multiprocessing is ~10–20% faster while the working set fits, and Dask’s flatter, disk-backed memory profile is what lets it finish the run that kills the pool. Below the memory wall, pay for simplicity; above it, pay for Dask.

Verification Jump to heading#

Confirm the harness is fair and the decision is grounded before you commit to an engine.

mp_res, mp_t = run_multiprocessing(paths, workers=8)
dk_res, dk_t = run_dask(paths, workers=8)

# 1. Same work, same result — non-negotiable.
assert_equivalent(mp_res, dk_res)

# 2. Both engines processed every partition (success or captured failure).
assert len(mp_res) == len(dk_res) == len(paths), "lost partitions"

# 3. The decision is data-driven, not assumed.
mp_rec = log_decision("multiprocessing", mp_res, mp_t, peak_bytes=mp_peak)
dk_rec = log_decision("dask", dk_res, dk_t, peak_bytes=dk_peak)
assert mp_rec["failed"] == dk_rec["failed"], "engines diverged on failures"

A healthy run logs two JSON records with identical failed lists and partitions counts. If multiprocessing peak RSS approaches physical RAM, that alone is the signal to move to Dask regardless of the throughput delta.

Troubleshooting Jump to heading#

Related Jump to heading#

• Batch Schema Processing Pipelines — the parent pipeline that this engine decision plugs into
• Batch Transforming 10k+ Shapefiles Without Memory Leaks — the streaming discipline that keeps either engine under its memory ceiling
• Building a Dead-Letter Queue for Failed Feature Transforms — where failed partitions from either engine are captured and replayed
• pyarrow vs pandas for Type Coercion at Scale — keeping the in-worker transform memory-lean and type-safe
• Lineage Manifest Generation — recording the engine, throughput, and failure surface as audit evidence