Designing an IoT Data Pipeline: Device Ingestion, Time-Series Storage, and Real-Time Fleet Dashboards at Scale

Most backend engineers have never built a system where the clients are robots, sensors, or vehicles operating in the physical world. The mental model for web APIs does not transfer cleanly. Your “clients” may have intermittent connectivity, constrained hardware, strict battery budgets, and clocks that drift. They send telemetry whether or not anyone is listening, and they send it continuously at rates your typical REST endpoint was never designed for.

A fleet of 10,000 devices emitting readings every second generates 10,000 writes per second as a baseline. At 10 readings per second per device, you are at 100,000 writes per second. A conventional web application backend will fall over within minutes under this load, and the failure mode is silent: you lose device data without knowing it.

This article covers the architecture decisions that matter when building an IoT data pipeline: protocol selection for ingestion, time-series database selection for storage, stream processing for anomaly detection, and delivering real-time fleet dashboards that stay current under load.

The Problem With Treating Devices Like API Clients

The naive approach is to give each device an HTTPS endpoint and tell it to POST telemetry. This works at prototype scale. It breaks in production for several reasons.

First, HTTP is request-response. Each telemetry payload initiates a new connection or requires a kept-alive connection that consumes server resources proportional to device count. At 10,000 devices you have 10,000 persistent connections to manage. At 100,000 devices, this becomes a serious infrastructure problem.

Second, HTTP headers add overhead that matters on constrained hardware. A 20-byte sensor reading wrapped in an HTTP/1.1 request body carries 400-500 bytes of headers, retry logic, and TLS negotiation cost. On a microcontroller with a cellular modem, each of those bytes costs battery and money.

Third, HTTP does not model the publish-subscribe pattern that device fleets actually need. A fleet dashboard needs to subscribe to all device telemetry. A specific vehicle’s operations team needs to subscribe to that vehicle’s data. A rules engine needs to subscribe to temperature readings above a threshold. HTTP has no native way to express this. You end up bolting on polling, server-sent events, or WebSockets as a second transport layer.

The solution is to choose the right protocol for each leg of the data path from the start.

Ingestion Protocol Selection

MQTT: The Right Default for Constrained Devices

MQTT is a publish-subscribe messaging protocol designed for unreliable networks and constrained hardware. Devices connect to a broker and publish messages to topics. Other clients subscribe to topics and receive messages. The protocol overhead is minimal: a fixed header of 2 bytes, variable-length topic and payload.

The three Quality of Service levels are the key design decision:

• QoS 0 (at-most-once): Fire and forget. The broker delivers or it does not. Zero protocol overhead beyond the publish. Appropriate for high-frequency telemetry where individual reading loss is acceptable (temperature every 100ms, GPS every second).
• QoS 1 (at-least-once): The broker acknowledges receipt. The device retries until it receives an ACK. Appropriate for readings where you cannot tolerate loss but can tolerate duplicates (alert events, status changes).
• QoS 2 (exactly-once): Four-message handshake. Guarantees delivery exactly once. Use only when deduplication at the storage layer is expensive and payload loss is unacceptable (billing events, safety-critical state changes).

MQTT brokers (Mosquitto, EMQX, HiveMQ, AWS IoT Core) handle fan-out natively. A message published to fleet/vehicle-42/telemetry is delivered to every subscriber on that topic simultaneously. The broker handles the distribution, not your application.

Here is a TypeScript ingestion gateway that bridges MQTT messages to your storage layer:

import mqtt, { MqttClient } from "mqtt";

interface TelemetryPayload {
  deviceId: string;
  timestamp: number;
  readings: Record<string, number>;
}

interface NormalizedReading {
  deviceId: string;
  metric: string;
  value: number;
  timestamp: Date;
}

async function normalizeTelemetry(
  topic: string,
  raw: Buffer
): Promise<NormalizedReading[]> {
  const payload = JSON.parse(raw.toString()) as TelemetryPayload;

  // topic format: fleet/<fleetId>/devices/<deviceId>/telemetry
  const parts = topic.split("/");
  const deviceId = parts[3] ?? payload.deviceId;

  const ts = new Date(payload.timestamp);

  return Object.entries(payload.readings).map(([metric, value]) => ({
    deviceId,
    metric,
    value,
    timestamp: ts,
  }));
}

export function createIngestionGateway(
  brokerUrl: string,
  onBatch: (readings: NormalizedReading[]) => Promise<void>
): MqttClient {
  const client = mqtt.connect(brokerUrl, {
    clean: false,           // persist subscriptions across reconnects
    clientId: "ingestion-gateway-01",
    keepalive: 30,
    reconnectPeriod: 1000,
  });

  const buffer: NormalizedReading[] = [];
  const FLUSH_INTERVAL_MS = 500;
  const FLUSH_SIZE = 1000;

  async function flush() {
    if (buffer.length === 0) return;
    const batch = buffer.splice(0, buffer.length);
    await onBatch(batch);
  }

  setInterval(flush, FLUSH_INTERVAL_MS);

  client.on("connect", () => {
    // subscribe to all device telemetry across all fleets
    client.subscribe("fleet/+/devices/+/telemetry", { qos: 1 });
    client.subscribe("fleet/+/devices/+/events", { qos: 1 });
  });

  client.on("message", async (topic, payload) => {
    try {
      const readings = await normalizeTelemetry(topic, payload);
      buffer.push(...readings);

      if (buffer.length >= FLUSH_SIZE) {
        await flush();
      }
    } catch (err) {
      console.error({ topic, err }, "Failed to parse telemetry message");
    }
  });

  return client;
}

The batching here is intentional. Time-series databases are optimized for bulk inserts. Writing one row at a time will saturate your connection pool before you saturate the database’s write capacity. Flush on a timer and on buffer size threshold.

HTTP: Right for Event-Driven and High-Trust Clients

HTTP is appropriate when devices have reliable connectivity, are not battery-constrained, and are sending discrete events rather than continuous telemetry. A warehouse robot completing a task, a charging station reporting a session end, or a device sending a one-time configuration acknowledgment: these are good HTTP use cases.

REST endpoints also simplify authentication. You can issue per-device API keys or JWT tokens and validate them with standard middleware. MQTT brokers support authentication, but the patterns vary by broker and require more careful setup.

For cloud-to-cloud integration (a third-party hardware partner pushing device data to your platform), HTTPS webhooks are the right interface. You get standard HTTP security, retries, and the sender can use their existing HTTP client.

WebSocket: Right for Low-Latency Bidirectional Control

WebSocket connections are appropriate when you need both telemetry ingestion and real-time command delivery over a single connection. Teleoperation interfaces, remote diagnostics, and devices that need to receive configuration updates in response to their own telemetry all benefit from a persistent bidirectional channel.

The tradeoff is connection management complexity. A WebSocket server must maintain state per connection, handle reconnections gracefully, and route commands to the correct connection for a specific device. At fleet scale, this typically requires a connection registry (Redis is a common choice) so any server node can route to any device connection.

Ingestion Protocol Tradeoffs

Protocol	Overhead	Persistent conn	Pub/Sub native	Best for
MQTT	2-byte header	Yes (broker-managed)	Yes	Constrained devices, high-frequency telemetry
HTTP/REST	High (headers + TLS)	Optional (keep-alive)	No	Event-driven payloads, cloud-to-cloud
WebSocket	Low after handshake	Yes (server-managed)	No	Bidirectional control, teleoperation
gRPC streaming	Low (binary framing)	Yes (HTTP/2)	No	High-throughput, strongly-typed schemas

gRPC streaming deserves a mention for internal gateways where both sides are controlled services. The binary framing and generated client/server code from Protobuf schemas make it a strong choice when you need strict schema enforcement and want to avoid the overhead of JSON parsing at high ingestion rates.

Time-Series Database Selection

General-purpose relational databases handle time-series data poorly at scale. A PostgreSQL table of sensor readings with a timestamp column and a B-tree index will perform acceptably up to a few hundred million rows. Past that, query performance degrades, vacuum overhead grows, and storage efficiency becomes a concern because each row carries full tuple overhead.

Time-series databases address this by organizing data around time as a first-class concept: columnar storage for efficient scans, automatic partitioning by time (chunks or shards), compression optimized for the delta-encoding patterns that sensor data exhibits, and purpose-built query functions for time-based aggregation.

The three candidates most teams evaluate are TimescaleDB, InfluxDB, and QuestDB.

TimescaleDB

TimescaleDB is a PostgreSQL extension. Your data model is a normal PostgreSQL table called a hypertable. The extension automatically partitions it into chunks by time. You use standard SQL, standard PostgreSQL tooling, and standard PostgreSQL indexes. It integrates with your existing Postgres infrastructure.

The PostgreSQL compatibility is the main argument for TimescaleDB. If your team already operates Postgres, there is no new operations runbook to write. pgAdmin, your existing monitoring, your existing backup strategy: all of it applies. Continuous aggregates let you pre-compute rollups (1-minute averages, hourly min/max) as materialized views that update incrementally.

The tradeoff is write throughput. TimescaleDB inherits PostgreSQL’s MVCC-based write path. For very high write rates (hundreds of thousands of rows per second), you will need to batch aggressively and may need to scale out using Timescale’s distributed hypertables, which add operational complexity.

InfluxDB

InfluxDB v2+ uses a purpose-built storage engine (TSM, Time-Structured Merge Tree) optimized for the write-heavy append patterns of telemetry data. The data model is tag/field-based: tags are indexed string metadata (deviceId, location, sensor type), fields are the actual numeric measurements. Queries use Flux, InfluxDB’s functional query language.

InfluxDB is a strong choice when write throughput is the primary constraint and your team is willing to adopt the Flux query language. The TSM engine handles high-cardinality tag sets better than earlier versions, though cardinality at extreme scale (millions of unique tag value combinations) still requires careful schema design.

The operational model changed significantly between v1 and v2. If you are evaluating InfluxDB, plan for v2 (or v3 if your evaluation timeline is mid-2026). InfluxDB Cloud Serverless (v3, built on Apache Arrow/DataFusion) is architecturally different from the self-hosted v2 engine and worth evaluating separately if you want managed hosting.

QuestDB

QuestDB is built specifically for time-series performance. It uses a columnar storage format with memory-mapped files and SIMD-accelerated query execution. The query language is SQL with time-series extensions. Write throughput benchmarks consistently show QuestDB outperforming both TimescaleDB and InfluxDB on raw insert rates.

The tradeoff is maturity. QuestDB is a younger project with a smaller ecosystem, fewer integrations, and a smaller community than either competitor. The tooling around backups, replication, and high availability is less mature. For a new project with performance as the primary constraint and a team willing to operate a less-common database, QuestDB is worth serious evaluation. For a team that needs enterprise support and a large community, TimescaleDB or InfluxDB are safer.

Time-Series Database Tradeoffs

Database	Write throughput	Query language	PostgreSQL compat	Maturity	Best for
TimescaleDB	High (batched)	Full SQL	Yes	High	Teams on Postgres, need SQL, moderate write load
InfluxDB v2	Very high	Flux	No	High	Write-heavy telemetry, tag-based data model
QuestDB	Highest	SQL + extensions	Partial	Medium	Maximum throughput, greenfield projects
ClickHouse	Very high	SQL	No	High	Analytics-heavy workloads, large query fans

ClickHouse is worth noting as an alternative when the primary workload is analytical queries over large time ranges rather than writes. If your fleet dashboard runs “median battery voltage across all vehicles over the past 30 days” more often than it runs “latest reading from vehicle 42,” ClickHouse’s columnar scan performance is competitive.

Writing Device Telemetry to TimescaleDB

Here is the storage layer for normalized readings using TimescaleDB via pg and @timescale/toolkit:

import { Pool } from "pg";

interface NormalizedReading {
  deviceId: string;
  metric: string;
  value: number;
  timestamp: Date;
}

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

// Schema (run once during setup):
// CREATE TABLE device_readings (
//   time        TIMESTAMPTZ NOT NULL,
//   device_id   TEXT        NOT NULL,
//   metric      TEXT        NOT NULL,
//   value       DOUBLE PRECISION NOT NULL
// );
// SELECT create_hypertable('device_readings', 'time');
// CREATE INDEX ON device_readings (device_id, time DESC);

export async function writeBatch(readings: NormalizedReading[]): Promise<void> {
  if (readings.length === 0) return;

  // Build a multi-row INSERT for efficiency.
  // TimescaleDB processes bulk inserts significantly faster than individual rows.
  const values: unknown[] = [];
  const placeholders: string[] = [];

  readings.forEach((r, i) => {
    const base = i * 4;
    placeholders.push(
      `($${base + 1}, $${base + 2}, $${base + 3}, $${base + 4})`
    );
    values.push(r.timestamp, r.deviceId, r.metric, r.value);
  });

  const query = `
    INSERT INTO device_readings (time, device_id, metric, value)
    VALUES ${placeholders.join(", ")}
    ON CONFLICT DO NOTHING
  `;

  const client = await pool.connect();
  try {
    await client.query(query, values);
  } finally {
    client.release();
  }
}

The ON CONFLICT DO NOTHING handles duplicate delivery from QoS 1 MQTT. For a fully idempotent insert path, add a unique index on (time, device_id, metric), accepting the write amplification cost.

Stream Processing for Anomaly Detection

Raw telemetry stored in a time-series database is useful for historical analysis and dashboards. Anomaly detection needs to operate on the live data stream, before it reaches storage, to keep alert latency under a few seconds.

The architecture for this is a separate consumer group that reads from your MQTT gateway (or a Kafka topic that the gateway publishes to) and evaluates each reading against detection rules.

A simple but effective approach: maintain a sliding window of recent values per device per metric in Redis, and evaluate rules against the window on each new reading.

import { createClient } from "redis";

const redis = createClient({ url: process.env.REDIS_URL });

interface AnomalyRule {
  metric: string;
  windowSizeSeconds: number;
  thresholdFn: (values: number[]) => boolean;
  severity: "warning" | "critical";
  description: string;
}

const ANOMALY_RULES: AnomalyRule[] = [
  {
    metric: "temperature_celsius",
    windowSizeSeconds: 60,
    thresholdFn: (values) => Math.max(...values) > 85,
    severity: "critical",
    description: "Temperature exceeded 85°C in past 60 seconds",
  },
  {
    metric: "battery_voltage",
    windowSizeSeconds: 300,
    thresholdFn: (values) => {
      // Alert if voltage dropped more than 15% in past 5 minutes
      if (values.length < 2) return false;
      const first = values[0];
      const last = values[values.length - 1];
      return (first - last) / first > 0.15;
    },
    severity: "warning",
    description: "Battery voltage dropped >15% in 5 minutes",
  },
];

export async function evaluateReading(
  deviceId: string,
  metric: string,
  value: number,
  timestamp: number
): Promise<void> {
  const key = `window:${deviceId}:${metric}`;
  const cutoff = timestamp - 300_000; // keep up to 5 minutes of values

  // Store as sorted set: score = timestamp, member = "timestamp:value"
  await redis.zAdd(key, {
    score: timestamp,
    value: `${timestamp}:${value}`,
  });

  // Trim old entries beyond the max window we need
  await redis.zRemRangeByScore(key, "-inf", cutoff);
  await redis.expire(key, 600); // TTL: evict if device stops sending

  const matchingRules = ANOMALY_RULES.filter((r) => r.metric === metric);
  if (matchingRules.length === 0) return;

  for (const rule of matchingRules) {
    const windowStart = timestamp - rule.windowSizeSeconds * 1000;
    const rawEntries = await redis.zRangeByScore(key, windowStart, "+inf");

    const windowValues = rawEntries.map((entry) =>
      parseFloat(entry.split(":")[1])
    );

    if (rule.thresholdFn(windowValues)) {
      await emitAnomaly(deviceId, rule, timestamp);
    }
  }
}

async function emitAnomaly(
  deviceId: string,
  rule: AnomalyRule,
  timestamp: number
): Promise<void> {
  // Publish to anomaly topic; a downstream consumer handles notification routing
  await redis.publish(
    "anomalies",
    JSON.stringify({
      deviceId,
      metric: rule.metric,
      severity: rule.severity,
      description: rule.description,
      detectedAt: timestamp,
    })
  );
}

This approach keeps detection logic in application code, which makes rules easy to unit test and deploy without restarting the pipeline. The Redis sorted set gives you O(log N) inserts and O(log N + k) range queries, which is fast enough for per-reading evaluation at high throughput.

For more complex patterns (multi-metric correlation, statistical anomaly detection via Z-score, or ML-based detection), consider a proper stream processor like Apache Flink or a managed alternative like AWS Kinesis Data Analytics. The Redis approach is a good starting point for simple threshold rules and scales to several thousand readings per second on a single Redis instance.

Real-Time Fleet Dashboard Architecture

A fleet dashboard has two data access patterns that require different solutions.

Historical aggregates are served from the time-series database. “Average temperature per device over the past 24 hours” is a query: run it when the dashboard loads, cache the result for a few minutes, and re-run periodically. TimescaleDB continuous aggregates or InfluxDB tasks pre-compute common rollups so these queries return in milliseconds rather than scanning full hypertables.

Live readings require a push mechanism. Polling the database every second per device does not scale. At 1,000 devices, that is 1,000 queries per second for data that arrives faster than you can page-refresh anyway.

The right architecture: a WebSocket gateway that subscribes to device topics and pushes readings to dashboard clients.

import { WebSocketServer, WebSocket } from "ws";
import mqtt, { MqttClient } from "mqtt";

interface DashboardSubscription {
  ws: WebSocket;
  fleetId: string;
  deviceIds: Set<string>; // empty set = subscribe to entire fleet
}

const subscriptions = new Map<WebSocket, DashboardSubscription>();

export function createDashboardGateway(
  wss: WebSocketServer,
  mqttClient: MqttClient
): void {
  wss.on("connection", (ws) => {
    ws.on("message", (data) => {
      try {
        const msg = JSON.parse(data.toString());

        if (msg.type === "subscribe") {
          subscriptions.set(ws, {
            ws,
            fleetId: msg.fleetId,
            deviceIds: new Set(msg.deviceIds ?? []),
          });
        }
      } catch (err) {
        console.error({ err }, "Failed to parse dashboard message");
      }
    });

    ws.on("close", () => {
      subscriptions.delete(ws);
    });
  });

  mqttClient.on("message", (topic, payload) => {
    // topic: fleet/<fleetId>/devices/<deviceId>/telemetry
    const parts = topic.split("/");
    if (parts.length < 5) return;

    const fleetId = parts[1];
    const deviceId = parts[3];

    let parsed: unknown;
    try {
      parsed = JSON.parse(payload.toString());
    } catch {
      return;
    }

    const message = JSON.stringify({ deviceId, data: parsed });

    for (const [, sub] of subscriptions) {
      if (sub.fleetId !== fleetId) continue;
      if (sub.deviceIds.size > 0 && !sub.deviceIds.has(deviceId)) continue;
      if (sub.ws.readyState !== WebSocket.OPEN) continue;

      sub.ws.send(message);
    }
  });
}

This gateway fans out each MQTT message to all dashboard clients subscribed to that fleet. The connection is stateful: the client sends a subscribe message after connecting to declare which fleet and devices it cares about.

For large fleets (tens of thousands of devices), fan-out per message can become expensive if many dashboard sessions are open. The mitigation is to rate-limit delivery per WebSocket connection: buffer readings for 200-500ms and send a batch, rather than forwarding every individual reading immediately.

Production Considerations

Time synchronization on devices. Device clocks drift and may not have NTP access. A reading arriving with a timestamp three days in the past or ten minutes in the future causes problems for time-series queries and anomaly detection windows. Reject readings with timestamps more than a configurable tolerance (5 minutes is a reasonable default) from server time, or record both device_time and server_ingestion_time and let queries use server time as the authoritative timestamp.

Backpressure at the ingestion gateway. If your storage layer cannot absorb the incoming write rate, the MQTT broker’s in-memory message queues will grow unbounded. Configure your broker with flow control limits and monitor queue depth. When queue depth grows under sustained load, this is a signal to scale the ingestion gateway horizontally or increase the batch flush size.

Cardinality in your time-series schema. High cardinality in InfluxDB tags (unique device IDs, unique session IDs) causes memory pressure in the TSM index. If you have millions of unique devices, put device IDs in fields rather than tags, and index on a lower-cardinality dimension like fleet_id. In TimescaleDB, this problem appears as index bloat: use partial indexes and hypertable chunk exclusion to keep query plans efficient.

Handling device reconnections during anomaly detection. When a device reconnects after a network outage, it may send a burst of buffered readings with old timestamps. Your anomaly detection window will see a sudden spike of historical data and may trigger false positives. Add a reconnect detection signal (MQTT’s Last Will and Testament combined with a reconnect event) and pause anomaly evaluation for a configurable warm-up period after reconnect.

Schema evolution for device firmware updates. Devices ship to the field and stay there for years. A firmware update may change the structure of the telemetry payload. Plan a versioned topic format (fleet/+/devices/+/telemetry/v2) or a version field in the payload, and run both parsers in parallel during the rollout window. Removing support for an old format is a deployment decision that needs coordination with the hardware team.

The Layer Map

The complete pipeline for a device fleet backend:

1. Device firmware publishes to MQTT broker using QoS 0 (high-frequency telemetry) or QoS 1 (events and alerts)
2. MQTT broker (Mosquitto / EMQX / AWS IoT Core) handles fan-out and protocol termination
3. Ingestion gateway subscribes to device topics, normalizes payloads, batches writes, and publishes to an internal Kafka topic for parallel consumption
4. Storage consumer reads from Kafka, writes to TimescaleDB (or InfluxDB / QuestDB depending on write throughput requirements)
5. Anomaly detection consumer reads from Kafka, evaluates rules against Redis sliding windows, publishes anomalies to an alert topic
6. Alert router consumes anomalies and delivers notifications via email, SMS, or webhook
7. Dashboard WebSocket gateway subscribes to the MQTT broker and pushes live readings to browser clients
8. Historical query layer serves pre-computed aggregates from TimescaleDB continuous aggregates or InfluxDB tasks

Each consumer group in step 3 and 4 can be scaled independently. The Kafka topic is the durability buffer that decouples ingestion rate from storage write rate.

Closing

IoT pipeline architecture is mostly about choosing the right protocol and storage model for the access patterns you actually have, then being honest about where your throughput ceilings are before you hit them in production. MQTT for constrained devices, time-series storage with intentional schema design, Redis-backed sliding windows for anomaly detection, and WebSocket fan-out for dashboards: this stack covers the majority of fleet telemetry use cases without requiring exotic infrastructure.

The problems that catch teams off guard are operational: clock drift, cardinality growth, reconnect storms, and schema evolution. None of these are hard to handle if you account for them during the design phase. They are genuinely painful to retrofit into a production system that is already ingesting from 10,000 devices.