Building OmniRoute: A Real-Time Fleet Intelligence Platform on Google Cloud

A data engineering deep-dive into designing, building, and operating an end-to-end GCP data pipeline for fleet operations, safety compliance, and financial analytics — on a shoestring compute budget.

1. Introduction

OmniRoute Smart Logistics Engine is a data engineering platform built to simulate and solve the kinds of problems a real fleet-management company faces every day: tracking which driver is in which vehicle, catching drivers who speed or enter restricted zones, flagging vehicles that burn too much fuel, and translating all of that into automatic pay deductions.

What makes this build interesting isn't just the logic — it's the constraint it was built under. This was explicitly a training/educational project with a hard minimum-cost mandate, which shaped almost every architectural decision, from picking a classic Dataproc cluster over Dataproc Serverless to running Kafka ourselves on a small VM instead of reaching for a managed streaming service.

This post walks through the architecture, the business requirements that drove it, the commands and permissions that got it running on GCP, representative code, and where the system stands today.

2. Architecture

The system is split into five layers — ingest, process, store, orchestrate, and visualize — all running on Google Cloud.

Python Event Generator
        │
        ▼
Kafka (Docker, KRaft mode) — VM: kafka-event-generator
  topics: fleet.gps.events | fleet.fuel.events | fleet.safety.events
        │
        ▼
Dataproc classic cluster (omniroute-cluster, n1-standard-2, 2 workers)
  PySpark Structured Streaming + batch jobs
        │
        ├──► GCS (raw Parquet landing, partitioned by event_date / vehicle_id)
        │
        └──► BigQuery dataset: omniroute_dwh (fact tables → gold layer)
        │
        ▼
Apache Airflow 3.x (Docker Compose) — VM: airflow-server
  orchestrates ingestion, SCD2 merges, cooldown jobs
        │
        ▼
Apache Superset — VM: superset-bi (port 8088)
  BI / reporting layer

Core components:

Layer	Technology	Notes
Ingest	Python event generator → Kafka (Docker, KRaft mode)	VM `kafka-event-generator`, internal IP `10.142.0.2:9092`
Process	Dataproc classic cluster `omniroute-cluster`	`n1-standard-2`, 2 workers, `us-east1-b`
Store	GCS + BigQuery	Raw Parquet landing + `omniroute_dwh` dataset
Orchestrate	Apache Airflow 3.x	Docker Compose on `airflow-server` VM
Visualize	Apache Superset	Port 8088 on `superset-bi` VM

A key early decision was rejecting Dataproc Serverless. On paper it looks like the "modern" choice for Spark on GCP, but its hard resource floors — 750GB+ of disk and a 12+ core minimum — made it fundamentally incompatible with a minimum-cost training project. A small classic cluster with two n1-standard-2 workers was the only way to keep this affordable while still running real Structured Streaming and batch jobs.

3. System Requirements & Business Objectives

The platform processes real-time vehicle telemetry and daily batch data across four core domains:

Asset Lifecycle Management — Maintain an accurate, historical record of vehicle-driver assignments using SCD Type 2 logic: when a new assignment arrives, the previous record is closed (end_date set, status → ARCHIVED) and a new record is opened (status → IN-TRANSIT). Where duplicate records exist for the same vehicle and timeframe, the highest daily rate wins, resolved via a ROW_NUMBER() window function.
Operational Efficiency — Detect vehicles whose fuel consumption exceeds the fleet baseline by more than 12%, excluding weekends and scheduled maintenance days, so drivers aren't penalized for idling at a workshop.
Safety Compliance — Real-time streaming detection of speeding (>110 km/h) and geofencing violations (restricted-zone intersection), each logged as a Safety Strike.
Financial Accountability — Each strike deducts 5% from a driver's daily rate. Strikes and rates reset monthly via a cooldown job, except for drivers who've accumulated 10 strikes, who are marked SUSPENDED and excluded from that reset.

Data flows in two speeds: real-time telemetry (GPS, fuel, safety events) via Kafka, and daily batch reference data (vehicle registry, vehicle-driver assignment, maintenance schedules) landed as CSVs and merged into the warehouse.

4. Commands Used

A sample of the commands that actually stood up and operated the infrastructure:

Standing up Kafka on the VM (KRaft mode, no ZooKeeper):

docker run -d \
  --name kafka \
  -p 9092:9092 \
  -e KAFKA_NODE_ID=1 \
  -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://10.142.0.2:9092 \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  apache/kafka:4.3.1

Submitting the streaming job to Dataproc, with Ivy/Maven resolution disabled (Dataproc workers have no internet access, so connector jars must be pre-staged in GCS):

gcloud dataproc jobs submit pyspark \
  gs://omniroute-scripts-project-f8c7a8ef-e885-466f-954/sql/kafka_to_gcs_streaming.py \
  --cluster=omniroute-cluster \
  --region=us-east1 \
  --jars=gs://omniroute-scripts-.../kafka-clients-3.4.1.jar,\
gs://omniroute-scripts-.../spark-sql-kafka.jar,\
gs://omniroute-scripts-.../spark-token-provider-kafka-0-10.jar,\
gs://omniroute-scripts-.../commons-pool2.jar \
  --properties=spark.jars.packages=""

Fixing an immutable VM access scope (scopes can't be changed on a running instance, and gcloud compute instances set-scopes needs the alpha/beta track):

gcloud compute instances stop kafka-event-generator --zone=us-east1-b
gcloud compute instances set-service-account kafka-event-generator \
  --zone=us-east1-b \
  --service-account=617809984627-compute@developer.gserviceaccount.com \
  --scopes=cloud-platform
gcloud compute instances start kafka-event-generator --zone=us-east1-b

Deleting (never just stopping) the Dataproc cluster to avoid orphaned disk billing:

gcloud dataproc clusters delete omniroute-cluster --region=us-east1

5. Permissions Required

Because org policy blocks service account key creation (constraints/iam.disableServiceAccountKeyCreation), every piece of this pipeline authenticates through Application Default Credentials (ADC) and attached service accounts — no downloaded JSON keys anywhere in the system.

The Compute Engine default service account, 617809984627-compute@developer.gserviceaccount.com, needed roles across the stack:

BigQuery: roles/bigquery.dataEditor (write fact/dimension tables), roles/bigquery.jobUser (run load/query jobs), and later roles/bigquery.readSessionUser to enable the BigQuery Storage Read API path used by the fuel-efficiency and safety-violation Spark jobs.
GCS: roles/storage.objectAdmin on the scripts and staging buckets, for reading SQL/PySpark scripts and writing raw Parquet and CSV drops.
Dataproc: roles/dataproc.worker on the cluster service account so jobs can execute and write intermediate state.
Compute: access scopes set to cloud-platform on the ingest and orchestration VMs (via set-service-account, since scopes are immutable at runtime).

6. Code Snippets

Generating a simulated safety event (Python producer → Kafka):

def gen_safety_event():
    event_subtype = random.choices(
        ["harsh_braking", "harsh_acceleration", "overspeeding", "idle_too_long"],
        weights=[30, 25, 30, 15]
    )[0]
    return {
        "event_type": "safety",
        "event_subtype": event_subtype,
        "vehicle_id": random.choice(VEHICLE_IDS),
        "driver_id": random.choice(DRIVER_IDS),
        "severity": random.choice(["low", "medium", "high"]),
        "speed_kmph": round(random.uniform(0, 140), 1) if event_subtype == "overspeeding" else None,
        "timestamp": datetime.now(timezone.utc).isoformat()
    }

Structured Streaming fan-out — one Kafka read, multiple sinks. A single subscribePattern on fleet.*.events covers all three topics, and a broadcast join attaches the current driver for each vehicle before writing to both GCS and BigQuery in the same foreachBatch:

raw_stream = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", KAFKA_BOOTSTRAP)
    .option("subscribePattern", "fleet.*.events")
    .option("startingOffsets", "latest")
    .load()
)

def process_batch(batch_df, batch_id):
    enriched = batch_df.join(broadcast(driver_mapping_df), "vehicle_id", "left")

    enriched.write.mode("append").partitionBy("event_date", "vehicle_id") \
        .parquet(GCS_RAW_PATH)

    enriched.write.format("bigquery") \
        .option("table", "omniroute_dwh.fact_gps_events") \
        .mode("append").save()

query = raw_stream.writeStream.foreachBatch(process_batch) \
    .option("checkpointLocation", CHECKPOINT_PATH).start()

MERGE for SCD Type 2 — via the BigQuery client, not Spark SQL. The BigQuery Spark connector doesn't push spark.sql("MERGE ...") down to BigQuery, so DML runs through google.cloud.bigquery directly:

from google.cloud import bigquery

client = bigquery.Client()

merge_sql = """
MERGE `omniroute_dwh.dim_asset_history` T
USING `omniroute_dwh.staging_assignment` S
ON T.vin = S.vin AND T.status = 'IN-TRANSIT'
WHEN MATCHED AND S.start_date > T.start_date THEN
  UPDATE SET end_date = S.start_date, status = 'ARCHIVED'
WHEN NOT MATCHED THEN
  INSERT (vin, driver_id, start_date, end_date, daily_rate, status)
  VALUES (S.vin, S.driver_id, S.start_date, NULL, S.daily_rate, 'IN-TRANSIT')
"""

client.query(merge_sql).result()

7. End Result

Today, the pipeline runs a daily cycle:

00:05–00:15 UTC — cron jobs generate the vehicle registry, vehicle-assignment (with 40% reassignment randomness to simulate real driver churn), and maintenance-schedule CSVs, landing them in GCS.
00:20 UTC — an Airflow ingestion DAG pulls the CSVs, applies the SCD Type 2 merge into dim_asset_history, and resolves same-day conflicts by highest daily rate.
Continuously — the Structured Streaming job reads GPS, fuel, and safety events off Kafka, enriches them with the current driver via a broadcast join, and lands them in both GCS (raw Parquet) and BigQuery fact tables (fact_gps_events, fact_fuel_events, fact_safety_events).
04:00 UTC — a reports DAG builds fact_safety_violations, fact_driver_strikes, and the fuel-efficiency audit ahead of the 05:00 UTC management deadline.
1st of the month, 05:00 UTC — the cooldown job resets strikes and restores rates for eligible drivers, explicitly excluding anyone flagged SUSPENDED.

The BigQuery warehouse (omniroute_dwh) now holds a full gold layer — fact_gps_events, fact_fuel_events, fact_safety_events, fact_safety_violations, fact_gps_violations_daily, fact_driver_strikes, dim_driver_penalty_status, dim_asset_history (SCD2), dim_maintenance_schedule, fact_fuel_efficiency_audit, and gold_active_fleet_snapshot — with Superset connected and ready for dashboarding.

Still open: confirming the safety-violations load job succeeds end-to-end after the recent roles/bigquery.readSessionUser grant, finishing a schema alignment on fact_safety_violations (several columns were renamed and added via ALTER TABLE), hardening the ingestion DAG with request_id and execution_timeout on its Dataproc job submissions, and building out the actual Superset dashboards on top of the now-stable gold layer.

Lessons that would save the next person time:

Dataproc worker nodes have no internet access — pre-stage connector jars in GCS and pass them with --jars, not spark.jars.packages.
Stopped Dataproc clusters still bill for attached disk — delete, don't stop.
BigQuery DML from Spark needs the native client library, not spark.sql(...).
Keep SQL files strictly ASCII — a stray em dash in a comment silently broke the GCS → XCom → BigQuery pipeline.

OmniRoute continues to evolve as a hands-on sandbox for GCP data engineering patterns — streaming and batch convergence, SCD2 at scale, and running a full warehouse-to-BI stack on a training budget.

Pages

Sep 10, 2026