Urban Transport Analytics with PySpark — Go North East (GNEL)

A Big Data coursework project that builds a service-compliance classification platform on the UK Bus Open Data Service (BODS). It ingests all four BODS catalogues for Go North East, reconstructs real-world delays by matching live GPS observations to scheduled stop times in PySpark, labels every trip compliant / non-compliant against the brief's ±2-minute urban Service Reliability tolerance, and compares four CrossValidator-tuned classifiers — surfaced through a Streamlit dashboard and a fully executed analysis notebook.

Student Information

• Manoj Bhatta
• AI37B -ID: 240562 -ST5011 CEM Big Data Programming Project

Stakeholders

• Go North East (operator): find routes/timeslots that quietly under-perform so schedule buffers go where they're needed.
• Nexus (regional transport authority / regulator): benchmark operator reliability objectively against the ≥85% on-time target.

Architecture

BODS API ──► src/ingestion/bods_harvester.py   (batch: TransXChange + NeTEx, streamed to CSV)
         ──► scripts/caching_daemon.py         (30s poll: SIRI-VM GPS + SIRI-SX, ON CONFLICT dedupe)
         ──► PostgreSQL bronze_* tables        (4 catalogues, 780,987 raw rows)
         ──► src/processing/etl_pipeline.py    (PySpark: nearest-stop GPS↔schedule join,
              500 m distance gate, −15/+45 min window, median trip delay,
              feature engineering, 4-partition repartition + MEMORY_AND_DISK cache)
         ──► transport_features.parquet        (668,650 stop-level rows — EDA / scale evidence)
         ──► trip_features.parquet             (3,881 trip-level rows — ML training grain)
         ──► src/ml/train_models.py            (4 classifiers, trip-keyed 3-fold CrossValidator,
              parallelism=4, zero-variance feature filter, class weights)
         ──► model_comparison.json / lr_coefficients.json / service_metrics.csv
         ──► dashboard/app.py                  (Streamlit, emerald design system, 5 pages)

Rendered diagrams: docs/mermaid_src/architecture.mmd and docs/mermaid_src/erd.mmd.

Data sources (4 BODS catalogues)

Catalogue	Format	Bronze table	Rows	How ingested
Timetables	TransXChange XML	bronze_timetable_events	629,985	Full GNEL dataset (220 lines), NOC-filtered, streamed row-by-row to CSV (no in-memory accumulation)
Fares	NeTEx XML	bronze_fare_products	32,227	Batch harvest
Locations (live GPS)	SIRI-VM	bronze_location_observations	118,282	scripts/caching_daemon.py polls every 30 s, parameterised ON CONFLICT upserts
Disruptions	SIRI-SX	bronze_disruption_summaries	493	Same daemon

Total: 780,987 raw records — ~8× the brief's 100,000-record threshold via multi-catalogue ingestion alone (no synthetic augmentation). The live feeds have no historical archive — the labelled-trip count grows only while the daemon runs.

Label construction (the heart of the project)

1. Each GPS observation is equi-joined to candidate scheduled stops on the same journey + line, then the geographically nearest stop wins via a window ranked by distance — O(n log n) per partition instead of the naive O(n×m) against the 630k-row timetable.
2. A match is only trusted if its implied delay is within −15/+45 min and the vehicle was within 500 m of the stop (equirectangular metres). Bad matches are nulled, never clamped — clamping would relabel matching noise as "on time".
3. A trip's delay is the median of its surviving observations; |median| ≤ 2 min ⇒ compliant. Training additionally requires ≥ 2 gated observations per trip.

Result on the current harvest: 3,881 trips observed → 3,831 labelled → 3,049 trustworthy training trips (52.4% compliant).

Route KPIs (service_metrics.csv)

Computed at trip grain, not stop grain — averaging a trip-constant label across ~100 near-identical stop rows would weight a route by how many stops its journeys have rather than by how many trips were observed.

• Service Reliability — % of a route's observed trips that are compliant. Measurable on all 195 routes carrying a labelled trip; network mean 48.8%.
• Headway Regularity (delay_stddev) — SD of gaps between successive observed departures on the same route + service date. Real gaps only, no default fill ⇒ measurable on 110 routes.
• Travel Time Variability — CV of observed ÷ scheduled running time, where observed = scheduled duration + median delay ⇒ measurable on 159 routes.
• Every statistic is NULL where unmeasurable, so "no evidence" stays distinct from "measured as zero". A one-trip route has no headway; that is not perfect regularity.
• Service Efficiency / completion rate is deliberately not reported. SIRI-VM only reports vehicles that are transmitting, so trips-completed ÷ trips-scheduled is not measurable from this feed. A previous completion_rate_pct derived from the (always-zero) disruption count was constant 100% for every route and has been removed rather than restated.

⚠️ 29 routes appear to clear the ≥85% target, but 19 of them rest on a single observed trip and 6 more on two (median trip count 1, vs 14 below target); the best of the 60 highest-volume routes reaches 75.0%. This harvest cannot certify any route as meeting the target — those routes are under-observed, not proven reliable.

ML formulation

• Unit of prediction: a trip (service_date + journey_code + line_name) — one row each in trip_features.parquet. The label is a trip-level property, so a random 80/20 split is group-safe by construction (no stop-level leakage to manage).
• Features (27 numeric + 2 categorical): temporal (incl. cyclical hour_sin/hour_cos), schedule structure (journey_stop_count, sched_duration_min, lag-window sched_headway_min), fares, disruptions, and leak-free historical priors — the route's and network's smoothed compliance rate from strictly earlier service dates only (rowsBetween(unboundedPreceding, -1) over per-date aggregates, m=10 smoothing).
• Leakage guards: delay_minutes and all its derivatives excluded; post-hoc observation counts excluded; CV folds assigned per trip via a deterministic hash passed to CrossValidator(foldCol=…); a zero-variance filter drops constant features (currently the six disruption features — BODS SIRI-SX returns no GNEL situations — and is_weekend, since the harvest window contains no weekend).
• Models (4), each 3-fold CV-tuned, parallelism=4:

Model	Accuracy	Precision	Recall	F1	ROC-AUC	Train (s)	Best params
Majority-class baseline	0.510	0.260	0.510	0.345	0.500	—	—
Logistic Regression	0.599	0.599	0.599	0.599	0.648	52.4	regParam 0.01, elasticNet 0
Decision Tree	0.566	0.564	0.566	0.565	0.500	18.7	maxDepth 3
Random Forest	0.580	0.582	0.580	0.580	0.626	38.7	100 trees, maxDepth 8
Gradient Boosted Trees	0.616	0.618	0.616	0.617	0.640	176.5	maxIter 50, maxDepth 3

Fit times are from a 2-core laptop-class container and will vary by machine; the metrics are deterministic (seed 42).

• Reading the result honestly: GBT wins at the 0.5 operating point (accuracy/F1); Logistic Regression wins ranking (ROC-AUC). That is not a contradiction — the informative features (historical priors, cyclical time-of-day) are smooth, monotone signals that a regularised linear model represents with one coefficient each, while shallow trees fragment them; and with ~2,500 training trips and ~34% of labels sitting within one minute of the ±2 threshold (high label noise from GPS matching), low-variance models rank better. The depth-3 Decision Tree emits only ~8 distinct scores, which is why its AUC collapses to chance even though its F1 is respectable.

Running the project

# 1. Start services (PostgreSQL, Jupyter/Spark, Streamlit)
docker-compose up --build -d

# 2. Harvest + seed everything (timetables, fares, and a realtime polling loop)
docker-compose exec jupyter python -m src.database.seed_db
#    ...or re-seed one catalogue without touching the daemon-accumulated realtime tables:
docker-compose exec jupyter python -m src.database.seed_db --only timetables

# 3. Keep live GPS/disruptions accumulating (long-running; more uptime = more labels)
docker-compose exec -d jupyter python scripts/caching_daemon.py

# 4. Spark ETL (writes both parquets + service_metrics.csv + Postgres service_metrics)
docker-compose exec jupyter python -m src.processing.etl_pipeline

# 5. Train + tune the 4 classifiers (exports model_comparison.json + lr_coefficients.json)
docker-compose exec jupyter python -m src.ml.train_models

# 6. Run the test suite (22 tests, no database or network required)
docker-compose exec jupyter python -m pytest tests -q

# 7. Execute the analysis notebook end-to-end (~10 min; all figures regenerate)
docker-compose exec jupyter bash -c "cd /app && PYTHONPATH=/app jupyter nbconvert \
  --to notebook --execute --inplace notebooks/classification_notebook.ipynb"

# Dashboard: http://localhost:8501   ·   Spark UI during jobs: http://localhost:4040

Big-data & performance engineering

• Partitions: every Spark session runs spark.default.parallelism=4 / 4 shuffle partitions; frames are repartitioned by route_id before grouped aggregations; JDBC reads use fetchsize=10000 + repartition(4) (a plain JDBC read is single-partition).
• Caching: the stop-level and trip-level frames persist at StorageLevel.MEMORY_AND_DISK; the notebook loads each catalogue once and reuses the cached frames across all phases. Spark UI evidence in docs/spark_ui_shots/ (Storage, Jobs, Stages, Environment).
• Broadcast joins: AQE automatically broadcasts the small fares/disruptions frames — visible as BroadcastHashJoin in the captured query plans.
• Single-pass statistics: null-profiling is one agg() per catalogue; the 11×11 correlation matrix uses pyspark.ml.stat.Correlation on an assembled vector (one distributed job, ~4 s) instead of pairwise stat.corr calls (~60 jobs).
• Training cost: the feature pipeline (StringIndexer → OneHotEncoder → VectorAssembler) is fit once; CrossValidator(parallelism=4) tunes all four models in ~154 s total.
• Memory: spark.driver.memory=4g — the 630k-row ingest overflows the 1 GB default heap in local mode (the driver hosts the executors).

Security & professional practice

• Parameterised SQL everywhere (batch loader and daemon) — no string-concatenated queries.
• Credentials from environment variables / .env, never hard-coded.
• Git version control; SparkSession and DB settings documented as code in src/config/settings.py and PipelineConfig.
• Ethics: vehicle-level GPS only (no passenger data → low GDPR exposure); low-volume routes are framed as under-observed, not unreliable, and the trip-level sample size is stated up front in every artefact.

Known limitations & future work

• Effective harvest window is short — the trainable trips span four service dates (2026-07-13 → 07-16), so network_prior_compliance behaves partly as a date indicator under a random split. The honest next step is a temporal split (train on earlier dates, test on the latest) alongside the random one; with only four dates there are not yet enough to run it meaningfully, which is itself an argument for more harvester uptime before drawing forecasting conclusions.
• Label noise bounds everything: ~34% of trips have a median delay within 1 minute of the ±2 boundary — GPS matching noise near the threshold caps achievable AUC regardless of model.
• Highest-value improvements, in order: weeks more harvester uptime (more dates → priors become a real time series, weekend/day-of-week effects become learnable); an autoregressive feature (the same vehicle/route's previous trip delay); target-encoding route_id for the tree models instead of a 114-column one-hot; probability calibration + PR-AUC reporting.

Repository layout

• src/ingestion/bods_harvester.py — BODS API harvester (all catalogues; streaming CSV writes; NOC row filter).
• src/ingestion/parsers.py — TransXChange / NeTEx / SIRI parsers shared by harvester and daemon.
• scripts/caching_daemon.py — 30 s live poller upserting deduplicated GPS/disruption rows into Postgres.
• src/database/seed_db.py — harvest + load bronze tables (--only to scope safely; restores realtime unique indexes after overwrites).
• src/processing/etl_pipeline.py — Spark ETL: joins, distance-gated delay reconstruction, features, priors, metrics, parquet outputs.
• src/ml/train_models.py — trip-level training of 4 tuned classifiers; exports comparison metrics + the LR scoring spec.
• src/config/settings.py — environment-driven configuration (DB, BODS API, paths).
• dashboard/emerald.py — light-mode emerald design system (tokens, sidebar nav, components, Plotly template).
• dashboard/app.py — the dashboard: Overview, Model Performance, Route Analytics, Compliance Predictor (Spark-free in-browser scoring with per-factor contributions), Live Feed.
• notebooks/classification_notebook.ipynb — the full executed analysis (ingestion → EDA → training → evaluation → business findings), section-numbered to match docs/listofdiagrams.md.
• scripts/generate_diagram.py / generate_erd.py — architecture + ERD figure generators.
• tests/ — 22 pytest cases over schedule-time parsing, NOC normalisation, the ±2.0 label boundary (including NULL-vs-zero for unobserved trips), and the route KPIs' "NULL where unmeasurable" guarantee. No database or network needed.
• scripts/build_report.py — regenerates docs/big_data_report.docx from the pipeline's own outputs, so the report cannot drift from the run that produced it.
• db/init.sql — bronze + metrics schemas.
• db/exports/ — database deliverables: schema_dump.sql (live pg_dump --schema-only), sample_data_dump.sql (50 rows per table as INSERTs), sample_queries.sql (11 validated analytics queries), schema.dbml (dbdiagram.io ERD source), lucidchart_erd.sql (paste into Lucidchart's ERD import to auto-draw the diagram).