The gap between event and insight is money
A temperature sensor spikes on a refrigerated truck carrying vaccines. A fraud pattern emerges across twelve transactions in three seconds. An F1 driver’s brake bias drifts toward a failure threshold mid-corner. In each case, you have seconds to act. Not minutes. Not a dashboard that refreshes tomorrow.
Most businesses still run on batch. Hourly extracts. Overnight ETL. Reports that land in inboxes the next morning. That cadence worked when decisions moved at the speed of meetings. It breaks when decisions move at the speed of the thing you are watching.
You do not need real-time for everything. Monthly churn reports can wait. But fraud scoring cannot. Fleet rerouting cannot. Patient monitor alerts cannot. The question is not whether real-time matters. The question is which parts of your business need it and how to build it without overengineering the rest.
The architecture that runs our F1 telemetry pipeline
We stream live Formula 1 telemetry using a stack that costs less than a single enterprise license and runs on a commodity VPS. The same pattern serves logistics operators, IoT manufacturers, and SaaS platforms today.
Four layers. Each does one thing.
[FastF1 API] -> [Python ingestion] -> [InfluxDB] -> [Grafana]
|
v
[PostgreSQL] -> [Metabase]The ingestion layer pulls data from the source. InfluxDB stores the high-frequency time-series stream for real-time dashboards. PostgreSQL holds the enriched, aggregated records for historical analysis in Metabase. You run both databases because they solve different problems. More on that in a moment.
Ingestion: Python, FastF1, and the art of not losing data
The FastF1 library exposes F1 telemetry through a clean Python API. You call get_session() with a year, round, and session type, then pull lap data, car telemetry, and position streams.
import fastf1
from datetime import datetime
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
fastf1.Cache.enable_cache("cache/")
session = fastf1.get_session(2025, 14, "R")
session.load()
bucket = "f1_telemetry"
client = InfluxDBClient(
url="http://localhost:8086",
token="your-token",
org="cloud-intelligence",
)
write_api = client.write_api(write_options=SYNCHRONOUS)
for _, driver in session.laps.iterlaps():
telemetry = driver.get_telemetry()
for _, row in telemetry.iterrows():
point = (
Point("car_telemetry")
.tag("driver", driver["Driver"])
.tag("lap", str(driver["LapNumber"]))
.field("speed", int(row["Speed"]))
.field("rpm", int(row["RPM"]))
.field("throttle", float(row["Throttle"]))
.field("brake", bool(row["Brake"]))
.field("gear", int(row["nGear"]))
.time(datetime.utcfromtimestamp(row["SessionTime"].total_seconds()))
)
write_api.write(bucket=bucket, record=point)
write_api.close()
client.close()A few things matter in that loop. The .time() field uses the session clock from the car, not wall time, because you want alignment across cars, not server latency. Tags go on low-cardinality dimensions — driver, lap number, compound. Fields hold the actual measurements. This distinction determines query performance later.
Batching writes
Writing one point per HTTP request burns a second per lap. Instead, batch.
from influxdb_client.client.write_api import ASYNCHRONOUS
from itertools import islice
write_api = client.write_api(write_options=ASYNCHRONOUS)
def batched(iterable, size=1000):
iterator = iter(iterable)
return iter(lambda: list(islice(iterator, size)), [])
points = []
for _, driver in session.laps.iterlaps():
telemetry = driver.get_telemetry()
for _, row in telemetry.iterrows():
point = (
Point("car_telemetry")
.tag("driver", driver["Driver"])
.tag("lap", str(driver["LapNumber"]))
.field("speed", int(row["Speed"]))
.field("rpm", int(row["RPM"]))
.field("throttle", float(row["Throttle"]))
.field("brake", bool(row["Brake"]))
.field("gear", int(row["nGear"]))
.time(datetime.utcfromtimestamp(row["SessionTime"].total_seconds()))
)
points.append(point)
for batch in batched(points, 5000):
write_api.write(bucket=bucket, record=batch)ASYNCHRONOUS mode queues writes in a background thread. The InfluxDB client handles the internal batching. The outer batched() iterator keeps memory flat when a session produces millions of points.
Error handling and retry
Networks fail. APIs rate-limit. The ingestion layer must survive both.
import time
from fastf1.core import SessionNotAvailableError
MAX_RETRIES = 5
BASE_DELAY = 2
for attempt in range(MAX_RETRIES):
try:
session = fastf1.get_session(2025, 14, "R")
session.load()
break
except SessionNotAvailableError:
if attempt == MAX_RETRIES - 1:
raise
delay = BASE_DELAY * (2 ** attempt)
print(f"Session not available, retrying in {delay}s")
time.sleep(delay)Exponential backoff with jitter prevents thundering-herd retries when the source goes down. For production, wrap this in a decorator and add a dead-letter queue for points that fail after all retries.
Storage: InfluxDB for the live stream, PostgreSQL for the archive
InfluxDB and PostgreSQL both store time-series data. The difference matters when your query patterns diverge.
InfluxDB answers questions about now: current speed delta, last-lap tire temperatures, throttle traces for the past three laps. It compresses time-series data aggressively, supports downsampling with retention policies, and serves Grafana dashboard queries in milliseconds. Its Flux query language handles windowed aggregations without the verbosity of SQL window functions.
PostgreSQL answers questions about trends: which driver gains positions on medium tires in wet conditions, how sector times correlate with pit-stop windows across a season. It joins telemetry with reference tables — circuits, weather, compounds — in ways that InfluxDB does not. Metabase connects to it for ad-hoc exploration.
Use both. InfluxDB feeds the live dashboard. PostgreSQL holds the enriched record for analysts who need SQL.
InfluxDB schema design
Tags define the lookup keys. Fields hold the numbers. Every choice here affects query speed.
Measurement: car_telemetry
Tags: driver (string), lap (string), compound (string), track_status (string)
Fields: speed (int), rpm (int), throttle (float), brake (bool), gear (int),
drs (bool), brake_temp_f (float), brake_temp_r (float)
Timestamp: session_timeKeep tags low-cardinality. A tag on every millisecond timestamp destroys index performance. Use fields for high-cardinality values. Store time in UTC and let Grafana handle the timezone shift in the panel.
Retention policies
Live telemetry grows fast. A single F1 race produces millions of data points. Define a retention policy that downsamples automatically.
influx bucket update \
--name f1_telemetry \
--retention 30d
influx task create \
--name f1_downsample \
--every 1h \
--org cloud-intelligence \
--file downsample.fluxThe downsample task aggregates raw telemetry into one-second buckets — average speed, min and max RPM, throttle position. The raw data expires after 30 days. The downsampled data stays indefinitely. You keep the high-resolution detail for recent races and the aggregated trend for historical comparison, without the storage cost.
Visualization: Grafana for now, Metabase for later
Grafana connects to InfluxDB and updates dashboards in real time. Engineers watch telemetry streams and call undercuts before the pit wall does. The dashboards auto-refresh at one-second intervals. A single Grafana panel queries the last 30 seconds of data and plots speed, throttle, and brake traces for a selected driver.
-- Grafana variable query for driver selection
import "influxdata/influxdb/schema"
schema.tagValues(
bucket: "f1_telemetry",
tag: "driver",
start: -1h
)Metabase connects to PostgreSQL. Business analysts, not engineers, build their own questions. “Show me average pit-stop duration grouped by circuit for the past three seasons.” No SQL required. That is the split — Grafana for operational dashboards, Metabase for investigative analytics.
We wrote a deeper comparison in Metabase vs Grafana vs Superset. Short version: Grafana excels at time-series monitoring. Metabase excels at ad-hoc business questions. Superset sits between them with more chart types and more complexity. Pick two.
Deployment: Docker Compose, health checks, monitoring the pipeline
Every component runs in Docker. One compose file. One command.
version: "3.8"
services:
influxdb:
image: influxdb:2.7
ports:
- "8086:8086"
volumes:
- influxdb_data:/var/lib/influxdb2
environment:
DOCKER_INFLUXDB_INIT_MODE: setup
DOCKER_INFLUXDB_INIT_USERNAME: admin
DOCKER_INFLUXDB_INIT_PASSWORD: ${INFLUXDB_PASSWORD}
DOCKER_INFLUXDB_INIT_ORG: cloud-intelligence
DOCKER_INFLUXDB_INIT_BUCKET: f1_telemetry
DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: ${INFLUXDB_TOKEN}
healthcheck:
test: ["CMD", "influx", "ping"]
interval: 10s
timeout: 5s
retries: 3
postgres:
image: postgres:16
ports:
- "5433:5432"
volumes:
- postgres_data:/var/lib/postgresql/data
- ./init.sql:/docker-entrypoint-initdb.d/init.sql
environment:
POSTGRES_USER: analyst
POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
POSTGRES_DB: racing
healthcheck:
test: ["CMD-SHELL", "pg_isready -U analyst -d racing"]
interval: 10s
timeout: 5s
retries: 3
grafana:
image: grafana/grafana:11.0
ports:
- "3000:3000"
volumes:
- grafana_data:/var/lib/grafana
- ./grafana/dashboards:/etc/grafana/provisioning/dashboards
- ./grafana/datasources:/etc/grafana/provisioning/datasources
environment:
GF_SECURITY_ADMIN_PASSWORD: ${GRAFANA_PASSWORD}
depends_on:
influxdb:
condition: service_healthy
metabase:
image: metabase/metabase:latest
ports:
- "3001:3000"
environment:
MB_DB_TYPE: postgres
MB_DB_DBNAME: racing
MB_DB_HOST: postgres
MB_DB_PORT: 5432
MB_DB_USER: analyst
MB_DB_PASS: ${POSTGRES_PASSWORD}
depends_on:
postgres:
condition: service_healthy
ingestion:
build: ./ingestion
environment:
INFLUXDB_TOKEN: ${INFLUXDB_TOKEN}
INFLUXDB_URL: http://influxdb:8086
POSTGRES_URL: postgresql://analyst:${POSTGRES_PASSWORD}@postgres:5432/racing
depends_on:
influxdb:
condition: service_healthy
postgres:
condition: service_healthy
restart: on-failure
volumes:
influxdb_data:
postgres_data:
grafana_data:Health checks matter. depends_on with condition: service_healthy means Grafana does not start until InfluxDB responds. The ingestion service restarts on failure. In production, add a process supervisor outside Docker — systemd or a Kubernetes pod restart policy — because Docker’s restart: on-failure does not handle all crash modes.
Monitoring the pipeline itself
You monitor the thing that monitors the thing. InfluxDB’s own metrics bucket tracks write throughput, query latency, and error rates. Create a pipeline health dashboard in Grafana that queries these metrics.
from(bucket: "f1_telemetry")
|> range(start: -5m)
|> filter(fn: (r) => r._measurement == "car_telemetry")
|> aggregateWindow(every: 30s, fn: count)
|> yield(name: "points_per_30s")If points per 30 seconds drops below a threshold, the pipeline stalled. Set an alert. Our threshold is zero for more than 60 seconds.
import requests
def check_pipeline_health():
query = """
from(bucket: "f1_telemetry")
|> range(start: -1m)
|> filter(fn: (r) => r._measurement == "car_telemetry")
|> count()
"""
response = requests.post(
"http://influxdb:8086/api/v2/query?org=cloud-intelligence",
headers={"Authorization": f"Token {INFLUXDB_TOKEN}"},
data=query,
)
if response.text.strip():
return True
return FalseA cron job runs this check every minute. A Discord webhook fires on failure. When the pipeline goes quiet during a race, someone knows.
Production considerations
Live telemetry is a gentle data rate. F1 generates roughly 4 MB per car per lap. At 20 cars and 60 laps, the raw volume fits in memory. Production problems start at higher rates — IoT fleets with 10,000 sensors, financial tick data, ad-bid streams. Here is what breaks and how to fix it.
Backpressure
When the source produces data faster than the sink can write, backpressure builds. Memory fills. The process dies. Two levers: buffer and throttle.
Use a bounded queue between the ingest loop and the write path. When the queue hits capacity, block the producer or drop non-critical measurements.
from queue import Queue, Full
BATCH_SIZE = 5000
write_queue = Queue(maxsize=100_000)
def produce():
for point in telemetry_stream():
try:
write_queue.put(point, timeout=0.1)
except Full:
metrics.dropped_points.inc()
continue
def consume():
batch = []
while True:
try:
batch.append(write_queue.get(timeout=1))
except Empty:
if batch:
write_api.write(bucket=bucket, record=batch)
batch = []
continue
if len(batch) >= BATCH_SIZE:
write_api.write(bucket=bucket, record=batch)
batch = []Two threads. One produces. One consumes. The queue absorbs bursts. The maxsize prevents runaway memory. The timeout on put() means you drop rather than deadlock.
Downsampling
Raw data resolution declines in value over time. Nobody queries millisecond-precision tire temperatures from three seasons ago. Downsampling reduces storage cost without reducing utility.
Define continuous queries in InfluxDB that aggregate to coarser intervals.
option task = {
name: "f1_downsample_hourly",
every: 1h,
}
from(bucket: "f1_telemetry")
|> range(start: -1h)
|> filter(fn: (r) => r._measurement == "car_telemetry")
|> aggregateWindow(every: 1s, fn: mean)
|> to(bucket: "f1_telemetry_downsampled")Run the task hourly. It takes 1-millisecond raw data and collapses it to 1-second means. The downsampled bucket costs a fraction of the raw bucket. Grafana queries for historical dashboards point to the downsampled bucket. Live dashboards query the raw bucket for the current session.
Alerting on pipeline failures
Three alert conditions cover the common failures:
- No data arriving. Points-per-minute drops to zero for two consecutive checks. The source went down or the ingestion process died.
- Write failures. InfluxDB rejections or timeouts exceed 5% over a rolling five-minute window. The database is overloaded or out of disk.
- Lag exceeds threshold. The timestamp of the latest ingested point falls more than 10 seconds behind wall clock. The pipeline fell behind.
Each condition fires to a Slack or Discord channel. The first alert wakes someone. The second two alert a different severity because they indicate degradation, not outage.
From F1 to your domain
The pattern transfers directly. Swap FastF1 for your data source. MQTT topics for IoT sensors. WebSocket streams for financial data. CDC logs from PostgreSQL for SaaS event streams. The architecture does not change.
We built the same pipeline pattern for SaaS churn detection in SaaS Retention Pulse. The ingestion layer pulls synthetic SaaS event data. InfluxDB tracks DAU, WAU, and session quality. Grafana dashboards highlight accounts with declining engagement before they churn. Same four layers. Different domain.
Fleet operators track vehicle telemetry the same way. Manufacturer IoT platforms monitor equipment health. Logistics companies watch shipment temperature and location. The stack — Python ingestion, InfluxDB for the stream, PostgreSQL for the archive, Grafana and Metabase for visualisation — handles all of them.
What you need to start
A single machine with 4 GB RAM runs this stack for thousands of data points per second. The code is small. The complexity lives in the edge cases — backpressure, retry, alerting — and you now have the patterns for all three.
Start with the Live F1 Telemetry demo to see the stack running. Then read Metabase vs Grafana vs Superset to pick the right visualisation tools for your team. When you are ready to operationalize, 10 SaaS Metrics Every Dashboard Needs shows what to measure once the pipeline runs.
If you have a data stream and need a dashboard that refreshes in seconds, not hours, get in touch. We stand up the pipeline and iterate until it fits your domain.