Ingest MQTT Data into EMQX Tables

EMQX Tables is a native, fully managed time-series data storage service built into the EMQX Platform. It is optimized for high-throughput, low-latency ingestion and analysis of MQTT data, making it ideal for Internet of Things (IoT) use cases.

Powered by GreptimeDB, EMQX Tables integrates seamlessly with EMQX Broker and supports InfluxDB Line Protocol, enabling efficient storage, querying, and visualization of telemetry data.

To learn more, see the EMQX Tables Overview.

This guide walks you through ingesting MQTT data into EMQX Tables by:

• Establishing a private connection between EMQX Broker and EMQX Tables
• Creating a connector
• Creating a rule with a Line Protocol writer
• Testing data ingestion and querying results

Prerequisites

You have created both:

• EMQX Dedicated Flex deployment
• EMQX Tables deployment

You’re familiar with:

• EMQX Rules
• Data Integration in EMQX
• InfluxDB Line Protocol

Establish Private Connection with EMQX Tables

To allow secure communication between your EMQX Broker and EMQX Tables deployments, you need to connect the Broker deployment with the EMQX Tables deployment through a private connection.

1. In your Broker deployment, navigate to Network Management.

2. Click + Connect to EMQX Tables in the Connect to EMQX Tables section.

3. In the Select an EMQX Tables Instance dropdown, select the deployment you want to connect to.

4. Click Confirm to establish the connection.

5. Wait for the endpoint status to change to Running. Once the status is Running, the connection is established and ready for use.

   

Create an EMQX Tables Connector

Before you can write data, create a connector to EMQX Tables:

1. Go to your EMQX Broker deployment. Click Data Integration from the left menu.

2. If it is the first time for you to create a connector, navigate to or search for EMQX Tables. If you have already created connectors, click + New Connector and then select EMQX Tables.

3. On the New Connector page, fill in the connection details by copying them from the Connection Info area on your EMQX Tables deployment's Deployment Overview page:

   Field	Connection Info	Example Value
   Connector Name		Auto-generated
   Server Host	Private Host	e.g.: xxx.us-east-1.aws.xxx.com:4001
   Database	Database Name	public
   Username	Username	Auto-generated
   Password	Password	Auto-generated

4. Click Test to verify connectivity. If the EMQX Tables service is accessible, a success prompt will be returned.

5. Click New to complete the creation. You can now proceed to create a rule using this connector.

Create a Rule for Data Ingestion into EMQX Tables

Next, you need to create a rule to specify the data to be written and add corresponding actions in the rule to write them to EMQX Tables.

1. In the Rules section, click New Rule, or use the Actions icon next to your connector.

2. Define the SQL rule in the SQL Editor. In this demonstration, the goal is to trigger the engine when the client sends a temperature and humidity message to the temp_hum/emqx topic. Here you can set the SQL as follows:

     SELECT
       timestamp,
       payload.location as location,
       payload.temp as temp,
       payload.hum as hum
     FROM "temp_hum/emqx"

   TIP

   If you are a beginner user, click Try It Out to learn and test the SQL rule.

3. Click Next to append an action to the rule.

4. In the Connector dropdown, select the connector you just created.

5. Set Time Precision to millisecond (default).

6. Configure the Write Syntax to define the Line Protocol format for data parsing and writing into EMQX Tables.

   Specify a text-based format that provides the measurement, tag set, field set, timestamp of a data point, and supported placeholders. See also InfluxDB 2.3 Line Protocol and InfluxDB 1.8 Line Protocol.

   For example:

    temp_hum,location=${location} temp=${temp},hum=${hum} ${timestamp}

   TIP

   • To write a signed integer type value, add i as the type identifier after the placeholder, for example, ${payload.int}i. See InfluxDB 1.8 write integer value.
   • To write an unsigned integer type value, add u as the type identifier after the placeholder, for example, ${payload.int}u. See InfluxDB 1.8 write integer value.

7. Click Confirm to save the rule.

8. In the Successful new rule pop-up, click Back to Rules to complete the rule creation.

Test the Rule and Query the Data

You are recommended to use MQTTX or other client tools to simulate temperature and humidity data reporting. For quick demonstration, you can just use built-in diagnostic tool inside your broker deployment by clicking the Online Test from the left menu.

1. In Online Test, connect to the deployment using username and password or auto-generated authentication.

2. In the Messages section, send a message:

   • Topic: temp_hum/emqx

   • Payload:

     {
       "temp": 27.5,
       "hum": 41.8,
       "location": "Prague"
     }

   

3. Go to your EMQX Tables deployment. Click Data Explorer from the left menu.

4. Run the following SQL to query the data ingested into the table public:

   select * from "temp_hum"

   You should see there is one record in the Query Result table.

   

5. View the rule statistics in your EMQX Broker deployment. Click the rule ID in the rule list, and you can see the statistics of the rule and action associated with this rule.