Index MQTT Messages in Elasticsearch or OpenSearch

To index MQTT messages in Elasticsearch or OpenSearch, run a consumer that subscribes to your topics, converts each message into a JSON document with a timestamp, and writes batches through the _bulk API into a data stream with an explicit mapping. You can use Filebeat's built-in MQTT input, route through Logstash, or write a small consumer yourself. The choice matters less than the mapping: get that right first, and the rest is plumbing.

This tutorial covers the options, builds a Python bulk consumer that works with both Elasticsearch and OpenSearch, and shows how to test the pipeline end to end.

Pick an ingestion path

For Logstash, the most robust pattern is usually indirect: MQTT → Kafka → Logstash kafka input, or MQTT → Filebeat → Logstash beats input. That keeps Logstash on plugins that ship with it.

Design the index first

MQTT payloads are schemaless, and dynamic mapping turns that into trouble: the first message decides each field's type forever. If one device sends "temp": "21.5" as a string before the others send numbers, temp becomes text and your averages stop working. If a firmware bug sends random keys, you hit the field limit.

Use a data stream with an explicit index template and dynamic: false — unknown fields are kept in _source but not indexed:

curl -X PUT localhost:9200/_index_template/mqtt-telemetry \
  -H 'Content-Type: application/json' -d '{
  "index_patterns": ["mqtt-telemetry*"],
  "data_stream": {},
  "priority": 200,
  "template": {
    "mappings": {
      "dynamic": false,
      "properties": {
        "@timestamp":  { "type": "date" },
        "device_id":   { "type": "keyword" },
        "topic":       { "type": "keyword" },
        "qos":         { "type": "byte" },
        "temp":        { "type": "float" },
        "humidity":    { "type": "float" },
        "battery":     { "type": "integer" },
        "payload_raw": { "type": "text", "index": false }
      }
    }
  }
}'

Why these choices:

• keyword for IDs and topics — exact match, terms aggregations, cheap filtering. Never text.
• @timestamp from the device when it sends one, otherwise from the consumer. Data streams require it.
• payload_raw stored, not indexed — handy for debugging without paying for indexing it.
• Priority 200 — higher than the built-in templates, so your template wins for this name.

The same request works on OpenSearch, which also supports _index_template with data_stream. For lifecycle, attach an ILM policy (Elasticsearch) or an ISM policy (OpenSearch) to roll over and delete old backing indices — see MQTT data retention and storage for choosing retention periods.

Option 1: Filebeat MQTT input

Filebeat ships an mqtt input. A minimal configuration:

filebeat.inputs:
  - type: mqtt
    hosts: ["ssl://broker.connectmq.com:8883"]
    topics: ["testmqtt/es/+/telemetry"]
    qos: 1
    client_id: filebeat-mqtt-1
    username: "${MQTT_USER}"
    password: "${MQTT_PASS}"

processors:
  - decode_json_fields:
      fields: ["message"]
      target: "payload"

output.elasticsearch:
  hosts: ["http://localhost:9200"]

The MQTT payload lands in message, and decode_json_fields parses it into payload.*. By default Filebeat writes to its own data stream with its own template; if you want documents in mqtt-telemetry with the mapping above, set the output index and template options as described in the Filebeat documentation for your version, or reshape fields in an ingest pipeline. Filebeat is ideal when you prefer configuration over code and already run Beats.

Option 2: a Python bulk consumer

A custom consumer gives you exact control over the document shape and over what happens to bad messages. This one uses paho-mqtt 2.x and the official elasticsearch client (8.x). For OpenSearch, swap the import for from opensearchpy import OpenSearch, helpers — the bulk helper works the same way.

import json, os, queue, time
from datetime import datetime, timezone
import paho.mqtt.client as mqtt
from elasticsearch import Elasticsearch, helpers, ConnectionError as ESConnectionError

es = Elasticsearch("http://localhost:9200")
STREAM = "mqtt-telemetry"
FIELDS = {"temp", "humidity", "battery"}
buf: "queue.Queue[dict]" = queue.Queue(maxsize=50_000)

def on_connect(client, userdata, flags, reason_code, properties):
    print("connected:", reason_code)
    client.subscribe("testmqtt/es/+/telemetry", qos=1)

def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload)
    except ValueError:
        data = {}
    doc = {
        "@timestamp": data.get("ts") or datetime.now(timezone.utc).isoformat(),
        "device_id": msg.topic.split("/")[2],
        "topic": msg.topic,
        "qos": msg.qos,
        "payload_raw": msg.payload.decode("utf-8", "replace"),
        **{k: v for k, v in data.items() if k in FIELDS},
    }
    buf.put(doc)   # blocks when full -> natural backpressure on the MQTT socket

def actions(batch):
    for doc in batch:
        yield {"_op_type": "create", "_index": STREAM, "_source": doc}

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="es-indexer-1")
client.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
client.tls_set()
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 8883)
client.loop_start()

while True:
    batch, deadline = [], time.monotonic() + 1.0
    while len(batch) < 500 and time.monotonic() < deadline:
        try:
            batch.append(buf.get(timeout=0.1))
        except queue.Empty:
            pass
    if not batch:
        continue
    while True:
        try:
            ok, errors = helpers.bulk(es, actions(batch), raise_on_error=False)
            break
        except ESConnectionError as err:
            print("Elasticsearch unreachable, retrying in 2 s:", err)
            time.sleep(2)
    for e in errors:
        print("rejected:", json.dumps(e)[:300])
    print(f"indexed {ok}/{len(batch)}")

Design notes:

• Batching — flush every 500 documents or every second, whichever comes first. One HTTP request per MQTT message is the most common reason these pipelines fall behind.
• _op_type: create — data streams are append-only and only accept create.
• Backpressure — when Elasticsearch is slow, the bounded queue fills, buf.put blocks paho's network thread, and the broker stops delivering. Nothing grows without limit.
• Per-document errors — raise_on_error=False returns rejected documents instead of failing the whole batch. Log them, or index them into a separate mqtt-rejects index.
• Device timestamps — ts in epoch milliseconds works with the default date format. Parsing tips are in JSON payloads in Python.

The consumer acknowledges MQTT messages on receipt, so documents still in the buffer are lost if it crashes. If that matters, acknowledge after the bulk response with paho's manual_ack=True and a persistent session on your own broker, as shown in the Kafka bridge. To scale out, run several consumers behind a shared subscription; ordering does not matter here because every document carries its own timestamp.

Test the pipeline

Start a single-node Elasticsearch for local testing (security disabled — never do this outside your laptop):

docker run -d --name es -p 9200:9200 \
  -e discovery.type=single-node -e xpack.security.enabled=false \
  docker.elastic.co/elasticsearch/elasticsearch:8.15.0

Create the index template from above, start python mqtt_to_es.py, then publish test data:

for d in dev-a dev-b; do
  for i in 1 2 3; do
    mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
      -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
      -t testmqtt/es/$d/telemetry -m "{\"temp\": 2$i.5, \"battery\": 9$i}"
  done
done

# a payload that violates the mapping
mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
  -t testmqtt/es/dev-a/telemetry -m '{"temp": "n/a"}'

Check the result:

curl -s 'localhost:9200/mqtt-telemetry/_search?size=3&sort=@timestamp:desc' | jq '.hits.hits[]._source'

curl -s localhost:9200/mqtt-telemetry/_search -H 'Content-Type: application/json' -d '{
  "size": 0,
  "aggs": { "by_device": { "terms": { "field": "device_id" },
            "aggs": { "avg_temp": { "avg": { "field": "temp" } } } } }
}' | jq '.aggregations.by_device.buckets'

You should see six documents with per-device averages, and the consumer should log one rejected: line containing document_parsing_exception (older versions say mapper_parsing_exception) and failed to parse field [temp] of type [float]. That is the mapping doing its job. If you would rather keep such documents with the bad field dropped, add "ignore_malformed": true to the numeric fields. You can also send payloads by hand from the online MQTT client to try edge cases.

Publish test telemetry from your browser

Indexing MQTT data in Elasticsearch or OpenSearch is mostly a mapping problem: define fields explicitly, use a data stream with a lifecycle policy, batch writes, and decide in advance what happens to messages that do not fit. For pure numeric metrics, also look at MQTT to InfluxDB with Telegraf — many teams use both, events in one and metrics in the other.

Frequently asked questions