Error Logging and Data Quality Categorization in GTFS Pipelines

Problem Framing

Public transit feeds ingested without a structured validation layer become a liability. Silent failures — a foreign key mismatch between trips.txt and routes.txt, a stop_times.txt row with a non-UTF-8 byte sequence, a shape coordinate outside any plausible geographic bound — cascade into broken routing engines, inaccurate schedule displays, and unreliable mobility analytics. The challenge is not just detecting these problems but classifying them at the moment of detection so that on-call engineers can triage in seconds rather than hours: which feed failed, what kind of violation occurred, how many rows are affected, and whether the feed is safe to promote to production. A disciplined error logging and data quality categorization system transforms silent failures into traceable, actionable signals before they reach any downstream consumer.

This page covers the architecture, Python implementation, and operational practices for building that system — from structured JSON telemetry through tiered severity classification to CI/CD gating and agency feedback workflows.

Prerequisites

Before deploying a structured validation pipeline, confirm the following:

• Python 3.9+ with pandas>=2.0, pyarrow>=12.0, and standard library modules (logging, json, re, pathlib)
• GTFS files on disk: agency.txt, stops.txt, routes.txt, trips.txt, stop_times.txt, and either calendar.txt or calendar_dates.txt
• Familiarity with how pandas and the partridge library load GTFS tables with foreign-key enforcement at load time — this validation layer sits downstream of that read step
• A staging directory layout: raw/, extracted/, valid/, quarantine/, logs/
• An observability sink capable of consuming newline-delimited JSON (ELK, Datadog, CloudWatch, or a rotating file handler)

Install dependencies:

pip install "pandas>=2.0" "pyarrow>=12.0"

Concept and Specification Background

The GTFS specification distinguishes between Required, Conditionally Required, and Optional fields. This distinction maps directly onto severity tiers in a validation pipeline:

Beyond the spec’s own distinctions, anomalies fall into four domain categories that map to operational impact:

• Referential Integrity: Foreign key mismatches across routes.txt, trips.txt, stops.txt, and stop_times.txt. These break routing graph construction entirely.
• Temporal Consistency: Invalid service dates in calendar.txt, overlapping calendar_dates.txt exceptions, or stop_times.txt rows that violate chronological stop sequence order. These corrupt schedule displays and trip planners. Timezone normalization applied downstream amplifies any P1 temporal warnings from this stage into DST-related trip duplication.
• Spatial Accuracy: Coordinates outside the feed’s declared bounding box, duplicate stop_id values at zero-distance offsets, or shape points that regress in travel direction. These affect map rendering and coordinate reference system transforms.
• Format Compliance: Non-UTF-8 byte sequences, Windows-style CRLF line endings, trailing delimiters, or BOM markers that shift column header parsing.

The cross-product of severity tier and domain category gives a triage matrix. Engineering teams configure alerting thresholds per cell — suppressing P2 Spatial notices during off-hours but immediately paging on any P0 Referential failure.

Step-by-Step Implementation

Step 1: Configure the Structured Logger

Python’s native logging module supports custom Formatter subclasses that emit JSON. Moving away from free-text log messages is the single highest-leverage change in a validation pipeline: when an incident occurs at 02:00, parsing unstructured lines to determine which feed failed, which rule triggered, and how many rows are affected is slow and error-prone. JSON payloads enable immediate filtering and aggregation in ELK, Datadog, or CloudWatch without any log-parsing middleware.

import logging
import json
from datetime import datetime, timezone
from pathlib import Path


class GTFSLogFormatter(logging.Formatter):
    """Emit one JSON object per log record, enriched with GTFS pipeline metadata."""

    def format(self, record: logging.LogRecord) -> str:
        entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "feed_id": getattr(record, "feed_id", "unknown"),
            "domain": getattr(record, "domain", "general"),
            "severity_tier": getattr(record, "severity_tier", "P2"),
            "rule_id": getattr(record, "rule_id", None),
            "row_index": getattr(record, "row_index", None),
            "raw_value": getattr(record, "raw_value", None),
            "message": record.getMessage(),
            "logger": record.name,
        }
        return json.dumps(entry, default=str)


def build_gtfs_logger(log_path: Path, feed_id: str) -> logging.Logger:
    """
    Return a logger that writes one JSON line per validation event.
    The 'if not logger.handlers' guard prevents duplicate handlers
    in long-running process pools that call this function more than once.
    """
    logger = logging.getLogger(f"gtfs.validator.{feed_id}")
    logger.setLevel(logging.DEBUG)
    if not logger.handlers:
        handler = logging.FileHandler(log_path, encoding="utf-8")
        handler.setFormatter(GTFSLogFormatter())
        logger.addHandler(handler)
    return logger

Step 2: Define the Rule Engine

A rule engine decouples validation logic from ETL transformation. Each rule is a callable that receives a pd.DataFrame chunk and returns a boolean pd.Series mask — True for rows that pass. Attaching metadata to each rule (domain, severity tier, rule ID) keeps categorization automatic without manual tagging at each call site.

from dataclasses import dataclass
from typing import Callable
import pandas as pd
import re


@dataclass
class ValidationRule:
    rule_id: str
    domain: str          # "referential" | "temporal" | "spatial" | "format"
    severity_tier: str   # "P0" | "P1" | "P2"
    description: str
    check: Callable[[pd.DataFrame], pd.Series]


_TIME_PATTERN = re.compile(r"^\d+:[0-5]\d:[0-5]\d$")


def _is_valid_gtfs_time(series: pd.Series) -> pd.Series:
    """
    GTFS permits HH:MM:SS where HH >= 24 for post-midnight trips.
    Accept any non-negative integer hour with valid MM and SS ranges.
    """
    return series.astype(str).str.match(_TIME_PATTERN)


STOP_TIMES_RULES: list[ValidationRule] = [
    ValidationRule(
        rule_id="ST001",
        domain="temporal",
        severity_tier="P1",
        description="arrival_time must match HH:MM:SS (extended hours permitted for overnight trips)",
        check=lambda df: _is_valid_gtfs_time(df["arrival_time"])
        if "arrival_time" in df.columns
        else pd.Series(True, index=df.index),
    ),
    ValidationRule(
        rule_id="ST002",
        domain="temporal",
        severity_tier="P1",
        description="departure_time must not precede arrival_time on the same row",
        check=lambda df: (
            df["departure_time"].astype(str) >= df["arrival_time"].astype(str)
        )
        if {"arrival_time", "departure_time"}.issubset(df.columns)
        else pd.Series(True, index=df.index),
    ),
    ValidationRule(
        rule_id="ST003",
        domain="temporal",
        severity_tier="P0",
        description="stop_sequence must be a non-negative integer",
        check=lambda df: pd.to_numeric(df["stop_sequence"], errors="coerce").notna()
        if "stop_sequence" in df.columns
        else pd.Series(True, index=df.index),
    ),
]

Step 3: Verify Schema File Presence Before Row-Level Validation

Row-level rules are meaningless if mandatory files are absent. Perform file-presence and column header checks before opening any chunked reader. These are P0 failures — any missing mandatory file or required column means the entire ingestion run is aborted and quarantined. The GTFS validation rules and common schema errors reference covers which fields are Required versus Conditionally Required across every file in the spec.

MANDATORY_FILES: dict[str, list[str]] = {
    "agency.txt":     ["agency_name", "agency_url", "agency_timezone"],
    "stops.txt":      ["stop_id", "stop_name", "stop_lat", "stop_lon"],
    "routes.txt":     ["route_id", "route_type"],
    "trips.txt":      ["route_id", "service_id", "trip_id"],
    "stop_times.txt": ["trip_id", "arrival_time", "departure_time",
                       "stop_id", "stop_sequence"],
}

CALENDAR_ALTERNATIVES = {"calendar.txt", "calendar_dates.txt"}


def verify_schema(
    feed_dir: Path,
    logger: logging.Logger,
    feed_id: str,
) -> bool:
    """
    Confirm mandatory files exist and contain required columns.
    Returns False (and logs P0 events) on any violation.
    """
    passed = True
    present = {p.name for p in feed_dir.iterdir() if p.is_file()}

    if not CALENDAR_ALTERNATIVES.intersection(present):
        logger.error(
            "Neither calendar.txt nor calendar_dates.txt is present",
            extra={
                "feed_id": feed_id, "domain": "referential",
                "severity_tier": "P0", "rule_id": "SCHEMA001",
            },
        )
        passed = False

    for filename, required_cols in MANDATORY_FILES.items():
        if filename not in present:
            logger.error(
                f"Mandatory file missing: {filename}",
                extra={
                    "feed_id": feed_id, "domain": "format",
                    "severity_tier": "P0", "rule_id": "SCHEMA002",
                },
            )
            passed = False
            continue

        # Read only the header row to avoid loading entire file
        actual_cols = pd.read_csv(
            feed_dir / filename,
            nrows=0,
            encoding="utf-8-sig",
        ).columns.tolist()

        missing = set(required_cols) - set(actual_cols)
        if missing:
            logger.error(
                f"{filename} is missing required columns: {sorted(missing)}",
                extra={
                    "feed_id": feed_id, "domain": "format",
                    "severity_tier": "P0", "rule_id": "SCHEMA003",
                },
            )
            passed = False

    return passed

Step 4: Run Chunked Validation Against stop_times.txt

stop_times.txt is the largest file in almost every GTFS feed — regularly 10–100 million rows in metro-area archives. The same stops and stop_times relationships that make this table central to routing also make it the costliest to validate. Loading it fully into RAM before validating is the single most common cause of MemoryError in transit pipelines. Chunked validation keeps peak RSS bounded and enables early error detection on the first chunk without committing time to the full file. For dtype strategy and category encoding that further reduce per-chunk footprint, see optimizing pandas memory usage for transit feeds.

STOP_TIMES_DTYPES: dict[str, str] = {
    "trip_id":              "str",
    "arrival_time":         "str",   # Keep as str — GTFS allows HH >= 24
    "departure_time":       "str",
    "stop_id":              "str",
    "stop_sequence":        "str",   # Validated separately; coerce to int later
    "stop_headsign":        "str",
    "pickup_type":          "str",
    "drop_off_type":        "str",
    "shape_dist_traveled":  "str",
    "timepoint":            "str",
}


def validate_stop_times(
    stop_times_path: Path,
    rules: list[ValidationRule],
    logger: logging.Logger,
    feed_id: str,
    chunk_size: int = 50_000,
) -> tuple[list[pd.DataFrame], list[pd.DataFrame]]:
    """
    Validate stop_times.txt in chunks.

    Returns:
        valid_chunks: DataFrames containing rows that passed all rules
        quarantine_chunks: DataFrames with failed rows plus _error_chunk annotation
    """
    valid_chunks: list[pd.DataFrame] = []
    quarantine_chunks: list[pd.DataFrame] = []

    reader = pd.read_csv(
        stop_times_path,
        dtype=STOP_TIMES_DTYPES,
        keep_default_na=False,
        chunksize=chunk_size,
        encoding="utf-8-sig",   # strips BOM markers from Windows agency exports
    )

    for chunk_num, chunk in enumerate(reader):
        # Accumulate a combined pass mask across all rules for this chunk
        pass_mask = pd.Series(True, index=chunk.index)

        for rule in rules:
            rule_pass = rule.check(chunk)
            failed_rows = chunk.loc[~rule_pass]

            for row_idx, row in failed_rows.iterrows():
                logger.log(
                    logging.ERROR if rule.severity_tier == "P0" else logging.WARNING,
                    rule.description,
                    extra={
                        "feed_id":       feed_id,
                        "domain":        rule.domain,
                        "severity_tier": rule.severity_tier,
                        "rule_id":       rule.rule_id,
                        "row_index":     int(row_idx),
                        "raw_value":     row.to_dict(),
                    },
                )

            pass_mask &= rule_pass

        valid_chunks.append(chunk[pass_mask].copy())

        quarantine_chunk = chunk[~pass_mask].copy()
        if not quarantine_chunk.empty:
            quarantine_chunk["_error_chunk"] = chunk_num
            quarantine_chunks.append(quarantine_chunk)

    return valid_chunks, quarantine_chunks

Step 5: Write Quarantined Records to Parquet Atomically

Storing quarantined rows as Parquet alongside the valid subset enables atomic swaps, efficient predicate pushdown in audit queries, and compact on-disk footprints. The temp-then-rename pattern prevents partial reads by concurrent pipeline consumers. This integrates naturally with memory-efficient processing for large feeds where Parquet is already the preferred output format. For batch processing across multi-agency feeds, partition the quarantine dataset by feed_id as well so that per-agency audits scan only one Hive partition.

import pyarrow.parquet as pq


def write_validated_outputs(
    valid_chunks: list[pd.DataFrame],
    quarantine_chunks: list[pd.DataFrame],
    valid_path: Path,
    quarantine_path: Path,
) -> None:
    """
    Concatenate chunk lists and write to Parquet with atomic rename.
    Prevents partial reads during concurrent pipeline executions.
    """
    for chunks, out_path in [
        (valid_chunks, valid_path),
        (quarantine_chunks, quarantine_path),
    ]:
        if not chunks:
            continue
        combined = pd.concat(chunks, ignore_index=True)
        tmp = out_path.with_suffix(".tmp.parquet")
        combined.to_parquet(tmp, engine="pyarrow", index=False, compression="snappy")
        tmp.rename(out_path)   # atomic on POSIX filesystems

Validation and Verification

After running the pipeline, confirm output integrity before promoting feeds to production:

def audit_outputs(
    valid_path: Path,
    quarantine_path: Path,
    original_row_count: int,
    logger: logging.Logger,
    feed_id: str,
) -> bool:
    """
    Verify that valid + quarantine row counts reconcile with the raw feed.
    Returns True if counts balance and no P0 records appear in the valid set.
    """
    valid_df = pd.read_parquet(valid_path) if valid_path.exists() else pd.DataFrame()
    quarantine_df = (
        pd.read_parquet(quarantine_path)
        if quarantine_path.exists()
        else pd.DataFrame()
    )

    recovered = len(valid_df) + len(quarantine_df)
    if recovered != original_row_count:
        logger.error(
            f"Row count mismatch: expected {original_row_count}, got {recovered}",
            extra={
                "feed_id": feed_id, "domain": "format",
                "severity_tier": "P0", "rule_id": "AUDIT001",
            },
        )
        return False

    logger.info(
        f"Audit OK: {len(valid_df)} valid rows, {len(quarantine_df)} quarantined",
        extra={"feed_id": feed_id, "domain": "general", "severity_tier": "P2"},
    )
    return True

Supplementary verification checklist:

• Confirm valid/stop_times.parquet has no null trip_id values: assert valid_df["trip_id"].notna().all()
• Confirm quarantine/stop_times.parquet contains an _error_chunk column for lineage tracing
• Referential integrity spot-check: all trip_id values in the valid set must appear in trips.txt
• Check stop_sequence is monotonically increasing within each trip_id group in the valid set
• Verify assert len(valid_df) + len(quarantine_df) == original_row_count before any downstream job proceeds

Failure Modes and Edge Cases

Real agency feeds expose edge cases that specification-only rules will not anticipate:

• Times beyond 24:00:00: GTFS explicitly permits arrival_time and departure_time values such as 25:30:00 for trips that start before midnight and arrive the following calendar day. A regex anchored to ^[0-2]\d: will incorrectly quarantine valid post-midnight rows. The _is_valid_gtfs_time function above accepts any non-negative integer hour.
• BOM markers in UTF-8 exports: Several Windows-based agency scheduling tools export CSV files with a UTF-8 BOM (\xef\xbb\xbf). This shifts the first column header by three bytes, causing required-column checks to fail. Use encoding="utf-8-sig" in every pd.read_csv call to strip the BOM transparently.
• Duplicate stop_sequence within a trip: Some agencies repeat the same stop_sequence integer for two consecutive rows — commonly when a route loops back through a stop. This violates the spec but does not trigger a foreign-key error. Add a rule that groups by trip_id and checks stop_sequence uniqueness within each group after chunked loading completes.
• Empty agency_id in single-agency feeds: The spec permits omitting agency_id when only one agency is present. Referential checks that join routes.txt on agency_id will raise key errors against a feed that legitimately leaves the column blank. Check the row count in agency.txt before enforcing the join. The agency metadata and feed versioning practices guide covers how to detect and handle this pattern during ingestion.
• calendar_dates.txt-only feeds: Many point-in-time event feeds and holiday-modified schedules omit calendar.txt entirely. Schema verification must accept either file, not require both. The CALENDAR_ALTERNATIVES check in verify_schema handles this pattern.
• Non-monotonic stop_sequence with valid gaps: GTFS requires only that sequences be non-negative integers in increasing order within a trip — not that they increment by 1. Rules that flag gaps (stop_sequence[n+1] != stop_sequence[n] + 1) will produce false P1 warnings on valid feeds. Check strictly for decrease, not gaps.

Performance and Scale Notes

For context on chunked reading strategies and dtype enforcement that keep memory bounded, see memory-efficient processing for large feeds. Several additional considerations apply specifically to validation workloads:

• Chunk size tuning: For stop_times.txt files exceeding 1 GB, start with chunksize=100_000 and reduce if validation rules involve groupby operations across chunk boundaries. stop_sequence monotonicity checks require buffering the last row of each trip_id across chunk edges.
• Parallel agency processing: When running validation across a multi-agency batch, launch one multiprocessing.Process per feed rather than per file. The if not logger.handlers: guard in build_gtfs_logger prevents handler duplication across forked processes.
• Parquet partitioning for quarantine archives: Partition quarantine outputs by severity_tier using pyarrow.parquet.write_to_dataset with Hive-style partitioning so audit queries can scan only the P0 partition.
• CI/CD gating: In a GitHub Actions or Jenkins pipeline, parse the JSON log file after each run and exit with a non-zero code if any record has severity_tier == "P0". Zero-tolerance P0 gating prevents feeds with broken referential integrity from ever reaching a routing engine.
• Feed size profiling before extraction: Read zipfile.ZipFile.infolist() to get compressed and uncompressed sizes before extracting. Log a P0 format event and abort ingest if the uncompressed total exceeds your available RAM headroom. This is faster than OOM-killing a running process and produces a structured log entry rather than a kernel OOM message.

For teams running frequent feed rotations, integrating this validation step with automating feed updates with gtfs-kit creates a continuous quality gate from download through publication. Pairing structured validation with consistent agency metadata and feed versioning practices ensures that quarantine records carry enough provenance — feed version, download timestamp, source URL — to trace errors back to a specific agency export.

Related

• Memory-Efficient Processing for Large GTFS Feeds — chunked reading and Parquet output patterns that underpin large-feed validation
• Automating Feed Updates with gtfs-kit — integrate validation into scheduled download and publication pipelines
• Batch Processing Strategies for Multi-Agency Feeds — parallel feed ingestion and per-agency quarantine partitioning
• GTFS Validation Rules and Common Schema Errors — specification-level validation rules and schema error taxonomy
• Optimizing Pandas Memory Usage for Transit Feeds — dtype enforcement and category encoding to shrink in-memory DataFrame footprint

Up: Python Parsing & Data Normalization | Home