IoT Integration with EMQX and InfluxDB

Overview

In this blog, we explore an alternative solution for AWS IoT Core and Amazon Timestream, both of which are serverless services AWS provides for tackling IoT use cases. But as Amazon Timestream is unavailable in the ap-south-1 region, InfluxDB is a good fit for the scenarios where we need the Amazon Timestream database. The goal is to minimize the IoT broker cost, which AWS IoT Core bills per message and rule-action triggered. Moreover, IoT Core is based on an EMQX broker and provides almost the same features.

EMQX

EMQX is an open-source MQTT broker designed for scalable and reliable messaging in IoT applications. It supports concurrent connections of up to 100 million MQTT connections with a single EMQX 5.0 cluster. It has robust features like message routing, high-availability clustering, and support for Webhooks and MQTT over WebSockets.

Here, the EMQX broker is installed as a single node in a single Amazon EC2 instance running Ubuntu and can handle up to 100,000 requests at a time.

The following steps must be followed to install an EMQX broker in Ubuntu.

1. Configure the EMQX APT source.

2. Install the latest version of EMQX.

• On successful installation, use the following command to start EMQX.

Once successfully installed, the user interface can be accessed from “http://public ip of ec2:18083. The first page will prompt the creation of an admin user and admin password, which can be created in this way only once. Below, the EMQX broker UI, which comes after the login, is shown.

InfluxDB OSS

InfluxDB OSS is an open-source time series database (TSDB). The InfluxDB data model efficiently organizes time series data using buckets and measurements. A bucket is a designated storage entity for time series data and can contain multiple measurements. Each measurement acts as a logical grouping for specific data types, ensuring that all points within a measurement share common tags. These tags, consisting of key-value pairs, hold metadata such as identifying the source (e.g., host, location, station). Fields, another type of key-value pair, represent data points that change over time, like temperature, pressure, or stock price. Timestamps associated with data entries ensure chronological ordering, aiding in seamless storage and retrieval.

Here, an Ubuntu Amazon EC2 instance is hosting an InfluxDB OSS v2.

The following steps should be followed to install.

1. To download the appropriate package.

2. Start the InfluxDB service:

Once installed, the UI to create a bucket and necessary API keys can be accessed at http:public ip of ec2:8086

The first page will prompt the creation of an admin user, admin password, organization, and Admin API key, which needs to be saved for future use in case of taking backups, etc.

Below, the UI interface is shown.

The below image shows how to create a bucket, which can be found on the left side menu bar at Load Data>>Bucket and then clicking the Create bucket button. While creating the bucket, we can set the retention period of the data.

Here, we have created a bucket named TestBucket, in which we’ll try to write data from EMQX.

The next step is to create an API key with fine grain access at bucket level. Here, we generate one API Key with only the necessary permissions to read/write into TestBucket.

Creation of a user and Webhook in EMQX

Unlike AWS IoT Core, EMQX allows three different types of authentication.

1. Password-Based, using the client ID or username and password
2. JWT, where the client can carry a JWT Token in the username or password.
3. SCRAM, enhanced authentication in MQTT 5.0, which enables two-way authentication between client and server.

For simplicity, we will use Password based authentication here. In EMQX, every device is coined as a user, and to create one, go to Access Control>>Authentication>>Add user(+ button). Below is a reference snapshot for the same.

Next is to create a webhook with the POST method, which can be done in Integration>>Webhook>>create. Below, a webhook created to write into a Bucket InfluxDB is shown.

The parameters that must be filled out are name, topic, Method, URL, and headers. The endpoint URL to write into the InfluxDB bucket is http://private or public ip of ec2 instance:8086/ api/v2/write?org= YOUR_ORG&bucket=YOUR_BUCKET&precision=ns.  The Headers that need to be added are {“Accept”:”application/json”, “Authentication”:”Api token generated in the previous step in InfluxDB”, “Content Type”:” text/plain; charset=utf-8”}  (in key value pair).

Template Creation and Connecting with InfluxDB

Once the webhook is created, the next step is to create a rule action similar to AWS IoT Core. To create one, go to Integration>>Rules>>Create, then give a rule name(optional), note(optional), and edit the SQL editor. The SQL syntax is of form

Tip: The SQL script can be verified in the rule page creation without triggering an action

Below is a sample rule action page.

Next, add an action, select “ Forwarding with Data Bridge” and select the webhook created in the previous section. Once selected, a form will populated, which can be edited if required. For writing data into the InfluxDB bucket, it is mandatory to edit the template body, which is located in the bottom in line format which is of form

Here this line template is used to send the data to InfluxDB “airSensors,sensor_id=${sensor_id} temperature =${temperature},humidity=${humidity},co=${co}”

Below a Data Bridge configuration is shown.

The EMQX broker is now ready to route the payload received in the topics mentioned in the SQL editor.

AWS IoT Core EMQX also provides a WebSocket client that can be used to check the whole pipeline located in Diagnose>>WebSocket Client.

Here we are going to publish a sample JSON payload {“sensor_id”: “TLM0201”,  “temperature”: 73.97038159354763,  “humidity”: 35.23103248356096,  “co”: 0.48445310567793615} to Topic “test/data” which will get written into “TestBucket” in “airSensor“ measurement with “sensor_id” as tag key and “temperature” ,”humidity” as field keys. Below, a snapshot of the WebSocet client is shown.

Below is an image of the received data in the bucket.

In addition, InfluxDB also provides data visualization tools like a graph, table, scatter plot, etc., which can be arranged in a dashboard to help verify the data received. Below, a sample dashboard is shown.

Conclusion

The integration of EMQX with InfluxDB forms a potent technical alliance, enabling seamless handling of vast IoT data streams. This integration optimizes IoT data flow, efficiently channeling MQTT messages from devices to InfluxDB for storage and analysis.

Drop a query if you have any questions regarding EMQX or InfluxDB and we will get back to you quickly.

About CloudThat

CloudThat is a leading provider of Cloud Training and Consulting services with a global presence in India, the USA, Asia, Europe, and Africa. Specializing in AWS, Microsoft Azure, GCP, VMware, Databricks, and more, the company serves mid-market and enterprise clients, offering comprehensive expertise in Cloud Migration, Data Platforms, DevOps, IoT, AI/ML, and more.

CloudThat is recognized as a top-tier partner with AWS and Microsoft, including the prestigious ‘Think Big’ partner award from AWS and the Microsoft Superstars FY 2023 award in Asia & India. Having trained 650k+ professionals in 500+ cloud certifications and completed 300+ consulting projects globally, CloudThat is an official AWS Advanced Consulting Partner, AWS Training Partner, AWS Migration Partner, AWS Data and Analytics Partner, AWS DevOps Competency Partner, Amazon QuickSight Service Delivery Partner, Amazon EKS Service Delivery Partner, Microsoft Gold Partner, AWS Microsoft Workload Partners, Amazon EC2 Service Delivery Partner, and many more.

To get started, go through our Consultancy page and Managed Services Package, CloudThat’s offerings.