Processing 1 Million Events Per Day Without Breaking the Bank (Or Your Database)

The real-world math behind event batching, and why it changed everything for me

---

The Problem

VehicleMetrics receives sensor data from vehicles in real-time.

Each vehicle sends: 10 GPS updates/second + 100 sensor readings/second = 110 messages/second per vehicle

With 50 vehicles: 5,500 messages/second = 473 million messages per day

I needed to:

1. Ingest fast (< 50ms response time)
2. Store reliably (no data loss)
3. Not go bankrupt (costs matter at scale)
4. Not break the database (PostgreSQL can only write so fast)

This is where most people fail. They think: "Just insert each event into the database."

Wrong.

---

The Naive Approach (That Destroys Databases)

# ❌ NAIVE: Direct insert per event
@app.post("/api/vehicles/{vehicle_id}/sensor-data")
async def ingest(vehicle_id: str, data: SensorData):
    await db.execute(
        "INSERT INTO sensor_data VALUES (...)",
        [vehicle_id, data.timestamp, data.speed, ...]
    )
    return {"status": "created"}

The math:

• 5,500 events/second × 60 = 330,000 inserts/minute
• Each insert: ~1ms database overhead
• PostgreSQL write throughput: ~10,000 inserts/second max
• Result: Queue builds. Latency explodes. System crashes.

---

What Actually Works: Batch Processing

Instead of inserting individually, batch events and write them together.

# ✅ REAL: Batch events before writing
from asyncio import gather
import asyncio

class EventBatcher:
    def __init__(self, batch_size=1000, flush_interval=5):
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self.queue = []
        self.lock = asyncio.Lock()

    async def add(self, event: dict):
        async with self.lock:
            self.queue.append(event)
            if len(self.queue) >= self.batch_size:
                await self._flush()

    async def _flush(self):
        if not self.queue:
            return

        # Batch write to database
        events = self.queue[:]
        self.queue = []

        # Single multi-row insert
        await db.execute(
            """INSERT INTO sensor_data (vehicle_id, timestamp, speed, sensors)
               VALUES """ + ", ".join(["(%s, %s, %s, %s)"] * len(events)),
            [val for event in events for val in [
                event['vehicle_id'],
                event['timestamp'],
                event['speed'],
                event['sensors']
            ]]
        )

        logger.info(f"Flushed {len(events)} events")

batcher = EventBatcher(batch_size=1000, flush_interval=5)

@app.post("/api/vehicles/{vehicle_id}/sensor-data")
async def ingest(vehicle_id: str, data: SensorData):
    event = {
        'vehicle_id': vehicle_id,
        'timestamp': data.timestamp,
        'speed': data.speed,
        'sensors': data.sensors
    }

    # Add to batch (doesn't block)
    await batcher.add(event)

    # Return immediately (< 5ms)
    return {"status": "queued"}

# Flush periodically
@app.on_event("startup")
async def flush_periodically():
    while True:
        await asyncio.sleep(5)
        await batcher._flush()

The math changes:

• Instead of 5,500 individual inserts/second
• We do 6 batch inserts/second (5,500 ÷ 1,000 batch size)
• Database load: 5,500 events written, but only 6 inserts executed
• Result: Database stays relaxed. Latency stays low. We scale.

---

The Real Numbers

Before Batching:

Throughput: 50 rps (bottlenecked by database)
Latency: 2,500ms (queued inserts waiting)
Database CPU: 95% (constant write pressure)
Cost: $150/month RDS (need bigger instance)

After Batching:

Throughput: 1,000+ rps (limited by API, not database)
Latency: 18ms (batch waits 5 seconds max)
Database CPU: 15% (batches are efficient)
Cost: $100/month RDS (smaller instance works fine)

That's 20x throughput improvement + 33% cost reduction.

---

The Batching Trade-offs

Trade-off 1: Latency vs Throughput

# Option C: Real-time + Batch split
# - Real-time data: Redis stream (for dashboard)
# - Persistent data: Batch to database (for analytics)

@app.post("/api/vehicles/{vehicle_id}/sensor-data")
async def ingest(vehicle_id: str, data: SensorData):
    # Real-time: Update Redis immediately
    await redis.hset(
        f"vehicle:{vehicle_id}:latest",
        mapping={
            'speed': data.speed,
            'timestamp': data.timestamp,
        }
    )

    # Persistent: Add to batch (writes later)
    await batcher.add({
        'vehicle_id': vehicle_id,
        'timestamp': data.timestamp,
        'speed': data.speed,
        'sensors': data.sensors
    })

    return {"status": "accepted"}

Result: Dashboard gets data instantly (Redis), analytics get data within 1-5 seconds (database batch).

---

Advanced: Batch Size Selection

Formula:

batch_size = (target_latency × events_per_second) / batches_per_second

Example:
- Target latency: 5 seconds (acceptable delay)
- Events per second: 5,500
- Desired batches per second: 10 (don't stress DB)

batch_size = (5 × 5,500) / 10 = 2,750

---

The Real Cost Analysis

Without Batching:

RDS Instance: db.t3.large ($300/month)
Kinesis: 10 shards ($300/month)
Lambda: Heavy ($50/month)
EC2 API servers: 10 instances ($500/month)

Total: ~$1,150/month

With Batching:

RDS Instance: db.t3.small ($100/month)
Kinesis: 1 shard ($30/month)
Lambda: Minimal ($5/month)
EC2 API servers: 2 instances ($100/month)

Total: ~$235/month

Batching saves 80% on infrastructure costs.

---

TL;DR

• Don't insert events one-at-a-time (database will hate you)
• Batch 500-2,000 events (sweet spot)
• Flush every 1-5 seconds (or when batch is full)
• Split real-time (Redis) from persistent (batch) (best of both)
• Result: 20x throughput, 80% cost savings

The real lesson: One tiny optimization (batching) can change your entire infrastructure.

---

GitHub: beltagyy/vehicle-metrics
Author: Mohamed ElBeltagy (@beltagyy)
Topic: Real-Time Data | Performance | Scale

Comments

[leob]

Yeah good patterns ... how do you measure throughput/TPS/latency and the other metrics?

[Le Beltagy]

Great question!

I measure throughput with a Prometheus counter on batcher.add() calls and graph the 1-minute rate in Grafana.
For latency, I track three tiers: FastAPI middleware gives me ~2-5ms ingest time to the queue, the batch itself waits 0-5s depending on flush_interval, and end-to-end latency to queryable Postgres rows sits around 5-7 seconds.

On the database side, I watch pg_stat_database.tup_inserted rate, pg_stat_activity for IO wait events, and I run EXPLAIN (ANALYZE, BUFFERS) on the actual batched INSERT every few hours to catch plan regressions before they hurt.

The one gap I know I have: my dashboard shows average batch latency but not P99, so spikes during traffic bursts get hidden — histogram buckets are next on my list. Happy to share the Grafana JSON. Also was thinking about creating a collection python script to automate all.