Batch Processing Strategies for Multi-Agency GTFS Feeds

Q: How do I avoid duplicate stop_id values when merging feeds from different agencies?

Prefix every entity ID with a normalised agency code before any join. For example, transform stop_id '42' from agency 'MTA' to 'MTA_42' immediately after ingestion. This prevents silent collisions when two agencies independently assign the same numeric identifier.

Q: What chunk size should I use when streaming stop_times.txt from a large metropolitan feed?

Start with 200,000 rows and monitor RSS with psutil. On a machine with 16 GB RAM processing a 5 million-row stop_times.txt, a 200k-row chunk typically peaks at 400–600 MB per worker. Reduce the chunk size if you run parallel workers, since each process maintains its own chunk in memory simultaneously.

Q: How should I handle agencies that omit calendar.txt entirely and use only calendar_dates.txt?

Detect the absence of calendar.txt during pre-validation and route service_id resolution exclusively through calendar_dates.txt. Build a date-indexed lookup that maps (service_id, date) to an active/inactive boolean, then join against this lookup rather than the regular calendar range logic.

Ingesting public transit data from a single operator is straightforward. Scaling to dozens of independent agencies — each with its own schema conventions, timezone offsets, calendar quirks, and update cadence — is a fundamentally different engineering problem. A production-grade multi-agency pipeline must reconcile divergent agency_id namespaces, resolve overlapping geographic boundaries, handle asynchronous feed publication cycles, and remain memory-safe when a single metropolitan feed exceeds 500 MB compressed. This guide describes a staged, idempotent architecture for consolidating GTFS static feeds from multiple providers into a unified analytical store, with runnable Python code at every step.

Prerequisites

Before building the pipeline, confirm your environment meets the following requirements. Install the Python dependencies with:

pip install pandas>=2.0 pyarrow>=12.0 pandera requests loguru

Required checklist:

Python 3.10+ — for structural pattern matching and improved zipfile streaming.
pandas>=2.0 with pyarrow>=12.0 — columnar intermediates and Parquet output.
pandera — declarative DataFrame schema enforcement at ingestion time.
requests — HTTP client for feed discovery and HEAD-based staleness checks.
loguru — structured JSON logging for per-feed audit trails.
GTFS files in scope: agency.txt, stops.txt, routes.txt, trips.txt, stop_times.txt, calendar.txt, calendar_dates.txt, shapes.txt (optional), feed_info.txt (optional).
Storage: High-throughput SSD or cloud object storage (S3 / GCS) for raw ZIP archives and Parquet outputs.
Compute: 4+ CPU cores; the extraction stage parallelises across a ProcessPoolExecutor.
Agency registry: A JSON or database table mapping agency_id → feed URL, agency_timezone, and last_fetched timestamp.

If you are new to GTFS feed ingestion, review Parsing GTFS with Pandas and Partridge for baseline loading patterns and foreign-key validation before adding multi-agency complexity.

GTFS Spec Background and Multi-Agency Data Model

The GTFS specification does not define a global namespace. Every agency assigns stop_id, route_id, trip_id, and service_id values independently, with no guarantee of uniqueness across providers. The spec’s only cross-agency linkage mechanism is the agency_id field in routes.txt and agency.txt — and even this is optional when a feed contains only one operator.

The table below summarises the files and keys relevant to multi-agency consolidation:

File	Primary key	Foreign keys	Multi-agency consideration
`agency.txt`	`agency_id`	—	Namespace root; must be prefixed before any join
`routes.txt`	`route_id`	`agency_id`	`route_id` collisions are common across feeds
`trips.txt`	`trip_id`	`route_id`, `service_id`, `shape_id`	`trip_id` values frequently overlap between agencies
`stop_times.txt`	(`trip_id`, `stop_sequence`)	`trip_id`, `stop_id`	Largest table; drives memory budget for the entire pipeline
`stops.txt`	`stop_id`	—	Intermodal stops appear multiple times with offset coordinates
`calendar.txt`	`service_id`	—	Agencies may omit this file and use `calendar_dates.txt` only
`calendar_dates.txt`	(`service_id`, `date`)	`service_id`	Exception semantics differ: agencies use `REMOVED` for unscheduled days
`shapes.txt`	(`shape_id`, `shape_pt_sequence`)	—	Optional; absent from many paratransit and dial-a-ride feeds

Key constraint: every entity identifier must be globally unique within your consolidated store before any cross-table join occurs. The standard approach is to prepend a normalised agency slug immediately on ingest — e.g., MTA_42 for stop_id 42 from agency MTA. Deferring this transformation until after joins produces silent cross-agency collisions that corrupt referential integrity checks downstream.

Pipeline Architecture

The diagram below shows the five stages of the multi-agency batch pipeline. Each stage writes its output to a durable intermediate before the next stage reads it, so any failed feed can be retried independently without reprocessing healthy ones.

Step-by-Step Implementation

Step 1: Feed Discovery and Pre-Validation

The pipeline opens by reading an agency registry — a JSON file or database table — and checking each feed for staleness before downloading. A HEAD request is far cheaper than pulling a 300 MB ZIP archive that has not changed since your last run.

import hashlib
import json
import requests
import zipfile
from pathlib import Path
from loguru import logger

REGISTRY_PATH = Path("agency_registry.json")
ARCHIVE_DIR = Path("raw_feeds")
ARCHIVE_DIR.mkdir(parents=True, exist_ok=True)


def load_registry() -> list[dict]:
    with REGISTRY_PATH.open() as f:
        return json.load(f)  # [{agency_id, feed_url, last_etag, agency_timezone}, ...]


def is_feed_stale(agency: dict) -> bool:
    """Return True when the remote feed has changed since our last download."""
    try:
        resp = requests.head(agency["feed_url"], timeout=10)
        resp.raise_for_status()
        return resp.headers.get("ETag", "") != agency.get("last_etag", "")
    except requests.RequestException as exc:
        logger.warning("HEAD request failed for {agency_id}: {exc}",
                       agency_id=agency["agency_id"], exc=exc)
        return False  # skip on connectivity errors; retry next run


def download_and_verify(agency: dict) -> Path | None:
    """Download the ZIP archive, verify it, and return the local path."""
    archive_path = ARCHIVE_DIR / f"{agency['agency_id']}.zip"
    try:
        resp = requests.get(agency["feed_url"], timeout=60, stream=True)
        resp.raise_for_status()
        sha = hashlib.sha256()
        with archive_path.open("wb") as out:
            for block in resp.iter_content(chunk_size=65_536):
                out.write(block)
                sha.update(block)
        # Verify basic ZIP integrity
        with zipfile.ZipFile(archive_path) as z:
            names = z.namelist()
            required = {"agency.txt", "stops.txt", "routes.txt", "trips.txt", "stop_times.txt"}
            missing = required - set(names)
            if missing:
                logger.error("Feed {id} missing required files: {missing}",
                             id=agency["agency_id"], missing=missing)
                archive_path.rename(archive_path.parent / "failed" / archive_path.name)
                return None
        logger.info("Downloaded {id} — SHA-256 {digest}",
                    id=agency["agency_id"], digest=sha.hexdigest()[:16])
        return archive_path
    except Exception as exc:
        logger.error("Download failed for {id}: {exc}", id=agency["agency_id"], exc=exc)
        return None

For teams already using Automating Feed Updates with GTFS-Kit, the staleness check and versioned archive storage described there can replace the is_feed_stale / download_and_verify pair above.

Step 2: Schema Normalisation and ID Prefixing

After download, each feed is parsed independently. All entity identifiers are prefixed with the agency slug before any data leaves the extraction worker. This is the single most important transformation in the pipeline — deferring it causes cross-agency trip_id collisions that are nearly impossible to detect after merging.

import pandas as pd
from concurrent.futures import ProcessPoolExecutor, as_completed

STOP_TIMES_DTYPES: dict[str, str] = {
    "trip_id": "string",
    "arrival_time": "string",
    "departure_time": "string",
    "stop_id": "string",
    "stop_sequence": "int32",
    "pickup_type": "Int8",
    "drop_off_type": "Int8",
    "shape_dist_traveled": "float32",
}

STOPS_DTYPES: dict[str, str] = {
    "stop_id": "string",
    "stop_name": "string",
    "stop_lat": "float32",
    "stop_lon": "float32",
    "location_type": "Int8",
    "parent_station": "string",
}


def prefix_ids(df: pd.DataFrame, id_columns: list[str], slug: str) -> pd.DataFrame:
    """Prefix all entity-ID columns with <slug>_ to guarantee global uniqueness."""
    for col in id_columns:
        if col in df.columns:
            df[col] = slug + "_" + df[col].astype("string").fillna("")
    return df


def extract_agency_feed(archive_path: str, agency_id: str, agency_timezone: str) -> dict[str, pd.DataFrame]:
    """
    Open a GTFS ZIP, load core tables with explicit dtypes, and prefix all entity IDs.
    Returns a dict of {table_name: DataFrame}.
    """
    slug = agency_id.upper().replace(" ", "_")[:8]  # e.g. "MTA"
    result: dict[str, pd.DataFrame] = {}

    with zipfile.ZipFile(archive_path) as z:
        # --- stops.txt ---
        with z.open("stops.txt") as f:
            stops = pd.read_csv(f, dtype=STOPS_DTYPES)
        stops = prefix_ids(stops, ["stop_id", "parent_station"], slug)
        stops["agency_id"] = agency_id
        result["stops"] = stops

        # --- stop_times.txt  (streamed in chunks for memory safety) ---
        with z.open("stop_times.txt") as f:
            chunks = []
            for chunk in pd.read_csv(f, dtype=STOP_TIMES_DTYPES, chunksize=200_000):
                chunk = prefix_ids(chunk, ["trip_id", "stop_id"], slug)
                chunks.append(chunk)
            result["stop_times"] = pd.concat(chunks, ignore_index=True)

        # --- trips.txt ---
        with z.open("trips.txt") as f:
            trips = pd.read_csv(f, dtype={"trip_id": "string", "route_id": "string",
                                           "service_id": "string", "shape_id": "string"})
        trips = prefix_ids(trips, ["trip_id", "route_id", "service_id", "shape_id"], slug)
        trips["agency_id"] = agency_id
        result["trips"] = trips

        # --- routes.txt ---
        with z.open("routes.txt") as f:
            routes = pd.read_csv(f, dtype={"route_id": "string", "agency_id": "string",
                                            "route_type": "Int16"})
        routes = prefix_ids(routes, ["route_id"], slug)
        routes["agency_id"] = agency_id
        result["routes"] = routes

    return result

Note that stop_times.txt is streamed in 200,000-row chunks, concatenated after prefixing, then returned. For feeds exceeding available RAM, write each chunk directly to a Parquet file instead of building chunks in memory — this is covered in Memory-Efficient Processing for Large Feeds.

Step 3: Parallel Extraction Across Agencies

CPU-bound CSV parsing across dozens of feeds is the natural fit for ProcessPoolExecutor. Each worker receives one agency’s archive path and returns its normalised DataFrames. A failed worker raises an exception that is caught per-future, so one malformed feed cannot stall the rest.

from concurrent.futures import ProcessPoolExecutor, as_completed
from loguru import logger


def run_parallel_extraction(
    registry: list[dict],
    archive_dir: Path,
    max_workers: int = 4,
) -> dict[str, dict[str, pd.DataFrame]]:
    """
    Fan out feed extraction to a process pool.
    Returns {agency_id: {table_name: DataFrame}}.
    """
    futures = {}
    results: dict[str, dict[str, pd.DataFrame]] = {}

    with ProcessPoolExecutor(max_workers=max_workers) as pool:
        for agency in registry:
            archive_path = archive_dir / f"{agency['agency_id']}.zip"
            if not archive_path.exists():
                logger.warning("Archive missing for {id}, skipping", id=agency["agency_id"])
                continue
            fut = pool.submit(
                extract_agency_feed,
                str(archive_path),
                agency["agency_id"],
                agency["agency_timezone"],
            )
            futures[fut] = agency["agency_id"]

        for fut in as_completed(futures):
            agency_id = futures[fut]
            try:
                results[agency_id] = fut.result()
                logger.info("Extracted {id} successfully", id=agency_id)
            except Exception as exc:
                logger.error("Extraction failed for {id}: {exc}", id=agency_id, exc=exc)

    return results

Step 4: Schema Validation with Pandera

Before merging anything into the consolidated store, validate each agency’s tables against a shared schema contract. Automating GTFS Version Control with Python Scripts describes how to track schema changes over time; here we enforce constraints at the row level.

import pandera as pa
from pandera.typing import Series


class StopsSchema(pa.DataFrameModel):
    stop_id: Series[str] = pa.Field(str_startswith="", nullable=False)
    stop_lat: Series[float] = pa.Field(ge=-90.0, le=90.0)
    stop_lon: Series[float] = pa.Field(ge=-180.0, le=180.0)
    agency_id: Series[str] = pa.Field(nullable=False)

    class Config:
        coerce = True
        drop_invalid_rows = False


class StopTimesSchema(pa.DataFrameModel):
    trip_id: Series[str] = pa.Field(nullable=False)
    stop_id: Series[str] = pa.Field(nullable=False)
    stop_sequence: Series[int] = pa.Field(ge=0)
    arrival_time: Series[str] = pa.Field(nullable=True)
    departure_time: Series[str] = pa.Field(nullable=True)

    class Config:
        coerce = True
        drop_invalid_rows = False


def validate_agency_tables(tables: dict[str, pd.DataFrame], agency_id: str) -> bool:
    try:
        StopsSchema.validate(tables["stops"])
        StopTimesSchema.validate(tables["stop_times"])
        return True
    except pa.errors.SchemaErrors as err:
        logger.error("Schema validation failed for {id}: {n} errors",
                     id=agency_id, n=len(err.failure_cases))
        return False

Step 5: Geographic and Temporal Reconciliation

Intermodal stops are the hardest reconciliation problem. Two agencies operating at the same rail hub often define the same physical stop with stop_lat/stop_lon values offset by 10–50 metres, and assign it entirely different stop_id values. Identifying these duplicates requires a spatial proximity check:

import numpy as np

EARTH_RADIUS_M = 6_371_000.0


def haversine_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    """Return great-circle distance in metres between two WGS84 points."""
    phi1, phi2 = np.radians(lat1), np.radians(lat2)
    dphi = np.radians(lat2 - lat1)
    dlam = np.radians(lon2 - lon1)
    a = np.sin(dphi / 2) ** 2 + np.cos(phi1) * np.cos(phi2) * np.sin(dlam / 2) ** 2
    return 2 * EARTH_RADIUS_M * np.arcsin(np.sqrt(a))


def flag_duplicate_stops(
    combined_stops: pd.DataFrame,
    radius_m: float = 50.0,
) -> pd.DataFrame:
    """
    Add a canonical_stop_id column that collapses stops within radius_m of each other.
    Uses a greedy nearest-neighbour approach suitable for <100k stops.
    """
    combined_stops = combined_stops.copy()
    combined_stops["canonical_stop_id"] = combined_stops["stop_id"]

    coords = combined_stops[["stop_lat", "stop_lon"]].to_numpy()
    canonical_map: dict[str, str] = {}

    for i, row_i in enumerate(combined_stops.itertuples(index=False)):
        if row_i.stop_id in canonical_map:
            continue
        for j in range(i + 1, len(combined_stops)):
            row_j = combined_stops.iloc[j]
            dist = haversine_distance(row_i.stop_lat, row_i.stop_lon,
                                      row_j["stop_lat"], row_j["stop_lon"])
            if dist <= radius_m:
                canonical_map[row_j["stop_id"]] = row_i.stop_id

    combined_stops["canonical_stop_id"] = (
        combined_stops["stop_id"].map(canonical_map).fillna(combined_stops["stop_id"])
    )
    return combined_stops

Temporal reconciliation addresses service calendar conflicts. Apply timezone normalisation before unifying calendars so that agencies in different timezones are all resolved to UTC date boundaries. Map agency_timezone from agency.txt to each feed’s calendar.txt records before computing the union of active service days.

Step 6: Consolidation and Parquet Output

The final step merges all validated, reconciled tables into a partitioned Parquet store and writes a manifest file for downstream consumers.

import pyarrow as pa
import pyarrow.parquet as pq
from datetime import datetime

OUTPUT_DIR = Path("parquet_store")
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)


def write_table_partition(
    df: pd.DataFrame,
    table_name: str,
    agency_id: str,
    service_date: str,
) -> Path:
    """Write a single agency/date partition to Parquet with ZSTD compression."""
    out_path = OUTPUT_DIR / table_name / f"agency_id={agency_id}" / f"date={service_date}" / "part-0.parquet"
    out_path.parent.mkdir(parents=True, exist_ok=True)
    arrow_table = pa.Table.from_pandas(df, preserve_index=False)
    pq.write_table(arrow_table, out_path, compression="zstd")
    return out_path


def write_manifest(
    agency_ids: list[str],
    row_counts: dict[str, int],
    run_timestamp: str,
) -> None:
    manifest = {
        "run_timestamp": run_timestamp,
        "agencies": agency_ids,
        "row_counts": row_counts,
        "parquet_root": str(OUTPUT_DIR),
    }
    (OUTPUT_DIR / "manifest.json").write_text(json.dumps(manifest, indent=2))

Validation and Verification

After the pipeline completes, verify consolidation correctness with these checks:

import pyarrow.dataset as ds

def verify_consolidation(expected_agencies: list[str]) -> bool:
    """Assert referential integrity and agency coverage in the Parquet store."""
    dataset = ds.dataset(OUTPUT_DIR / "stop_times", format="parquet", partitioning="hive")
    st_df = dataset.to_table(columns=["trip_id", "stop_id"]).to_pandas()

    trips_dataset = ds.dataset(OUTPUT_DIR / "trips", format="parquet", partitioning="hive")
    trips_df = trips_dataset.to_table(columns=["trip_id"]).to_pandas()

    # Every trip_id in stop_times must exist in trips
    orphaned_trips = set(st_df["trip_id"]) - set(trips_df["trip_id"])
    assert len(orphaned_trips) == 0, f"Orphaned trip_ids in stop_times: {len(orphaned_trips)}"

    # All expected agencies must have written data
    agencies_in_store = set(
        p.split("=")[1] for p in (OUTPUT_DIR / "stop_times").glob("agency_id=*")
        if "=" in str(p)
    )
    missing_agencies = set(expected_agencies) - {p.name.split("=")[1] for p in (OUTPUT_DIR / "stop_times").iterdir()}
    assert not missing_agencies, f"Agencies missing from store: {missing_agencies}"

    logger.info("Verification passed — {n} stop_time rows across {a} agencies",
                n=len(st_df), a=len(agencies_in_store))
    return True

Failure Modes and Edge Cases

Real agency feeds deviate from the spec in ways that silently corrupt multi-agency consolidations:

Omitted agency_id in single-operator feeds. When routes.txt lacks agency_id, it is inferred from the single row in agency.txt. Assign the inferred agency_id explicitly during normalisation; do not leave the column null, or cross-agency joins will drop those routes.
stop_times.txt times exceeding 24:00:00. The GTFS spec allows times like 25:30:00 for overnight trips that began the previous calendar day. Parse all time strings as timedelta objects (pd.to_timedelta), not datetime, to preserve arithmetic correctness. See Handling Frequency-Based vs Timetable Schedules for the canonical pattern.
Duplicate (trip_id, stop_sequence) pairs within a single agency. Some agencies re-use stop_sequence when a trip visits the same stop twice (e.g., loop routes). Detect and deduplicate with df.duplicated(subset=["trip_id", "stop_sequence"]) before writing; emit a structured warning rather than silently dropping rows.
Agencies that use only calendar_dates.txt and omit calendar.txt. Build your service-date resolution logic to handle both files independently. If calendar.txt is absent, resolve active dates purely from calendar_dates.txt rows where exception_type == 1. Treating a missing calendar.txt as an error causes the pipeline to reject valid feeds from dozens of US transit providers.
Non-UTF-8 encoded CSVs. Legacy feeds from some European and Latin American agencies use ISO-8859-1 or Windows-1252 encoding. Pass encoding="latin-1" as a fallback when the pd.read_csv call raises UnicodeDecodeError. Log the encoding override per feed for auditability.
ZIP archives containing subdirectories. The GTFS spec requires files at the archive root, but some agencies nest them inside a folder. After opening the ZIP, inspect ZipInfo.filename for path separators and normalise paths before reading.

Performance and Scale Notes

For pipelines processing 20+ metropolitan feeds:

Chunk size tuning. Start with chunksize=200_000 rows for stop_times.txt. On a 16 GB machine running 4 parallel workers, this targets ~400 MB peak RSS per worker. Reduce to 100,000 rows if you observe OOM kills; increase to 500,000 rows on machines with 32+ GB RAM to improve throughput.
Parquet partition sizing. Aim for 128–256 MB per Parquet file. Over-partitioning (e.g., one file per trip) creates thousands of small files that overwhelm metadata operations in query engines. Under-partitioning (a single file per agency) prevents predicate pushdown on service_date. The agency_id + service_date partition scheme in Step 6 hits the right balance for most regional pipelines.
Arrow interchange. Use pa.Table.from_pandas() rather than df.to_parquet() directly. The explicit Arrow conversion lets you attach schema metadata and control null encoding, preventing silent type promotion when different agencies disagree on whether pickup_type is an integer or a float.
Out-of-core joins for stop deduplication. The flag_duplicate_stops function above uses a greedy O(n²) loop, which becomes slow above ~50,000 unique stops. For larger networks, load the stops table into DuckDB and run the spatial proximity check as a self-join with ST_Distance; DuckDB executes this out-of-core without loading the full table into Python heap.
Observability. Emit one structured log line per agency per stage, including feed_id, stage, row_count, and wall-clock duration. Aggregate these with any JSON log collector to build a feed-health dashboard that shows which providers are chronically slow, missing files, or producing schema drift.

Frequently asked questions

How do I avoid duplicate stop_id values when merging feeds from different agencies?

Prefix every entity ID with a normalised agency slug immediately on ingest — before any join. Transform stop_id 42 from agency MTA to MTA_42 as the first operation inside the extraction worker. This prevents silent collisions when two agencies independently assign the same numeric identifier.

What chunk size should I use when streaming stop_times.txt from a large metropolitan feed?

Start with 200,000 rows and monitor RSS with psutil. On a 16 GB machine processing a 5 million-row stop_times.txt, a 200k-row chunk typically peaks at 400–600 MB per worker. Reduce chunk size if you run parallel workers, since each process maintains its own chunk in memory simultaneously.

How should I handle agencies that omit calendar.txt entirely and use only calendar_dates.txt?

Detect the absence of calendar.txt during pre-validation and route service_id resolution exclusively through calendar_dates.txt. Build a date-indexed lookup that maps (service_id, date) to an active/inactive boolean, then join against this lookup rather than the regular calendar range logic.

Related

Memory-Efficient Processing for Large Feeds — chunked streaming, dtype enforcement, and Parquet output for feeds exceeding 500 MB
Automating Feed Updates with GTFS-Kit — scheduled ingestion, delta detection, and versioned archive management
Handling Frequency-Based vs Timetable Schedules — normalising frequencies.txt and overnight time arithmetic before consolidation
Timezone Handling and Schedule Normalisation — resolving agency_timezone offsets before unifying multi-agency service calendars
Parsing GTFS with Pandas and Partridge — single-feed parsing fundamentals and foreign-key validation patterns

Up: Python Parsing & Data Normalization | Home