Node-RED InfluxDB Tutorial: Storing and Querying MQTT Data

Node-RED and InfluxDB together give you a small but complete telemetry pipeline: MQTT messages come in, Node-RED shapes them, and InfluxDB stores them as time series you can query and chart. In this Node-RED InfluxDB tutorial you will run InfluxDB 2 in Docker, write MQTT readings with proper fields and tags, and query them back with Flux to draw a chart. Every step uses a free broker, so you can test the whole chain in under half an hour.

Architecture: MQTT In, Time Series Out

The pipeline has three roles, and keeping them separate makes it easy to scale later:

1. Devices publish JSON readings to MQTT topics such as connectmq/demo/a7f3k2/room1/telemetry.
2. The broker routes those messages to every subscriber. It does not store history; that is the database's job. If that distinction is new, here is what an MQTT broker does in detail.
3. Node-RED subscribes, validates and converts each message into an InfluxDB point, then writes it. A second branch queries InfluxDB for charts.

Node-RED is a good fit when you also want routing, alerts or a dashboard in the same place. If you only need ingestion at high volume, Telegraf's MQTT consumer is an alternative, covered in the guide to storing MQTT data in InfluxDB.

Step 1: Run InfluxDB 2 with Docker

The official image can initialise an organisation, bucket and admin token on first start, which saves several clicks:

docker run -d --name influxdb -p 8086:8086 \
  -v influxdb-data:/var/lib/influxdb2 \
  -e DOCKER_INFLUXDB_INIT_MODE=setup \
  -e DOCKER_INFLUXDB_INIT_USERNAME=admin \
  -e DOCKER_INFLUXDB_INIT_PASSWORD=change-me-please \
  -e DOCKER_INFLUXDB_INIT_ORG=demo-org \
  -e DOCKER_INFLUXDB_INIT_BUCKET=sensors \
  -e DOCKER_INFLUXDB_INIT_RETENTION=30d \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-super-secret-token \
  influxdb:2.7

Open http://localhost:8086, sign in, and you will find the sensors bucket with 30 days of retention. For production, create a separate API token with write access to that bucket only, rather than giving Node-RED the admin token. This tutorial targets InfluxDB 2.x; InfluxDB 3 changes the query side, so check its compatibility notes before copying the Flux examples.

Step 2: Publish Test Data to MQTT

This script simulates two rooms and publishes to broker.connectmq.com, the free public MQTT broker. Change a7f3k2 to your own random string; topics on the public broker are visible to everyone.

# pip install "paho-mqtt>=2.0"
import json
import random
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/a7f3k2"
ROOMS = {"room1": 22.0, "room2": 25.0}

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"influx-sim-{uuid.uuid4().hex[:8]}")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

try:
    while True:
        for room in ROOMS:
            ROOMS[room] += random.uniform(-0.2, 0.2)
            payload = {
                "temperature": round(ROOMS[room], 2),
                "humidity": round(random.uniform(38, 60), 1),
                "ts": int(time.time() * 1000),
            }
            client.publish(f"{PREFIX}/{room}/telemetry", json.dumps(payload), qos=1)
        time.sleep(5)
except KeyboardInterrupt:
    pass
finally:
    client.loop_stop()
    client.disconnect()

Including a device timestamp (ts) is good practice: if messages are delayed, the database still records when the reading was taken, not when it arrived.

Step 3: Install node-red-contrib-influxdb

In Node-RED, open Manage palette → Install and install node-red-contrib-influxdb. You get three nodes: influxdb out for single writes, influxdb batch for many points at once, and influxdb in for queries. All share an influxdb config node.

Configure the config node:

Step 4: Map MQTT Payloads to Fields and Tags

InfluxDB distinguishes fields (the measured values, not indexed) from tags (indexed metadata you filter and group by). Getting this right is the most important design decision in the whole Node-RED InfluxDB setup:

• Put numeric readings such as temperature and humidity in fields.
• Put low-cardinality identifiers such as room, line or device ID in tags.
• Never put unbounded values (timestamps, random IDs, free text) in tags; every unique tag combination creates a new series and memory usage grows with it.

Add an mqtt in node subscribed to connectmq/demo/a7f3k2/+/telemetry, output "a parsed JSON object", then this function node. The influxdb out node accepts an array of two objects: fields first, tags second.

// Function node "to influx point"
const p = msg.payload;
if (typeof p !== "object" || p === null) {
    return null;
}

const fields = {};
if (typeof p.temperature === "number") fields.temperature = p.temperature;
if (typeof p.humidity === "number") fields.humidity = p.humidity;
if (Object.keys(fields).length === 0) {
    node.warn("No numeric fields in " + msg.topic);
    return null;
}

// connectmq/demo/a7f3k2/room1/telemetry -> room1
const room = msg.topic.split("/")[3];

// Use the device timestamp when present (milliseconds)
if (typeof p.ts === "number") {
    fields.time = new Date(p.ts);
}

msg.payload = [fields, { room: room, source: "node-red" }];
msg.measurement = "environment";
return msg;

Wire it to an influxdb out node with organisation demo-org, bucket sensors, and time precision set to milliseconds. Leave the measurement empty in the node so msg.measurement is used. Deploy, wait a minute, and open the Data Explorer in the InfluxDB UI: the environment measurement appears with temperature and humidity fields and a room tag.

If your node version does not accept a time field inside the fields object, remove those two lines; InfluxDB will then use the write time.

Step 5: Query with Flux and Chart the Result

Querying is just as simple. An inject node fires every minute into a function node that builds the Flux query:

// Function node "build flux query"
msg.query = `
from(bucket: "sensors")
  |> range(start: -6h)
  |> filter(fn: (r) => r._measurement == "environment" and r._field == "temperature")
  |> aggregateWindow(every: 1m, fn: mean, createEmpty: false)
  |> keep(columns: ["_time", "_value", "room"])
`;
return msg;

Send it to an influxdb in node with organisation demo-org. The node returns an array of rows, each with _time, _value and the tag columns. Convert them into chart points, one series per room:

// Function node "rows to chart"
const rows = Array.isArray(msg.payload) ? msg.payload : [];
const byRoom = {};

for (const r of rows) {
    if (!byRoom[r.room]) byRoom[r.room] = [];
    byRoom[r.room].push({ x: Date.parse(r._time), y: Math.round(r._value * 100) / 100 });
}

const out = [{ payload: [] }];   // clear the chart before redrawing
for (const room of Object.keys(byRoom)) {
    out.push({ topic: room, payload: byRoom[room] });
}
return [out];

Wire this into a Dashboard 2.0 ui-chart with series from msg.topic, X key x and Y key y. Because the data now comes from the database, the chart survives restarts automatically. The Node-RED chart history guide compares this with lighter options if InfluxDB is more than you need, and the Dashboard 2.0 tutorial explains the chart settings.

Production Checklist

1. Use a dedicated InfluxDB token with write access to one bucket for the writer flow and read access for the query flow.
2. Set a bucket retention policy that matches how long you need raw data; downsample older data into a second bucket with an InfluxDB task if you need long trends.
3. Keep tags low-cardinality and consistent across devices.
4. Validate payloads before writing, as the function node does, so one bad device cannot fill the bucket with strings.
5. Use QoS 1 on the subscription and a persistent session so short Node-RED restarts do not lose readings.
6. Use the influxdb batch node if you write hundreds of points per second, grouping messages with a join node first.
7. Monitor write errors: add a catch node scoped to the influxdb out node and route errors to a debug node or alert.

Troubleshooting

The field type conflict deserves attention: InfluxDB fixes a field's type on first write within a shard. If a device sends 22 and later 22.5, the integer-versus-float mismatch is rejected. Since JSON does not distinguish them, sensors that send whole numbers occasionally trigger this. Sending values consistently from firmware (for example always with one decimal place) and checking the field type in the Data Explorer after the first writes avoids surprises later.

Secure the MQTT Side Too

A database with careful tokens is wasted if anyone can publish fake readings into the topics it ingests. On the public broker that is exactly the situation, which is why it is for testing only. For production, switch the Node-RED mqtt-broker config node to a private broker with TLS on port 8883 and credentials. On ConnectMQ, each broker gets a generated username, password and client ID, and client IDs are bound to those credentials, so a random client cannot impersonate your ingestion flow.

A cloud MQTT broker also simplifies the network: devices in the field and Node-RED next to InfluxDB both connect outbound to the broker, so you do not have to open ports into the database network. ConnectMQ itself does not write into databases; the honest pattern is exactly the one in this article, a Node-RED (or Telegraf) subscriber that writes to InfluxDB. For basics on brokers, wildcards and config nodes, see the Node-RED MQTT tutorial.

Start Storing Your MQTT Data with ConnectMQ

Create a free account at app.connectmq.com on the Serverless plan, no credit card needed. Create a broker, copy the credentials into your Node-RED config node, and your InfluxDB pipeline is fed by a private, TLS-encrypted stream that you can inspect live in the portal's topic explorer.

Frequently Asked Questions

Which InfluxDB versions does node-red-contrib-influxdb support?

It supports InfluxDB 1.x with InfluxQL, 1.8 with Flux, and 2.x with tokens, organisations and buckets. Choose the version in the config node so the nodes show the right options.

Should I write every MQTT message to InfluxDB?

Write every reading you may need later, but downsample high-frequency data in a separate bucket for long-term trends. If devices publish faster than you need, average in Node-RED before writing.

What is the difference between fields and tags in InfluxDB?

Fields store the measured values and are not indexed. Tags store metadata like room or device ID, are indexed, and should have a limited number of distinct values.

Can I use Grafana instead of a Node-RED chart?

Yes. Point Grafana at the same InfluxDB bucket and use the Flux query from this guide. Node-RED keeps doing the MQTT ingestion while Grafana handles visualisation.