How Logistics Companies Use Data Engineering

# Simplified tracking pipeline (PySpark Structured Streaming)

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json

spark = SparkSession.builder.getOrCreate()

# Read shipment events from Kafka
raw = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "shipment_events")
    .load()
)

# Parse + validate
events = (
    raw.select(
        from_json(col("value").cast("string"),
                  schema).alias("e")
    )
    .select("e.*")
    .filter(col("shipment_id").isNotNull())
)

# Write to warehouse (Delta / Iceberg / BigQuery)
(
    events.writeStream
    .format("delta")
    .option("checkpointLocation", "/checkpoints/shipments")
    .option("mergeSchema", "true")
    .start("/warehouse/shipments")
)

# Every scanner ping becomes a queryable record.

Shipment event — canonical schema:

{
  "event_id":      "uuid",
  "shipment_id":   "SH-2026-000123",
  "event_type":    "picked_up | in_transit | out_for_delivery | delivered",
  "timestamp":     "2026-08-28T10:45:00Z",
  "location": {
    "lat": 28.6139,
    "lng": 77.2090,
    "hub_id": "DEL-01"
  },
  "partner_id":     "P-482",
  "device_id":      "SCN-1189",
  "metadata":       { "temp_c": 4, "door": "closed" }
}

Design notes:
- event_id → dedupe
- shipment_id → partition key
- timestamp → event-time analytics
- metadata → flexible extensions

# Compute ETA features — join historical + live

import pandas as pd

# Historical delivery times (from warehouse)
hist = pd.read_parquet("s3://warehouse/delivery_stats/")

# Live GPS (from a streaming sink)
live = pd.read_parquet("s3://streaming/live_gps/latest/")

# Join on shipment_id
features = live.merge(
    hist, on=["route_id", "hour_of_day"],
    how="left"
)

features["predicted_eta"] = (
    features["historical_p90"] * features["traffic_factor"]
)

# Write to feature store for the ETA model
features[[
    "shipment_id", "predicted_eta",
    "route_id", "traffic_factor"
]].to_parquet("s3://feature-store/eta/", mode="overwrite")

Streaming joins in Flink (simplified):

// Enrich live GPS with warehouse reference data

DataStream<GPS> gps = env.fromSource(kafkaSource, ...);
DataStream<Route> routes = env.fromSource(
    kafkaReferenceSource, ...);

DataStream<ETA> etas = gps
  .keyBy(GPS::getRouteId)
  .connect(routes.keyBy(Route::getId))
  .process(new EtaCalculator())
  .name("eta-enrichment");

etas.sinkTo(warehouseSink).name("write-eta");

// Real systems use:
// - Flink or Spark Structured Streaming
// - Redis for low-latency lookups
// - Feature stores (Feast, Tecton)
// - ML models for final ETA

-- models/fleet_daily.sql
-- Daily per-truck fuel efficiency + utilisation

with trips as (
    select
        truck_id,
        date(started_at) as trip_date,
        sum(distance_km) as total_km,
        sum(fuel_liters) as total_liters,
        sum(idle_minutes) as idle_minutes
    from {{ ref('trips') }}
    group by 1, 2
)

select
    truck_id,
    trip_date,
    total_km,
    round(total_km / nullif(total_liters, 0), 2)
        as km_per_liter,
    round(idle_minutes * 100.0
          / nullif(total_km, 0), 2)
        as idle_pct
from trips
order by trip_date desc, truck_id;

# Predictive maintenance — simple threshold example

def maintenance_alert(truck_telemetry):
    score = 0

    # High engine temperature
    if truck_telemetry["avg_engine_temp_c"] > 105:
        score += 3

    # Excessive idle
    if truck_telemetry["idle_hours"] > 20:
        score += 1

    # Distance since last service
    if truck_telemetry["km_since_service"] > 20000:
        score += 2

    # Recent harsh braking
    if truck_telemetry["harsh_brakes"] > 15:
        score += 2

    return score >= 4   # trigger alert

# Production systems use ML models
# trained on historical failures + telemetry.

Signals data engineers expose:

Delivery fraud:
□ "Delivered" with no GPS proof
□ 3+ deliveries at same timestamp
□ Delivered-to location ≠ customer address
□ Repeated "customer not available" claims

Fuel theft:
□ Fuel level drops without distance
□ Refuels at odd hours / odd locations
□ MPG suddenly drops (per driver / truck)

Route fraud:
□ Unplanned detours
□ Excessive idle time
□ Repeated stops at non-customer locations

These signals power fraud dashboards
and investigation workflows.

Typical pipeline design:

Sources:
- GPS pings (Kafka)
- Delivery scan events
- Fuel sensor telemetry
- Driver app events
- Historical fraud cases

Processing:
- Streaming: real-time anomaly flags
- Batch: daily deep analysis (Spark + dbt)
- Feature store: joins for ML models

Destinations:
- Fraud investigation dashboard
- SIU (Special Investigation Unit) queue
- Alerting (Slack / email / PagerDuty)
- Feature store for ML models

Key design concern:
"Explainable" alerts are more useful than
"black-box" fraud scores in ops workflows.

Daily tracker:

Date: _________

□ Reviewed fraud alerts (______ count)
□ Investigated top 5 cases
□ Updated false-positive notes
□ Reviewed new pipeline warnings
□ Checked data quality dashboard

Notes:
- 
- 
-