Parallel GTFS Processing with multiprocessing

Split the batch by what each half is waiting for: fetch every feed with a thread pool, because downloading is IO-bound and threads overlap the waiting for free, then parse with a process pool, because parsing is CPU work that the interpreter lock serialises. Pass file paths between the stages rather than DataFrames — anything crossing a process boundary is pickled and copied, and a 600 MB frame costs more to move than it did to build. Give every task a timeout, and record failures per feed so one broken agency cannot stall the other thirty-nine. The wider batch design is covered in batch processing strategies for multi-agency feeds.

Root Cause Analysis

A multi-agency GTFS run is two entirely different workloads wearing one script.

Fetching is network-bound. Forty feeds at three to twenty seconds each is ten minutes of almost pure waiting, and the CPU is idle throughout. Threads solve it completely: the interpreter lock is released during socket waits, so twenty concurrent downloads use one core and finish in roughly the time of the slowest.

Parsing is CPU-bound. Reading, typing and validating a 600 MB feed is real computation, and the lock serialises it. Threads give no speed-up at all here — a common and disappointing discovery for anyone who tried threading the whole pipeline. Processes are required.

Running the whole batch under one strategy is therefore wrong whichever strategy is chosen. Threads everywhere means parsing runs on one core; processes everywhere means forty processes sit blocked on sockets, each holding an interpreter’s worth of memory for no reason.

Two further mistakes make a correct split perform badly anyway.

Moving data between processes. ProcessPoolExecutor pickles arguments and return values. Passing a parsed frame back from a worker serialises the whole thing, sends it through a pipe and rebuilds it in the parent — routinely slower than the parse itself. Workers should return summaries and write their bulk output to disk.

No isolation. One agency’s endpoint hangs, or one feed is pathological, and a pool slot is occupied indefinitely. Without a timeout and a per-feed exception boundary, the batch’s completion time is set by its worst member, and a crash in one worker can take the whole run down.

Production-Ready Python Implementation

"""Process many GTFS feeds in parallel: threads to fetch, processes to parse."""
from __future__ import annotations

import logging
import os
import time
from concurrent.futures import (FIRST_COMPLETED, ProcessPoolExecutor,
                                ThreadPoolExecutor, TimeoutError as FutureTimeout, wait)
from dataclasses import dataclass
from pathlib import Path

import requests

log = logging.getLogger("gtfs.batch.parallel")

FETCH_WORKERS = 16                 # IO-bound: far more than the core count is fine
FETCH_TIMEOUT_S = 120
PARSE_TIMEOUT_S = 900
# CPU-bound AND memory-bound. Each worker holds a whole feed, so this is sized
# by the largest feed against available memory, not by the core count alone.
PARSE_WORKERS = max(1, min((os.cpu_count() or 4) - 1, 8))


@dataclass(frozen=True)
class FeedSource:
    slug: str
    url: str


@dataclass
class FetchResult:
    slug: str
    path: Path | None
    bytes_read: int = 0
    seconds: float = 0.0
    error: str | None = None

    @property
    def ok(self) -> bool:
        return self.path is not None and self.error is None


@dataclass
class ParseResult:
    slug: str
    trips: int = 0
    stop_times: int = 0
    seconds: float = 0.0
    error: str | None = None

    @property
    def ok(self) -> bool:
        return self.error is None


def fetch_one(source: FeedSource, into: Path) -> FetchResult:
    """Runs in a THREAD: this waits on a socket and holds no lock while it does."""
    started = time.monotonic()
    destination = into / f"{source.slug}.zip"
    try:
        with requests.get(source.url, stream=True, timeout=FETCH_TIMEOUT_S) as response:
            response.raise_for_status()
            written = 0
            with open(destination, "wb") as fh:
                for chunk in response.iter_content(1 << 20):
                    fh.write(chunk)
                    written += len(chunk)
        return FetchResult(slug=source.slug, path=destination, bytes_read=written,
                           seconds=time.monotonic() - started)
    except Exception as exc:                       # one agency, one failure
        return FetchResult(slug=source.slug, path=None,
                           seconds=time.monotonic() - started,
                           error=f"{type(exc).__name__}: {exc}")


def parse_one(slug: str, archive_path: str, output_dir: str) -> ParseResult:
    """Runs in a PROCESS. Takes and returns only small values — never a DataFrame."""
    import pandas as pd
    from zipfile import ZipFile

    started = time.monotonic()
    try:
        with ZipFile(archive_path) as archive:
            with archive.open("trips.txt") as fh:
                trips = pd.read_csv(fh, dtype="string", encoding="utf-8-sig",
                                    keep_default_na=False, na_values=[""])
            with archive.open("stop_times.txt") as fh:
                stop_times = pd.read_csv(
                    fh, dtype="string", encoding="utf-8-sig",
                    keep_default_na=False, na_values=[""],
                    usecols=["trip_id", "stop_id", "stop_sequence",
                             "arrival_time", "departure_time"])

        # Bulk output goes to DISK; only the counts travel back through the pipe.
        out = Path(output_dir) / slug
        out.mkdir(parents=True, exist_ok=True)
        stop_times.to_parquet(out / "stop_times.parquet", index=False)
        trips.to_parquet(out / "trips.parquet", index=False)

        return ParseResult(slug=slug, trips=len(trips), stop_times=len(stop_times),
                           seconds=time.monotonic() - started)
    except Exception as exc:
        return ParseResult(slug=slug, seconds=time.monotonic() - started,
                           error=f"{type(exc).__name__}: {exc}")


def run_batch(sources: list[FeedSource], download_dir: Path,
              output_dir: Path) -> tuple[list[FetchResult], list[ParseResult]]:
    download_dir.mkdir(parents=True, exist_ok=True)
    output_dir.mkdir(parents=True, exist_ok=True)

    fetches: list[FetchResult] = []
    parses: list[ParseResult] = []
    started = time.monotonic()

    with ThreadPoolExecutor(max_workers=FETCH_WORKERS) as fetch_pool, \
            ProcessPoolExecutor(max_workers=PARSE_WORKERS) as parse_pool:

        fetch_futures = {fetch_pool.submit(fetch_one, s, download_dir): s
                         for s in sources}
        parse_futures = {}
        pending = set(fetch_futures)

        # Hand each feed to the parse pool AS SOON as its download finishes,
        # rather than waiting for the whole fetch stage to complete.
        while pending:
            done, pending = wait(pending, return_when=FIRST_COMPLETED)
            for future in done:
                if future in fetch_futures:
                    result = future.result()
                    fetches.append(result)
                    if not result.ok:
                        log.error("%s: fetch failed — %s", result.slug, result.error)
                        continue
                    log.info("%s: fetched %.1f MB in %.1fs",
                             result.slug, result.bytes_read / 1024 ** 2, result.seconds)
                    submitted = parse_pool.submit(parse_one, result.slug,
                                                  str(result.path), str(output_dir))
                    parse_futures[submitted] = result.slug
                    pending.add(submitted)
                else:
                    slug = parse_futures[future]
                    try:
                        result = future.result(timeout=PARSE_TIMEOUT_S)
                    except FutureTimeout:
                        result = ParseResult(slug=slug,
                                             error=f"exceeded {PARSE_TIMEOUT_S}s")
                    except Exception as exc:       # a worker that died outright
                        result = ParseResult(slug=slug,
                                             error=f"{type(exc).__name__}: {exc}")
                    parses.append(result)
                    if result.ok:
                        log.info("%s: parsed %d trip(s), %d stop time(s) in %.1fs",
                                 slug, result.trips, result.stop_times, result.seconds)
                    else:
                        log.error("%s: parse failed — %s", slug, result.error)

    elapsed = time.monotonic() - started
    ok = sum(1 for p in parses if p.ok)
    log.info("batch finished in %.1f min: %d/%d feed(s) processed, %d fetch failure(s)",
             elapsed / 60, ok, len(sources), sum(1 for f in fetches if not f.ok))
    return fetches, parses

Step-by-Step Walkthrough

Two pools, opened together, sized independently. FETCH_WORKERS is 16 because threads waiting on sockets are nearly free; PARSE_WORKERS is bounded by cores and by memory, because each process holds a whole feed. Sizing the parse pool by core count alone on a machine with eight cores and 16 GB, against feeds peaking at 1.5 GB, is how a batch runner gets killed by the memory manager.

Parsing starts as soon as a download finishes. The wait(..., FIRST_COMPLETED) loop keeps one set of pending futures containing both kinds, and submits each parse the moment its fetch returns. Waiting for the whole fetch stage first would leave every core idle for the length of the slowest download — a barrier with no purpose, since parsing one feed does not depend on any other feed.

parse_one takes strings and returns a small dataclass. Paths in, counts out. The parsed frames are written to Parquet inside the worker and never cross the process boundary. This is the difference between a batch that scales with cores and one that spends most of its time pickling.

parse_one imports pandas inside the function. Each worker process imports it once on first use rather than the parent paying for it before forking, which matters on spawn-based platforms where the parent’s imports are re-executed in every child.

Every task catches its own exceptions. fetch_one and parse_one both return a result object carrying an error string rather than raising. One agency’s malformed archive becomes one failed row in the report, and the other thirty-nine complete.

The parse timeout is enforced at collection. future.result(timeout=...) bounds how long the batch will wait for a worker that has stopped making progress. Without it, a pathological feed occupies a pool slot for the remainder of the run.

Verification and Output

def verify(sources: list[FeedSource], fetches: list[FetchResult],
           parses: list[ParseResult]) -> None:
    assert len(fetches) == len(sources), (
        f"{len(sources)} sources but {len(fetches)} fetch results — a future was lost")

    fetched_ok = {f.slug for f in fetches if f.ok}
    parsed = {p.slug for p in parses}
    assert parsed <= fetched_ok, "a feed was parsed that never fetched"
    assert fetched_ok - parsed == set(), (
        f"{sorted(fetched_ok - parsed)} fetched but never parsed")

    slugs = [p.slug for p in parses]
    assert len(slugs) == len(set(slugs)), "a feed was parsed twice"

    for p in parses:
        if p.ok:
            assert p.stop_times >= p.trips, (
                f"{p.slug}: fewer stop times than trips, which cannot be right")

The second and third assertions are the ones that catch orchestration bugs. A feed that fetched and never parsed means a future was dropped from the pending set; a feed parsed twice means it was submitted from two branches. Both are easy to introduce in a hand-rolled scheduler loop and invisible in the output otherwise.

A forty-feed run:

INFO gtfs.batch.parallel: mbta: fetched 61.4 MB in 8.2s
INFO gtfs.batch.parallel: septa: fetched 22.1 MB in 4.7s
INFO gtfs.batch.parallel: septa: parsed 12841 trip(s), 604112 stop time(s) in 41.3s
ERROR gtfs.batch.parallel: county_transit: fetch failed — ConnectTimeout: HTTPSConnectionPool(host='...', port=443)
INFO gtfs.batch.parallel: mbta: parsed 31042 trip(s), 1840112 stop time(s) in 126.8s
ERROR gtfs.batch.parallel: harbor_ferry: parse failed — KeyError: "There is no item named 'stop_times.txt' in the archive"
INFO gtfs.batch.parallel: batch finished in 13.2 min: 38/40 feed(s) processed, 1 fetch failure(s)

Thirteen minutes against roughly ninety-six for the same forty feeds processed one at a time, with two failures named and isolated rather than stopping the run.

Gotchas and Edge Cases

Spawn versus fork. On macOS and Windows the default start method is spawn, which re-imports the module in every child. Anything at module scope runs again per worker, so the entry point must be guarded with if __name__ == "__main__": or the pool recursively spawns itself.
Memory, not cores, is the binding constraint. Eight workers each holding a 1.5 GB feed needs 12 GB before the parent’s own usage. Where feeds are large, fewer workers each using chunked reading beats more workers swapping.
Logging from worker processes. Each has its own logging configuration and its own stdout. Either configure logging inside the worker or return the messages with the result; log lines emitted in a child and never configured simply vanish.
Politeness to agencies. Sixteen concurrent downloads against sixteen different agencies is fine. Sixteen against one agency’s server is not, and will get the batch blocked. Key any concurrency limit on the host, not on the feed.
Partial output on failure. A worker that dies after writing trips.parquet and before writing stop_times.parquet leaves an inconsistent directory. Write to a temporary directory and rename on success, so a failed feed leaves nothing rather than half of something.
Retries belong outside the pool. Retrying inside fetch_one holds a thread through the backoff. Collect the failures and run a second, smaller batch over them instead.

Frequently Asked Questions

Why threads for fetching and processes for parsing?

Because they are bound by different resources. Downloading waits on the network, and threads overlap that wait without paying for a process. Parsing is CPU work that the global interpreter lock serialises, so it needs separate processes to use more than one core.

Why not pass DataFrames between processes?

Because every object crossing a process boundary is pickled and copied. A 600 MB frame becomes 600 MB of serialisation on one side and 600 MB of allocation on the other, which usually costs more than the parsing did. Pass the file path and let the worker read it.

How large should the process pool be?

Around the core count, and bounded by memory rather than by cores. Each worker holds a whole feed, so eight workers on feeds peaking at 1.5 GB needs 12 GB before anything else. Size it by the largest feed, not the average.

What happens when one agency's feed hangs?

With a per-task timeout, that task is cancelled and recorded as a failure while the rest of the batch continues. Without one, a single unresponsive endpoint or a pathological feed stalls a pool slot indefinitely and the batch never completes.

Merging Multiple GTFS Feeds in Python — what happens after the feeds are parsed in parallel
Chunked Reading of stop_times in pandas — keeping each worker’s memory bounded
Up: Batch Processing Strategies for Multi-Agency Feeds — the batch model this parallelises
Section: Python Parsing & Data Normalization · Home