Skip to content

Pipeline

The pipeline module provides the execution framework for running sequential data processing steps with configuration, caching, and logging support.

The usage pattern for the pipeline is a bit different than the typical numbered scripts you might see elsewhere. There is no monolithic integrated script. Instead there is a standardized data processing pipeline that is configurable via YAML files and executed via a runner script.

There are three main components:

  • Setup
    • Each project has a runner, projects/<project>/run.py, as the point of entry, and
    • a pipeline configuration, projects/<project>/config.yaml
  • Pipeline Execution
    • The central Pipeline class defined in src/pipeline/pipeline.py
    • A set of modular processing steps, one package each under src/processing/
  • Data Models and Validation
    • Canonical data models and validation logic defined in src/data_canon/
    • This can be used standalone, but also is integrated into the pipeline steps for automatic validation
    • The data models define the expected schema and content for each data table (households, persons, days, trips, tours, vehicles, etc.) Each field carries its constraints, and each step declares the fields it reads and writes in @step(requires=..., produces=...).

Conceptual Diagram

%%{init: {"flowchart": {"rankSpacing": 30, "nodeSpacing": 18}}}%%
flowchart LR
  subgraph Pipeline[Pipeline steps, in run order]
    direction TB
    Setup(["projects/#lt;project#gt;/<br/>run.py + config.yaml"]) --> Load
    Load([load_data]) --> Clean(["clean_#lt;project#gt;"]) --> Impute([imputation]) --> Link([link_trips])
    Link --> Hab([detect_habitual_locations]) --> Joint([detect_joint_trips]) --> Tours([extract_tours])
    Tours --> Zones([add_zone_ids]) --> Comp
    Comp["<b>cascade_completeness</b><br/>• usable_ctramp<br/>• usable_daysim<br/>• usable_analysis"] --> Weights
    PUMS[/PUMS + sample plan/] --> Weights
    Weights["<b>compute_weights</b><br/>for each profile:<br/>seed → balance<br/>→ propagate"]
    Weights --> CT([format_ctramp]) --> Write([write_data])
    Weights --> DS([format_daysim]) --> Write
  end

  subgraph Tables[Canonical tables]
    direction TB
    H[[households]] ~~~ P[[persons]] ~~~ D[[days]] ~~~ U[[unlinked_trips]] ~~~ L[[linked_trips]] ~~~ T[[tours]] ~~~ J[[joint_trips]] ~~~ JT[[joint_tours]] ~~~ HL[[habitual_locations]]
  end

  Pipeline <-.-> Tables

  style Setup fill:#fff4e1,color:#000
  style Pipeline fill:#eef4fb,stroke:#999,color:#000
  style Tables fill:#f3e5f5,stroke:#999,color:#000
  style PUMS fill:#ffe0b2,color:#000
  style Comp text-align:left,fill:#e8f5e9,color:#000
  style Weights text-align:left,fill:#e8f5e9,color:#000
  style CT fill:#ffe0b2,color:#000
  style DS fill:#ffe0b2,color:#000

The profile names are the bats_2023 project's; each project declares its own under cascade_completeness. Each formatter keeps the records, and reads the weights, of the profile named after it.

Configuring a step

Each entry under steps: names a step and gives its settings under params:. The pipeline matches them to the step function's arguments by name:

  • Canonical tables (households, persons, ...) come from the pipeline's data, never from params:. A key naming a table is an error.
  • A plain argument (label: str) takes the key of the same name.
  • A settings model (an argument annotated with a pydantic model, such as matching: JointTripConfig on detect_joint_trips) is built from its fields. Write the fields flat under params:; each key goes to the model that owns that field. A block named after the argument works too, for grouping:
# Flat: each key goes to whichever model owns it
- name: detect_joint_trips
  params:
    method: mahalanobis
    covariance: [10000, 10000, 100, 100]
    confidence_level: 0.90

# Nested, where a step has several models (compute_weights takes census,
# geography, balancing and diagnostics)
- name: compute_weights
  params:
    census:
      state_fips: "06"
      pums_year: 2023
      # ...the rest of the census settings

The rules that keep a setting from going astray:

Situation Result
A key the step does not take, or a misspelt one Error, naming the step and the key
A setting with no default left out Error: the config must give it
A block and flat keys for the same model Error: use one or the other
A setting with a default left out The default applies

A step's settings models set extra="forbid", so a misspelt key inside a block is caught too. Two models in one step may not share a field name; that fails when the step is defined, not when it runs.

The step-level keys beside params: are validate_input, validate_output and cache. Each overrides the step's @step() value only when given.

The pipeline system allows you to:

  • Define processing steps as Python functions
  • Configure pipeline execution via YAML files
  • Cache intermediate results for faster re-runs
  • Track and log processing progress
  • Validate data between steps

The Pipeline class, caching, logging and the @step() decorator are in the Developer Reference.