Skip to content
On this page

Ingest MQTT Data into OpenTSDB


EMQX Enterprise Edition features. EMQX Enterprise Edition provides comprehensive coverage of key business scenarios, rich data integration, product-level reliability, and 24/7 global technical support. Experience the benefits of this enterprise-ready MQTT messaging platform today.

OpenTSDB is a scalable, distributed time series database. EMQX supports integration with OpenTSDB. You can save MQTT messages to OpenTSDB for subsequent analysis and retrieval.

This page provides a comprehensive introduction to the data integration between EMQX and OpenTSDB with practical instructions on creating a rule and data bridge.

How It Works

OpenTSDB data integration is an out-of-the-box feature in EMQX that combines EMQX's real-time data capturing and transmission capabilities with OpenTSDB's data storage and analysis functionality. With a built-in rule engine component, the integration simplifies the process of ingesting data from EMQX to OpenTSDB for storage and analysis, eliminating the need for complex coding.

The diagram below illustrates a typical architecture of data integration between EMQX and OpenTSDB:

EMQX Integration OpenTSDB

EMQX inserts device data to OpenTSDB through the rule engine and data bridge. OpenTSDB provides extensive query capabilities, supporting the generation of reports, charts, and other data analysis results. Taking industrial energy management scenarios as an example, the workflow is as follows:

  1. Message publication and reception: Industrial devices establish successful connections to EMQX through the MQTT protocol and regularly publish energy consumption data using the MQTT protocol. This data includes production line identifiers and energy consumption values. When EMQX receives these messages, it initiates the matching process within its rules engine.
  2. Rule Engine Processes Messages: The built-in rule engine processes messages from specific sources based on topic matching. When a message arrives, it passes through the rule engine, which matches it with corresponding rules and processes the message data. This can include transforming data formats, filtering specific information, or enriching messages with context information.
  3. Data ingestion into OpenTSDB: Rules defined in the rule engine trigger operations to write messages to OpenTSDB.

After data is written to OpenTSDB, you can flexibly use the data, for example:

  • Connect to visualization tools like Grafana to generate charts based on the data, displaying energy storage data.
  • Connect to business systems for monitoring and alerting on the status of energy storage devices.

Features and Benefits

The OpenTSDB data integration offers the following features and advantages:

  • Efficient Data Processing: EMQX can handle a massive number of IoT device connections and message throughput, while OpenTSDB excels in data writing, storage, and querying, providing outstanding performance to meet the data processing needs of IoT scenarios without overburdening the system.
  • Message Transformation: Messages can undergo extensive processing and transformation through EMQX rules before being written into OpenTSDB.
  • Large-Scale Data Storage: By integrating EMQX with OpenTSDB, a vast amount of device data can be directly stored in OpenTSDB. OpenTSDB is a database designed for storing and querying large-scale time-series data, capable of efficiently handling the massive volume of time-series data generated by IoT devices.
  • Rich Query Capabilities: OpenTSDB's optimized storage structure and indexing enable rapid writing and querying of billions of data points, which is extremely beneficial for applications requiring real-time monitoring, analysis, and visualization of IoT device data.
  • Scalability: Both EMQX and OpenTSDB are capable of cluster scaling, allowing flexible horizontal expansion of clusters as business needs grow.

Before You Start

This section describes the preparations you need to complete before you start to create the OpenTSDB data bridges, including how to set up the OpenTSDB server.


Install OpenTSDB

Install OpenTSDB via Docker, and then run the docker image.

docker pull petergrace/opentsdb-docker

docker run -d --name opentsdb -p 4242:4242 petergrace/opentsdb-docker

Create OpenTSDB Data Bridge

This section demonstrates how to create an OpenTSDB data bridge in EMQX Dashboard. It assumes that you run both EMQX and OpenTSDB on the local machine. If you have OpenTSDB and EMQX running remotely, adjust the settings accordingly.

  1. Go to EMQX Dashboard, and click Integration -> Data Bridge.

  2. Click Create on the top right corner of the page.

  3. In the Create Data Bridge page, click to select OpenTSDB, and then click Next.

  4. Input a name for the data bridge. The name should be a combination of upper/lower case letters and numbers.

  5. Input the connection information:

    • URL: Input, or the actual URL if the OpenTSDB server runs remotely.
    • Leave other options as default.
  6. Advanced settings (optional): Choose whether to use sync or async query mode as needed. For details, see the relevant configuration information in Data Integration.

  7. Before clicking Create, you can click Test Connectivity to test that the bridge can connect to the OpenTSDB server.

  8. Then click Create to finish the creation of the data bridge.

    A confirmation dialog will appear and ask if you like to create a rule using this data bridge, you can click Create Rule to continue creating rules to specify the data to be saved into OpenTSDB. You can also create rules by following the steps in Create Rules for OpenTSDB Data Bridge.

Now the OpenTSDB data bridge should appear in the data bridge list (Integration -> Data Bridge) with Resource Status as Connected.

Create Rules for OpenTSDB Data Bridge

Now that you have successfully created the data bridge to OpenTSDB, you can continue to create rules to specify the data to be saved into OpenTSDB. The data reported by the client in this tutorial is as follows:

  • Topic: t/opents
  • Payload:
  "metric": "cpu",
  "tags": {
    "host": "serverA"
  1. Go to EMQX Dashboard, and click Integration -> Rules.

  2. Click Create at the upper right corner of the page.

  3. Input my_rule as the rule ID, and set the rules in the SQL Editor using the following statement, which means the MQTT messages under topic t/# will be saved to OpenTSDB.

    Note: If you want to specify your own SQL syntax, make sure that you have included all fields required by the data bridge in the SELECT part.

      		payload.metric as metric, payload.tags as tags, payload.value as value
  4. Click the Add Action button, select Forwarding with Data Bridge from the dropdown list, and then select the data bridge you just created under Data Bridge. Click the Add button.

  5. Click the Create button to finish the setup.

Now you have successfully created the data bridge to OpenTSDB. You can click Integration -> Flows to view the topology. It can be seen that the messages under topic t/# are sent and saved to OpenTSDB after parsing by rule my_rule.

Test Data Bridges and Rules

Use MQTTX to publish a message on topic t/opents.

mqttx pub -i emqx_c -t t/opents -m '{"metric":"cpu","tags":{"host":"serverA"},"value":12}'

Check the running status of the data bridge, there should be one new incoming and one new outgoing message.

Check whether the data is written into the OpenTSDB:

curl -X POST -H "Accept: Application/json" -H "Content-Type: application/json" http://localhost:4242/api/query -d '{
    "start": "1h-ago",
    "queries": [
            "aggregator": "last",
            "metric": "cpu",
            "tags": {
                "host": "*"
    "showTSUIDs": "true",
    "showQuery": "true",
    "delete": "false"

The formatted output of the query result is as follows:

    "metric": "cpu",
    "tags": {
      "host": "serverA"
    "aggregateTags": [],
    "query": {
      "aggregator": "last",
      "metric": "cpu",
      "tsuids": null,
      "downsample": null,
      "rate": false,
      "filters": [
          "tagk": "host",
          "filter": "*",
          "group_by": true,
          "type": "wildcard"
      "percentiles": null,
      "index": 0,
      "rateOptions": null,
      "filterTagKs": [
      "explicitTags": false,
      "useFuzzyFilter": true,
      "preAggregate": false,
      "rollupUsage": null,
      "rollupTable": "raw",
      "showHistogramBuckets": false,
      "useMultiGets": true,
      "tags": {
        "host": "wildcard(*)"
      "histogramQuery": false
    "tsuids": [
    "dps": {
      "1683532519": 12