--- # === IDENTITY === id: software/system-design/iot-data-pipeline/2026 canonical_question: "How do I design an IoT data ingestion pipeline?" aliases: - "IoT data ingestion architecture" - "How to build an IoT streaming pipeline" - "MQTT to time-series database pipeline" - "Real-time IoT data processing architecture" entity_type: software_reference domain: software > system-design > iot_data_pipeline region: global jurisdiction: global temporal_scope: 2020-2026 # === VERIFICATION === last_verified: 2026-02-23 confidence: 0.90 version: 1.0 first_published: 2026-02-23 # === TEMPORAL VALIDITY === temporal_validity: status: stable last_breaking_change: "MQTT 5.0 (2019-03-07)" next_review: 2026-08-22 change_sensitivity: medium # === CONSTRAINTS === constraints: - "Always implement backpressure handling between ingestion and processing layers to prevent data loss during traffic spikes" - "Use TLS 1.2+ for all device-to-broker connections; never transmit telemetry over unencrypted channels" - "Design for at-least-once delivery by default; exactly-once requires idempotent consumers or deduplication at the storage layer" - "Time-series databases require monotonically increasing timestamps; clock drift on edge devices must be corrected before ingestion" # === SKIP CONDITIONS === skip_this_unit_if: - condition: "You need to process batch ETL data from databases or files, not streaming device telemetry" use_instead: "software/system-design/batch-etl-pipeline/2026" - condition: "You need real-time analytics on user clickstream data, not IoT sensor data" use_instead: "software/system-design/event-streaming-architecture/2026" # === AGENT HINTS === inputs_needed: - key: "protocol_choice" question: "What protocol do your IoT devices use to send data?" type: choice options: ["MQTT", "AMQP", "HTTP", "CoAP", "Not decided yet"] - key: "scale" question: "How many devices will send data concurrently?" type: choice options: ["< 1,000", "1,000 - 100,000", "100,000 - 1,000,000", "> 1,000,000"] - key: "latency_requirement" question: "What is your acceptable end-to-end latency from device to queryable storage?" type: choice options: ["< 1 second", "1-10 seconds", "10-60 seconds", "Minutes are fine"] # === DISTRIBUTION === canonical_source: "https://knowledgelib.io/software/system-design/iot-data-pipeline/2026" suggested_citation: "Source: knowledgelib.io — AI Knowledge Library (verified 2026-02-23)" # === RELATED UNITS === related_kos: depends_on: [] related_to: - id: "software/system-design/event-streaming-architecture/2026" label: "Event Streaming Architecture (Kafka/Kinesis)" solves: - id: "software/system-design/time-series-storage/2026" label: "Time-Series Database Selection" alternative_to: - id: "software/system-design/batch-etl-pipeline/2026" label: "Batch ETL Pipeline Design" often_confused_with: - id: "software/system-design/message-queue-architecture/2026" label: "Message Queue Architecture (task queues vs. event streams)" # === SOURCES (7 authoritative sources) === sources: - id: src1 title: "7 Patterns for IoT Data Ingestion and Visualization" author: AWS url: https://aws.amazon.com/blogs/iot/7-patterns-for-iot-data-ingestion-and-visualization-how-to-decide-what-works-best-for-your-use-case/ type: technical_blog published: 2024-06-15 reliability: high - id: src2 title: "Architecture Best Practices for Azure IoT Hub" author: Microsoft url: https://learn.microsoft.com/en-us/azure/well-architected/service-guides/azure-iot-hub type: official_docs published: 2024-11-20 reliability: high - id: src3 title: "Building Industrial IoT Data Streaming Architecture with MQTT" author: HiveMQ url: https://www.hivemq.com/blog/building-industrial-iot-data-streaming-architecture-mqtt/ type: technical_blog published: 2024-09-10 reliability: moderate_high - id: src4 title: "MQTT with TimescaleDB: IoT Time-Series Data Management" author: EMQX url: https://www.emqx.com/en/blog/build-an-iot-time-series-data-application-for-energy-storage-with-mqtt-and-timescale type: technical_blog published: 2024-07-18 reliability: moderate_high - id: src5 title: "MQTT vs AMQP for IoT Communications" author: HiveMQ url: https://www.hivemq.com/blog/mqtt-vs-amqp-for-iot/ type: technical_blog published: 2024-05-22 reliability: moderate_high - id: src6 title: "Anti-patterns for Data Ingestion and Processing" author: AWS url: https://docs.aws.amazon.com/wellarchitected/latest/devops-guidance/anti-patterns-for-data-ingestion-and-processing.html type: official_docs published: 2024-08-01 reliability: high - id: src7 title: "Kafka for IoT: Key Capabilities and Top Use Cases" author: Instaclustr url: https://www.instaclustr.com/education/apache-kafka/kafka-for-iot-4-key-capabilities-and-top-use-cases-in-2025/ type: technical_blog published: 2025-01-15 reliability: moderate_high --- # IoT Data Ingestion Pipeline Design ## TL;DR - **Bottom line**: An IoT data pipeline consists of four layers -- device ingestion (MQTT broker), stream buffering (Kafka/Kinesis), processing (stream processor + rules engine), and time-series storage -- connected by backpressure-aware channels with at-least-once delivery guarantees. - **Key tool/command**: `mosquitto_pub -h broker.example.com -t sensors/temp -m '{"device":"d1","value":23.5,"ts":1708700000}'` - **Watch out for**: Skipping the stream buffer layer; connecting devices directly to your database causes cascading failures at scale. - **Works with**: MQTT 3.1.1/5.0, Apache Kafka 3.x, AWS IoT Core, Azure IoT Hub, TimescaleDB 2.x, InfluxDB 2.x/3.x. ## Constraints - Always implement backpressure handling between ingestion and processing layers to prevent data loss during traffic spikes - Use TLS 1.2+ for all device-to-broker connections; never transmit telemetry over unencrypted channels - Design for at-least-once delivery by default; exactly-once requires idempotent consumers or deduplication at the storage layer - Time-series databases require monotonically increasing timestamps; clock drift on edge devices must be corrected via NTP before ingestion - Separate the ingestion pipeline from the analytics/integration layer so complex queries never block device data intake [src2] ## Quick Reference | Component | Role | Technology Options | Scaling Strategy | |---|---|---|---| | Device Gateway | Accepts device connections, authenticates, terminates TLS | AWS IoT Core, Azure IoT Hub, EMQX, HiveMQ, Mosquitto | Horizontal broker clustering; 1M+ connections per cluster | | MQTT Broker | Pub/sub message routing with QoS levels | EMQX, HiveMQ, VerneMQ, Mosquitto, NanoMQ | Cluster with shared subscriptions; partition by topic namespace | | Protocol Adapter | Translates CoAP/HTTP/AMQP to internal format | AWS IoT Rules, Azure IoT Edge, custom bridge | Stateless; scale with load balancer | | Stream Buffer | Decouples ingestion from processing; absorbs bursts | Apache Kafka, Amazon Kinesis, Azure Event Hubs, Redpanda | Add partitions; increase retention for replay | | Schema Registry | Validates and evolves message schemas | Confluent Schema Registry, AWS Glue Schema Registry | Stateless reads; single-leader writes | | Stream Processor | Transforms, enriches, aggregates in real time | Apache Flink, Kafka Streams, AWS Lambda, Azure Stream Analytics | Parallel consumers per partition | | Rules Engine | Routes messages based on content/topic patterns | AWS IoT Rules, Node-RED, custom filter layer | Stateless; scale horizontally | | Time-Series DB | Stores telemetry with time-based partitioning | TimescaleDB, InfluxDB, Amazon Timestream, QuestDB | Hypertable chunking (TimescaleDB), sharding (InfluxDB) | | Object Storage | Long-term raw data archive | S3, Azure Blob, GCS | Lifecycle policies; tiered storage classes | | Edge Processor | Pre-filters and aggregates at the edge | AWS IoT Greengrass, Azure IoT Edge, Telegraf | Deploy per site/gateway | | Monitoring Stack | Pipeline health, throughput, latency dashboards | Grafana + Prometheus, CloudWatch, Datadog | Federated Prometheus for multi-region | | Device Registry | Tracks device identity, firmware, status | AWS IoT Device Defender, Azure DPS, custom DB | Read replicas; eventual consistency acceptable | ## Decision Tree > Full script: [decision-tree.txt](scripts/decision-tree.txt) (34 lines) ``` START: Choose your ingestion protocol ├── Devices are battery-powered / bandwidth-constrained? │ ├── YES → Use MQTT 5.0 (2-byte header, minimal overhead) [src5] │ │ ├── Need QoS 2 (exactly-once delivery)? │ │ │ ├── YES → MQTT QoS 2 (higher latency, 4-packet handshake) # ... (see full script) ``` ## Step-by-Step Guide ### 1. Set up the MQTT broker Deploy a clustered MQTT broker as the device-facing ingestion endpoint. Use MQTT 5.0 for shared subscriptions and topic aliases. Configure TLS with mutual authentication (mTLS) for production. [src3] ```bash # Deploy EMQX cluster via Docker Compose docker run -d --name emqx \ -p 1883:1883 \ -p 8883:8883 \ -p 8083:8083 \ -p 18083:18083 \ -e EMQX_LISTENERS__SSL__DEFAULT__SSL_OPTIONS__CERTFILE=/certs/server.pem \ -e EMQX_LISTENERS__SSL__DEFAULT__SSL_OPTIONS__KEYFILE=/certs/server.key \ emqx/emqx:5.5.1 ``` **Verify**: `mosquitto_pub -h localhost -p 1883 -t test/hello -m "ping"` and `mosquitto_sub -h localhost -p 1883 -t test/hello` should receive "ping" ### 2. Define a topic namespace convention Establish a hierarchical, lowercase topic structure before connecting any devices. This cannot easily be changed later. [src3] ``` # Topic structure: {org}/{site}/{device_type}/{device_id}/{measurement} # Examples: acme/factory-01/temperature/sensor-4a3b/reading acme/factory-01/vibration/motor-7c2d/reading acme/factory-01/+/+/alert # Wildcard subscription for all alerts ``` **Verify**: Subscribe to `acme/factory-01/#` and confirm messages from test publishers arrive with correct topic structure. ### 3. Bridge MQTT to a stream buffer (Kafka) Connect the MQTT broker to Apache Kafka using a Kafka connector or built-in MQTT-Kafka bridge. This decouples device ingestion from downstream processing. [src1] [src7] ```bash # Using Confluent MQTT Source Connector curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{ "name": "mqtt-source", "config": { "connector.class": "io.confluent.connect.mqtt.MqttSourceConnector", "mqtt.server.uri": "tcp://emqx:1883", "mqtt.topics": "acme/+/+/+/reading", "kafka.topic": "iot-telemetry-raw", "tasks.max": "4", "key.converter": "org.apache.kafka.connect.storage.StringConverter", "value.converter": "org.apache.kafka.connect.json.JsonConverter" } }' ``` **Verify**: `kafka-console-consumer --bootstrap-server localhost:9092 --topic iot-telemetry-raw --from-beginning` should show MQTT messages flowing into Kafka. ### 4. Add stream processing for transformation and enrichment Implement a stream processor that validates schemas, normalizes timestamps to UTC, enriches with device metadata, and routes to appropriate sinks. [src1] ```python # Using Faust (Python stream processing on Kafka) import faust from datetime import datetime, timezone app = faust.App('iot-processor', broker='kafka://localhost:9092') raw_topic = app.topic('iot-telemetry-raw', value_type=bytes) clean_topic = app.topic('iot-telemetry-clean', value_type=bytes) @app.agent(raw_topic) async def process_telemetry(stream): async for event in stream: data = json.loads(event) # Validate required fields if not all(k in data for k in ('device_id', 'value', 'ts')): continue # Drop malformed messages # Normalize timestamp to UTC milliseconds data['ts_utc'] = normalize_ts(data['ts']) # Enrich with device metadata from cache data['location'] = await device_registry.get(data['device_id']) await clean_topic.send(value=json.dumps(data).encode()) ``` **Verify**: Check `iot-telemetry-clean` topic for enriched, normalized messages with `ts_utc` field present. ### 5. Sink to time-series database Configure a Kafka Connect sink (or Telegraf consumer) to write processed telemetry into your time-series database with appropriate retention policies. [src4] ```sql -- TimescaleDB: Create hypertable for IoT telemetry CREATE TABLE telemetry ( time TIMESTAMPTZ NOT NULL, device_id TEXT NOT NULL, location TEXT, metric TEXT NOT NULL, value DOUBLE PRECISION NOT NULL ); SELECT create_hypertable('telemetry', 'time'); -- Add compression policy (compress chunks older than 7 days) ALTER TABLE telemetry SET ( timescaledb.compress, timescaledb.compress_segmentby = 'device_id', timescaledb.compress_orderby = 'time DESC' ); SELECT add_compression_policy('telemetry', INTERVAL '7 days'); -- Add retention policy (drop data older than 1 year) SELECT add_retention_policy('telemetry', INTERVAL '1 year'); ``` **Verify**: `SELECT count(*) FROM telemetry WHERE time > now() - interval '5 minutes';` should show increasing row counts as data flows in. ### 6. Configure monitoring and alerting Set up end-to-end pipeline observability covering broker metrics, Kafka consumer lag, processing throughput, and database write latency. [src3] ```yaml # Prometheus scrape config for IoT pipeline monitoring scrape_configs: - job_name: 'emqx' static_configs: - targets: ['emqx:18083'] metrics_path: '/api/v5/prometheus/stats' - job_name: 'kafka' static_configs: - targets: ['kafka:9308'] - job_name: 'timescaledb' static_configs: - targets: ['postgres-exporter:9187'] ``` **Verify**: Open Grafana at `http://localhost:3000` and confirm dashboards show broker connections, Kafka consumer lag, and DB write rates. ## Code Examples ### Python: MQTT Client Publishing Telemetry > Full script: [python-mqtt-client-publishing-telemetry.py](scripts/python-mqtt-client-publishing-telemetry.py) (25 lines) ```python # Input: Sensor readings from GPIO/serial # Output: JSON messages published to MQTT broker import paho.mqtt.client as mqtt # paho-mqtt==1.6.1 import json, time, ssl BROKER = "broker.example.com" # ... (see full script) ``` ### Python: TimescaleDB Time-Series Insertion with Batch Writes > Full script: [python-timescaledb-time-series-insertion-with-batc.py](scripts/python-timescaledb-time-series-insertion-with-batc.py) (30 lines) ```python # Input: List of telemetry dicts from stream processor # Output: Batch-inserted rows in TimescaleDB hypertable import psycopg2 # psycopg2-binary==2.9.9 from psycopg2.extras import execute_values conn = psycopg2.connect( # ... (see full script) ``` ### JavaScript/Node.js: MQTT Subscriber to Kafka Bridge ```javascript // Input: MQTT messages from broker // Output: Messages forwarded to Kafka topic const mqtt = require("mqtt"); // mqtt@5.3.0 const { Kafka } = require("kafkajs"); // kafkajs@2.2.4 const mqttClient = mqtt.connect("mqtts://broker.example.com:8883", { ca: fs.readFileSync("/certs/ca.pem"), cert: fs.readFileSync("/certs/client.pem"), key: fs.readFileSync("/certs/client.key"), protocolVersion: 5, }); const kafka = new Kafka({ brokers: ["kafka:9092"] }); const producer = kafka.producer(); async function start() { await producer.connect(); mqttClient.subscribe("acme/+/+/+/reading", { qos: 1 }); mqttClient.on("message", async (topic, message) => { const key = topic.split("/")[3]; // device_id as partition key await producer.send({ topic: "iot-telemetry-raw", messages: [{ key, value: message.toString() }], }); }); } start().catch(console.error); ``` ## Anti-Patterns ### Wrong: Connecting devices directly to the database ```python # BAD -- No buffer layer; DB outage = data loss, connection exhaustion at scale import psycopg2 def on_sensor_read(value): conn = psycopg2.connect(host="db.example.com", dbname="iot") cur = conn.cursor() cur.execute("INSERT INTO telemetry VALUES (now(), %s)", (value,)) conn.commit() conn.close() # Connection per write = catastrophic at 10K devices ``` ### Correct: Buffer through a message broker ```python # GOOD -- MQTT broker absorbs bursts; Kafka buffers for processing import paho.mqtt.client as mqtt def on_sensor_read(value): client.publish("sensors/temp", json.dumps({"value": value}), qos=1) # Kafka consumer writes to DB in batches, handles backpressure ``` ### Wrong: Using a single MQTT topic for all devices ```python # BAD -- No topic hierarchy; impossible to subscribe selectively client.publish("all-data", json.dumps({"device": "s1", "type": "temp", "value": 23})) # Every consumer gets every message; no filtering at broker level ``` ### Correct: Hierarchical topic namespace with wildcards ```python # GOOD -- Granular subscriptions; broker-level filtering client.publish("acme/factory-01/temperature/sensor-001/reading", payload) # Subscribe to specific: "acme/factory-01/temperature/+/reading" # Subscribe to all alerts: "acme/+/+/+/alert" ``` ### Wrong: Polling the database for new IoT data ```python # BAD -- Wastes resources; high latency; lock contention at scale while True: rows = db.query("SELECT * FROM telemetry WHERE processed = false LIMIT 100") for row in rows: process(row) db.execute("UPDATE telemetry SET processed = true WHERE id = %s", row.id) time.sleep(1) ``` ### Correct: Stream processing with Kafka consumers ```python # GOOD -- Push-based; partitioned parallelism; no database polling @app.agent(telemetry_topic) async def process(stream): async for batch in stream.take(100, within=5.0): results = [transform(msg) for msg in batch] await sink_to_db(results) ``` ### Wrong: Storing raw IoT data without compression or retention policies ```sql -- BAD -- Unbounded table growth; queries slow to a crawl within months CREATE TABLE telemetry ( time TIMESTAMPTZ, device_id TEXT, value FLOAT ); -- No partitioning, no compression, no retention = storage bomb ``` ### Correct: Hypertable with compression and retention ```sql -- GOOD -- Automatic chunking, compression, and cleanup SELECT create_hypertable('telemetry', 'time'); SELECT add_compression_policy('telemetry', INTERVAL '7 days'); SELECT add_retention_policy('telemetry', INTERVAL '1 year'); -- 10-20x compression ratio; old data automatically dropped ``` ## Common Pitfalls - **Clock drift on edge devices**: Devices without NTP sync produce out-of-order timestamps that corrupt time-series queries and aggregations. Fix: Mandate NTP on all devices; reject messages with timestamps > 5 min from server time. [src1] - **MQTT QoS 2 everywhere**: QoS 2 (exactly-once) requires a 4-packet handshake per message, cutting throughput by 50%+. Fix: Use QoS 1 (at-least-once) with idempotent consumers; reserve QoS 2 for critical commands only. [src5] - **No dead letter queue**: Malformed messages silently dropped, making debugging impossible. Fix: Route unparseable messages to a DLQ topic (e.g., `iot-telemetry-dlq`) and alert on DLQ depth. [src6] - **Single Kafka partition for all devices**: Destroys parallelism; one slow consumer blocks everything. Fix: Partition by `device_id` or `device_type` to enable parallel processing. Use at least `num_consumers * 2` partitions. [src7] - **Ignoring backpressure from the database**: Stream processor writes faster than the DB can ingest, causing OOM or connection pool exhaustion. Fix: Implement batch writes with configurable batch size and linger time; use circuit breakers. [src4] - **Flat JSON payloads without schema validation**: Schema evolution breaks downstream consumers silently. Fix: Use a Schema Registry (Avro/Protobuf) and validate at the broker or first processing stage. [src6] - **Not separating hot and cold storage paths**: Running analytics queries on the same table receiving real-time writes causes lock contention. Fix: Use continuous aggregates (TimescaleDB) or materialized views for dashboards; archive raw data to object storage. [src2] - **Hardcoded broker addresses in device firmware**: Makes broker migration or failover impossible without OTA firmware updates. Fix: Use DNS-based discovery or a bootstrap config endpoint that devices query on boot. [src3] ## Diagnostic Commands ```bash # Check MQTT broker status and connected clients mosquitto_sub -v -h broker.example.com -t '$SYS/broker/clients/connected' # Test MQTT connectivity and round-trip latency mosquitto_rtt -h broker.example.com -p 8883 --cafile ca.pem --count 10 # Check Kafka consumer group lag (detect processing bottlenecks) kafka-consumer-groups.sh --bootstrap-server kafka:9092 --group iot-processor --describe # Verify Kafka topic throughput kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list kafka:9092 --topic iot-telemetry-raw --time -1 # Check TimescaleDB hypertable chunk status psql -c "SELECT hypertable_name, chunk_name, range_start, range_end, is_compressed FROM timescaledb_information.chunks WHERE hypertable_name = 'telemetry' ORDER BY range_start DESC LIMIT 10;" # Monitor TimescaleDB write throughput psql -c "SELECT * FROM timescaledb_information.hypertable_stats WHERE hypertable_name = 'telemetry';" # Check InfluxDB write health (if using InfluxDB) curl -s http://localhost:8086/health | jq . ``` ## Version History & Compatibility | Version | Status | Breaking Changes | Migration Notes | |---|---|---|---| | MQTT 5.0 | Current (2019+) | Shared subscriptions, user properties, topic aliases | Upgrade from 3.1.1: update client libs; no wire-level breaking changes | | MQTT 3.1.1 | Widely deployed | — | Still supported by all brokers; lacks shared subs | | Kafka 3.x (KRaft) | Current (2024+) | ZooKeeper removed | Migrate from ZK mode: `kafka-metadata.sh` migration tool | | Kafka 2.8-2.x | Legacy | — | Upgrade to 3.x for KRaft; rolling upgrade supported | | TimescaleDB 2.x | Current | Continuous aggregates v2 | From 1.x: `timescaledb_pre_restore` / `timescaledb_post_restore` | | InfluxDB 3.x | Current (2024+) | New storage engine (Apache Arrow, DataFusion) | From 2.x: data migration required; API changes | | InfluxDB 2.x | Maintained | Flux query language deprecated in 3.x | Use InfluxQL for forward compatibility | ## When to Use / When Not to Use | Use When | Don't Use When | Use Instead | |---|---|---| | Thousands of devices sending periodic telemetry (temp, pressure, GPS) | Fewer than 10 devices sending data once per hour | Direct HTTP POST to a REST API + PostgreSQL | | You need sub-second ingestion latency for real-time dashboards | Data freshness of minutes/hours is acceptable | Batch file upload (CSV/Parquet) to object storage + scheduled ETL | | Devices are bandwidth-constrained (cellular, LoRa, satellite) | All devices are on reliable LAN/Wi-Fi with unlimited bandwidth | gRPC streaming or WebSocket connections | | You need message replay for reprocessing or debugging | Messages are fire-and-forget with no replay requirement | Simple pub/sub (Redis Streams, Amazon SNS) | | Multi-region deployment with edge processing requirements | Single-site deployment with local network only | Local MQTT broker + direct DB writes | ## Important Caveats - MQTT 5.0 shared subscriptions are required for horizontal scaling of consumers; MQTT 3.1.1 requires custom load-balancing logic at the application layer - TimescaleDB compression achieves 10-20x ratios but compressed chunks become read-only; decompression is needed before backfilling historical corrections - Kafka's exactly-once semantics (EOS) require `enable.idempotence=true` on producers and `isolation.level=read_committed` on consumers; misconfiguration silently falls back to at-least-once - Cloud-managed IoT services (AWS IoT Core, Azure IoT Hub) impose per-account message throughput limits that may require quota increases for large deployments (AWS default: 20,000 msg/sec per account per region) - InfluxDB 3.x dropped the Flux query language; new deployments should use InfluxQL or SQL to avoid migration pain ## Related Units - [Event Streaming Architecture (Kafka/Kinesis)](/software/system-design/event-streaming-architecture/2026) - [Time-Series Database Selection](/software/system-design/time-series-storage/2026) - [Batch ETL Pipeline Design](/software/system-design/batch-etl-pipeline/2026) - [Message Queue Architecture](/software/system-design/message-queue-architecture/2026)