Case study 05 · Logistics & Fulfilment

Real-Time Analytics Platform

Streaming order telemetry from Event Hubs to live operational dashboards in under a minute.

Data Engineer — streaming pipeline, serving layer, alerting 3 months

Latency: T+1 day → < 60s
Events / sec: ~1,500 peak
SLA breaches: −31%

Executive summary

Built a streaming analytics platform for a fulfilment operation: order and warehouse events flow through Azure Event Hubs into Databricks Structured Streaming, are aggregated into one-minute windows in Delta Lake, and power a live Power BI operations dashboard with threshold alerts.

Business problem

Operations managers saw yesterday's order backlog in a morning report. By the time a fulfilment bottleneck was visible, delivery SLAs had already been missed.

No intraday visibility of order throughput or backlog
Bottlenecks discovered after SLAs were breached
Batch reports unable to show per-region trends within the day
No historical event store for root-cause analysis

Architecture design

A hot path for real-time aggregates and a cold path that archives every raw event, both fed by the same Event Hubs stream.

L1
Producers
Order service and warehouse scanners publish JSON events to Event Hubs.
L2
Event broker
Azure Event Hubs (Kafka endpoint), 8 partitions, Capture to ADLS for the cold path.
L3
Stream processing
Structured Streaming parses, deduplicates within a watermark and aggregates 1-minute windows.
L4
Hot store
Delta tables for per-minute metrics, served through a Databricks SQL Warehouse.
L5
Consumption
Power BI dashboard on a 30-second page refresh, plus threshold alerts to Teams.

Source systems

System	Technology	Volume	Cadence
Order service	Event Hubs (JSON)	~1,500 events/sec peak	Streaming
Warehouse scanners	Event Hubs (JSON)	~600 events/sec	Streaming
Carrier status	Webhook → Event Hubs	~90K updates/day	Streaming
Region & SLA reference	Delta table	Small	Daily

Data flow

01Publish
Services publish events with order_id, region and timestamps.
02Ingest
Spark reads Event Hubs through its Kafka-compatible endpoint.
03Parse & dedupe
JSON parsed against a fixed schema; duplicates removed within a 2-minute watermark.
04Aggregate
Tumbling one-minute windows per region: orders, revenue, fulfilment time.
05Serve
Append to a Delta metrics table; a view adds rolling baselines.
06Alert
A scheduled query flags regions whose throughput drops below baseline.

Data model

A narrow time-series model optimised for 'last two hours' queries, plus the raw archive for replay.

Facts

gold.order_metrics_1m
Grain: One row per region per minute
silver.order_events
Grain: One row per unique event
bronze.order_events_archive
Grain: Raw Event Hubs Capture (Avro)

Dimensions

dim_region
Region, warehouse and SLA thresholds
dim_carrier
Carrier and service level

ETL process

Exactly-once sinks

Checkpointed Delta writes give end-to-end exactly-once semantics for the aggregates.

Late data handling

A 2-minute watermark balances completeness and latency; later events land in the cold path.

Backfill

Captured Avro files can be replayed through the same transformation code in batch mode.

Operational guardrails

Streaming query listener metrics (input rate, batch duration) are logged and alerted on.

Tech stack

Streaming: Azure Event HubsKafka protocolEvent Hubs Capture
Processing: DatabricksStructured StreamingPySpark
Storage: Delta LakeADLS Gen2
Serving: Databricks SQL WarehousePower BITeams alerts
Security: Databricks secret scopesKey VaultPrivate endpoints

SQL example

Live KPI view with a rolling baseline (Databricks SQL)

sql

CREATE OR REPLACE VIEW lakehouse.gold.v_live_order_kpis AS
SELECT  m.window_start,
        m.region,
        m.orders,
        m.revenue,
        m.avg_fulfilment_sec,
        AVG(m.orders) OVER (
            PARTITION BY m.region
            ORDER BY m.window_start
            ROWS BETWEEN 15 PRECEDING AND 1 PRECEDING
        )                                               AS orders_baseline_15m,
        r.sla_fulfilment_sec,
        m.avg_fulfilment_sec > r.sla_fulfilment_sec     AS sla_at_risk
FROM    lakehouse.gold.order_metrics_1m AS m
JOIN    lakehouse.gold.dim_region       AS r
        ON r.region = m.region
WHERE   m.window_start >= current_timestamp() - INTERVAL 2 HOURS;

Python example

Event Hubs → 1-minute windowed metrics (Structured Streaming)

python

from pyspark.sql import functions as F, types as T

NAMESPACE = "evhns-orders-prod"
TOPIC = "orders"
connection = dbutils.secrets.get(scope="streaming", key="eventhub-orders-listen")

kafka_options = {
    "kafka.bootstrap.servers": f"{NAMESPACE}.servicebus.windows.net:9093",
    "subscribe": TOPIC,
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.mechanism": "PLAIN",
    "kafka.sasl.jaas.config": (
        "kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required "
        f'username="$ConnectionString" password="{connection}";'
    ),
    "startingOffsets": "latest",
    "failOnDataLoss": "false",
}

schema = T.StructType([
    T.StructField("order_id", T.StringType()),
    T.StructField("region", T.StringType()),
    T.StructField("amount", T.DecimalType(18, 2)),
    T.StructField("created_at", T.TimestampType()),
    T.StructField("fulfilled_at", T.TimestampType()),
])

orders = (spark.readStream.format("kafka").options(**kafka_options).load()
    .select(F.from_json(F.col("value").cast("string"), schema).alias("o"))
    .select("o.*")
    .withWatermark("created_at", "2 minutes")
    .dropDuplicatesWithinWatermark(["order_id"]))

fulfilment_sec = F.col("fulfilled_at").cast("long") - F.col("created_at").cast("long")

metrics = (orders
    .groupBy(F.window("created_at", "1 minute"), "region")
    .agg(F.count("*").alias("orders"),
         F.sum("amount").alias("revenue"),
         F.avg(fulfilment_sec).alias("avg_fulfilment_sec"))
    .select(F.col("window.start").alias("window_start"),
            "region", "orders", "revenue", "avg_fulfilment_sec"))

(metrics.writeStream
    .outputMode("append")
    .option("checkpointLocation", "/Volumes/lakehouse/ops/checkpoints/order_metrics_1m")
    .trigger(processingTime="10 seconds")
    .toTable("lakehouse.gold.order_metrics_1m"))

Challenges

Challenge

Choosing a watermark: too short dropped late events, too long delayed every window.

Solution

Measured event lateness for two weeks; 2 minutes covered 99.7% of events, with the cold path catching the rest.

Challenge

Producers retried on timeouts and sent duplicate events.

Solution

dropDuplicatesWithinWatermark on order_id keeps state bounded while removing retries.

Challenge

Dashboard queries competed with streaming writes for resources.

Solution

Separated compute: streaming on a dedicated job cluster, reads on a serverless SQL Warehouse.

Performance improvements

Metric	Before	After	Change
Data latency to dashboard	T+1 day	< 60 s	Real-time
Micro-batch duration (P95)	22 s	7 s	−68%
Dashboard query (P95)	9 s	1.4 s	−84%
Streaming state size	Unbounded	Bounded by watermark	Stable

Business impact

Operations react to bottlenecks within minutes instead of the next morning
Delivery SLA breaches reduced by 31% in the first quarter after launch
Every raw event archived and replayable for root-cause analysis
Foundation reused for carrier performance and warehouse labour planning

Lessons learned

Measure lateness in your actual data before choosing a watermark.
Keep a cold path — real-time systems still need replay and backfill.
Separate streaming and serving compute so neither degrades the other.

Success metrics

End-to-end latency: < 60 sEvent to dashboard
Stream uptime: 99.9%Trailing 90 days
SLA breaches: −31%Quarter after launch
Events retained: 100%Via Event Hubs Capture

Next case studyEnterprise Data Warehouse