14 Lab: MQTT to InfluxDB and Grafana Pipeline

data-storage

lab

mqtt

influxdb

Lab Overview

Duration: 90 minutes | Difficulty: Intermediate

Learning Objectives:

Build end-to-end data pipeline: Sensor -> MQTT -> Database -> Dashboard
Store time-series data in InfluxDB
Create real-time visualizations with Grafana
Understand data flow in IoT architectures

Materials Needed:

Docker installed on your computer
Completed Lab 1
2GB free disk space

Prerequisites: Completed Lab: First MQTT Message, basic Docker knowledge helpful

14.0.0.1 Step 1: Start Infrastructure (15 min)

Create docker-compose.yml:

version: '3'
services:
  mosquitto:
    image: eclipse-mosquitto:2
    ports:
      - "1883:1883"
    volumes:
      - ./mosquitto.conf:/mosquitto/config/mosquitto.conf

  influxdb:
    image: influxdb:2.7
    ports:
      - "8086:8086"
    environment:
      - DOCKER_INFLUXDB_INIT_MODE=setup
      - DOCKER_INFLUXDB_INIT_USERNAME=admin
      - DOCKER_INFLUXDB_INIT_PASSWORD=adminpassword
      - DOCKER_INFLUXDB_INIT_ORG=iotclass
      - DOCKER_INFLUXDB_INIT_BUCKET=sensors

  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    depends_on:
      - influxdb

Create mosquitto.conf:

listener 1883
allow_anonymous true

Start the stack:

docker-compose up -d

Checkpoint 1

Verify all services are running:

docker-compose ps

All three services should show “Up” status.

14.0.0.2 Step 2: Bridge MQTT to InfluxDB (20 min)

Create mqtt_to_influx.py:

import paho.mqtt.client as mqtt
from influxdb_client import InfluxDBClient, Point
from influxdb_client.client.write_api import SYNCHRONOUS
import json

# InfluxDB connection
influx_client = InfluxDBClient(
    url="http://localhost:8086",
    token="your-token-here",  # Get from InfluxDB UI
    org="iotclass"
)
write_api = influx_client.write_api(write_options=SYNCHRONOUS)

def on_message(client, userdata, msg):
    try:
        # Parse JSON payload
        data = json.loads(msg.payload.decode())

        # Create InfluxDB point
        point = Point("sensor_reading") \
            .tag("sensor", msg.topic.split("/")[-1]) \
            .field("temperature", float(data.get("temp", 0))) \
            .field("humidity", float(data.get("humidity", 0)))

        # Write to InfluxDB
        write_api.write(bucket="sensors", record=point)
        print(f"Stored: {data}")

    except Exception as e:
        print(f"Error: {e}")

# MQTT setup
mqtt_client = mqtt.Client()
mqtt_client.on_message = on_message
mqtt_client.connect("localhost", 1883)
mqtt_client.subscribe("sensors/#")

print("Bridge running... Press Ctrl+C to stop")
mqtt_client.loop_forever()

14.0.0.3 Step 3: Simulate Sensors (15 min)

Create sensor_simulator.py:

import paho.mqtt.client as mqtt
import json
import time
import random

client = mqtt.Client()
client.connect("localhost", 1883)

sensors = ["living_room", "bedroom", "kitchen"]

while True:
    for sensor in sensors:
        data = {
            "temp": round(20 + random.gauss(0, 2), 1),
            "humidity": round(50 + random.gauss(0, 5), 1)
        }
        client.publish(f"sensors/{sensor}", json.dumps(data))
        print(f"{sensor}: {data}")

    time.sleep(5)

14.0.0.4 Step 4: Configure Grafana Dashboard (30 min)

Open Grafana: http://localhost:3000 (admin/admin)
Add InfluxDB data source (Configuration -> Data Sources)
Create new dashboard with time-series panel

Use this Flux query for the first panel:

from(bucket: "sensors")
  |> range(start: -1h)
  |> filter(fn: (r) => r._measurement == "sensor_reading")

Checkpoint 2

Your Grafana dashboard should show real-time temperature and humidity readings updating every 5 seconds.

14.0.0.5 Lab Validation

Success Criteria

Three Docker containers running (MQTT, InfluxDB, Grafana)
Sensor data flowing through MQTT to InfluxDB
Grafana dashboard displaying real-time charts
Can explain the role of each component in the pipeline

Build Lab 2 Dashboard in 90 Min

A student with basic Python knowledge wants to complete Lab 2 (Sensor Data Pipeline) in one sitting. Here’s the detailed timeline with actual commands and expected outcomes:

t=0-15 min: Infrastructure Setup

Create project directory: mkdir iot-lab2 && cd iot-lab2
Create docker-compose.yml (copy from lab), mosquitto.conf with 3 lines
Run docker-compose up -d — wait 2 minutes for images to download (first time only)
Verify: docker-compose ps shows all 3 services “Up” (green status)
Expected issue: InfluxDB takes 30 seconds to initialize — if Grafana shows “connection refused,” wait 1 minute

t=15-35 min: Bridge MQTT to InfluxDB

Get InfluxDB token: http://localhost:8086, create admin account, copy API token from Settings → Tokens
Install Python library: pip install influxdb-client paho-mqtt
Create mqtt_to_influx.py (code from Step 2 above), paste token at line 327
Run bridge: python3 mqtt_to_influx.py — expect “Bridge running… Press Ctrl+C to stop”

Test in another terminal:

mosquitto_pub -h localhost -t sensors/test -m '{"temp":25.5,"humidity":60}'

Verify: Check InfluxDB Data Explorer — see 1 point in “sensors” bucket under measurement “sensor_reading”

t=35-50 min: Sensor Simulator

Create sensor_simulator.py (38 lines from lab)
Run: python3 sensor_simulator.py — expect JSON output every 5 seconds
Verify bridge terminal shows “Stored: {temp: 20.3, humidity: 52.1}” messages

t=50-90 min: Grafana Dashboard

Open Grafana: http://localhost:3000, login admin/admin
Add the InfluxDB datasource with:
```
URL: http://influxdb:8086
Bucket: sensors
Organization: iotclass
```
Use the Docker service name influxdb, not localhost, and paste the API token you created earlier.
Test connection: should see green “Data source is working” message
Create new dashboard → Add panel → Time series visualization
Flux query builder: FROM bucket “sensors”, FILTER measurement = “sensor_reading”, FILTER _field = “temperature”, aggregate MEAN with 10s window
Add second query for humidity field
Panel title: “Smart Home Sensor Readings”, Y-axis label: “Value”, legend: show field names
Save dashboard as “IoT Lab 2”
Verify: See 3 lines (living_room, bedroom, kitchen temperatures) updating every 5 seconds

Success validation (t=90): Run all 3 terminals (bridge, simulator, Grafana), screenshot shows live updating graph with 10+ minutes of data. All Lab 2 success criteria boxes checked: ✓ Docker containers running, ✓ MQTT → InfluxDB flow, ✓ Grafana real-time dashboard, ✓ Can explain each component’s role.

Validate Checkpoints Before Debug

What they do wrong: A student starts Lab 2, follows the Docker Compose instructions, types docker-compose up -d, sees “done” in the terminal, and immediately jumps to writing the Python bridge script. After 30 minutes of coding, they run the script and get “Connection refused” errors. They spend 2 hours debugging Python code, reinstalling libraries, checking firewalls, and reading InfluxDB documentation — all while the actual problem is that InfluxDB container failed to start due to a port conflict.

Why it fails: Checkpoint 1 in Lab 2 explicitly says “Verify all services are running: docker-compose ps”. The student skipped this 10-second check. If they had run it, they would have seen InfluxDB status: “Restarting” instead of “Up”, indicating a startup failure. Running docker-compose logs influxdb would immediately show “Error: port 8086 already in use by process 4521”. Killing that process and restarting takes 2 minutes.

Correct approach: Treat every lab checkpoint as a STOP sign, not a suggestion. Checkpoints are placed at boundaries where skipped validation can turn one setup issue into several confusing symptoms. For Lab 2: - Checkpoint 1 (t=15 min): Run docker-compose ps and verify all 3 services show “Up” status. If any show “Restarting” or “Exit 1”, STOP and fix before proceeding. - Checkpoint 2 (t=50 min): Run curl http://localhost:8086/ping and expect HTTP 204 response. If connection refused, InfluxDB is not ready — wait 30 seconds and retry. - Checkpoint 3 (t=70 min): Before building the Grafana dashboard, confirm that InfluxDB has data:

influx query 'from(bucket:"sensors") |> range(start:-1h) |> limit(n:1)'

If the result is empty, the bridge is not working yet, so fix that before moving to Grafana.

Project consequence: Skipping checkpoints can turn a simple port conflict or token mismatch into hours of unrelated Python, Docker, or dashboard debugging. Checkpoint discipline catches boundary failures early instead of letting them compound.