MQTT to ClickHouse Integration: Fueling Real-Time IoT Data Analytics

Introduction

In today's fast-paced world, the ability to capture and process real-time data is essential for businesses to optimize operations and make informed decisions. This is where the powerful combination of MQTT (Message Queuing Telemetry Transport) integration with ClickHouse, an open-source columnar database management system, comes into play.

In this blog post, we will explore how MQTT integration with ClickHouse can unleash the power of data analysis and drive enhanced performance across these diverse industries.

Understanding MQTT and ClickHouse

The MQTT protocol, specifically designed for IoT applications, enables efficient and reliable communication between devices. It employs a lightweight publish-subscribe model, ensuring seamless data transmission even in resource-constrained environments. MQTT's low overhead and support for real-time data streaming make it an ideal choice for capturing and transmitting IoT data from various endpoints to data processing platforms.

ClickHouse, a high-performance, column-oriented SQL database management system (DBMS) for online analytical processing (OLAP), excels in processing and analyzing large volumes of data with minimal latency. Its columnar storage format and parallel query execution optimize data retrieval and aggregation, enabling lightning-fast analytical capabilities. ClickHouse is renowned for its scalability, allowing organizations to handle the ever-growing data volume generated by IoT devices while maintaining exceptional performance.

Benefits of MQTT Integration with ClickHouse

With MQTT as the communication layer, data from IoT devices can be easily transferred to ClickHouse, and get efficiently stored and processed with its high-performance analytical capabilities.

EMQX is the most popular MQTT broker. Its out-of-box data integration capabilities can seamlessly enable an MQTT-ClickHouse solution and bring several benefits:

• Real-time Data Streaming: EMQX is built for handling real-time data streams, ensuring efficient and reliable data transmission from source systems to ClickHouse. It enables organizations to capture and analyze data in real-time, making it ideal for use cases requiring immediate insights and actions.
• High Performance and Scalability: EMQX's distributed architecture and ClickHouse's columnar storage format enable seamless scalability as data volumes increase. This ensures consistent performance and responsiveness, even with large datasets.
• Flexibility in Data Transformation: EMQX provides a powerful SQL-based Rule Engine, allowing organizations to pre-process data before storing it in ClickHouse. It supports various data transformation mechanisms, such as filtering, routing, aggregation, and enrichment, enabling organizations to shape the data according to their needs.
• Easy Deployment and Management: EMQX provides a user-friendly interface for configuring data sources, pre-processing data rules, and ClickHouse storage settings. This simplifies the setup and ongoing management of the data integration process.
• Advanced Analytics: ClickHouse's powerful SQL-based query language and support for complex analytical functions empower users to gain valuable insights from IoT data, enabling predictive analytics, anomaly detection, and more.

By leveraging the EMQX data integration with the ClickHouse, organizations can streamline their data integration workflows, capture real-time data, ensure data security, and efficiently store and analyze it in ClickHouse. This made a powerful tool for those looking to harness the full potential of their data and drive data-driven insights and decision-making.

Use Cases in Various Industries

1. Industrial IoT and Predictive Maintenance: MQTT to ClickHouse integration enables real-time monitoring and analysis of data from machines, sensors, and production lines. Organizations can detect anomalies and patterns that indicate equipment failure or maintenance needs by collecting and analyzing data such as temperature, vibration, and operating conditions. This allows for proactive maintenance, reduced downtime, and optimized production processes.
2. Environmental Monitoring: MQTT to ClickHouse integration enables real-time data collection and analysis from environmental sensors, weather stations, and air quality monitoring devices. This use case finds agriculture, forestry, pollution control, and climate monitoring applications. Organizations can leverage the data to make informed decisions about irrigation, crop health, resource management, and environmental conservation.
3. Energy Management and Grid Optimization: MQTT to ClickHouse integration facilitates real-time monitoring and analysis of energy consumption data from smart meters, renewable energy sources, and grid infrastructure. This allows for effective energy management, load balancing, demand response, and integration of renewable energy resources into the grid.
4. Logistics and Supply Chain Optimization: MQTT to ClickHouse integration can be applied in logistics and supply chain management to monitor and analyze data from IoT devices such as GPS trackers, temperature sensors, and inventory management systems. This enables real-time tracking, route optimization, demand forecasting, and efficient inventory management, leading to streamlined operations and cost savings.

Integrating MQTT to ClickHouse with EMQX

This is a demo of an IoT-based car charging station. It utilizes EMQX to connect to the charging stations and uses MQTT to transmit their status and data. The data is stored in ClickHouse for real-time monitoring and analysis of the order count, charging duration, and charging status, providing insights into the usage frequency and charging demands of the charging stations.

You can also track the occupancy rate of the charging stations to evaluate their operational efficiency, which can serve as valuable decision-making information for market operations.

By integrating EMQX, ClickHouse, and relevant data analysis and visualization tools, you can build a powerful car charging station management system that enables real-time monitoring of charging station status and data, as well as data analysis and decision support.

Prerequisites

• Git
• Docker Engine: v20.10+
• Docker Compose: v2.20+

How it Works

This is a simple and effective architecture that utilizes the following key components.

Name	Version	Description
EMQX Enterprise	5.5.1+	MQTT broker used for message exchange between MQTT clients and the ClickHouse.
MQTTX CLI	1.9.9+	Command-line tool used to generate simulated data for testing.
ClickHouse	23.6.1	Charging station IoT data storage and management, as well as providing time aggregation and analysis capabilities for Grafana.
Grafana	9.5.1+	Visualization platform utilized to display and analyze the collected data.

Clone the Project Locally

Clone the emqx/mqtt-to-clickhouse repository locally using Git:

git clone https://github.com/emqx/mqtt-to-clickhouse
cd mqtt-to-clickhouse

the codebase consists of four parts:

• The emqx folder contains EMQX-Clickhouse integration configurations to automatically create rules and data bridges when launching EMQX.
• The clickhouse folder contains table init sql file.
• The mqttx/charging.js file offers scripts to simulate charging and transportation fleets for real-world data publishing.
• The docker-compose.yml orchestrates all components to launch the project with one click.

Start MQTTX CLI, EMQX and ClickHouse

Please make sure you have installed the Docker, and then run Docker Compose in the background to start the demo:

docker-compose up -d

MQTTX CLI will simulate the connection of 5 charging guns based on the OCPP protocol to EMQX. They will start simulating from 48 hours ago, publishing charging start and end messages (charging orders) via MQTT, and periodically reporting data such as power, voltage, meter readings, and charging duration for each order during the charging process. EMQX will create 2 rules to integrate the charging station with ClickHouse.

Next, we will show the structure of these messages and how EMQX writes them into ClickHouse through the rule engine and data integration functionality.

Charging Strat Message Processing

Topic

mqttx/simulate/charge/{clientId}/StartTransaction

Message Example

{"messageType":"Call","action":"StartTransaction","payload":{"connectorId":"f788a12a-1b7d-4205-9d8e-37307aae366a","transactionId":"6b738341-b9f3-4d42-adb8-9696e0b8aba6","idTag":"No. 2","timestamp":1710801744456,"reservationId":null,"stackLevel":0,"meterStart":20184.628600000484}}

ClickHouse does not need to store this event but rather performs storage after the charging process is completed. So there is no need to create corresponding rules in this step.

Meter Values Message Processing

Topic

mqttx/simulate/charge/{clientId}/MeterValues

Message Example

{"messageType":"Call","action":"MeterValues","payload":{"connectorId":"96b44678-186a-46b1-8c33-ef0be75600bb","transactionId":"45cb5c1a-f5f7-4dd9-b0ad-9e322e1cacd0","timestamp":1710811934439,"meterValue":{"voltage":450,"currentInput":126.67,"power":57,"meter":0.1583,"currentTemperature":97}}}

EMQX Rule SQL

During the charging process, the charging equipment periodically sends MeterValues data, providing metering data such as charging quantity, voltage, current, etc. EMQX will create a rule to handle the MeterValues data and write it into ClickHouse for analyzing electricity usage patterns. You can also open http://localhost:18083 in your browser to access the EMQX Dashboard, then navigate to the Integration → Rules page to modify this rule, utilizing EMQX's built-in SQL functions for custom processing.

SELECT
  payload.payload as record,
  record.meterValue as meterValue
FROM
  "mqttx/simulate/charge/+/MeterValues"

EMQX ClickHouse Data Integration

After processing the data through the rule, EMQX will use the rule action to write the meter values data in real-time to ClickHouse.

EMQX supports data integration with ClickHouse using SQL templates for data insertion, which can adapt well to complex data structures, enabling flexible data writing and business development.

EMQX will utilize the following SQL template to store each piece of charging data based on the charging gun and charging order:

INSERT INTO charging_record (
  connectorId,
  transactionId,
  timestamp,
  voltage,
  currentInput,
  power,
  meter,
  currentTemperature
) VALUES (
  '${record.connectorId}',
  '${record.transactionId}',
  toDateTime(${record.timestamp}/1000),
  ${meterValue.voltage},
  ${meterValue.currentInput},
  ${meterValue.power},
  ${meterValue.meter},
  ${meterValue.currentTemperature}
)

Stop Charging Message Processing

Topic

mqttx/simulate/charge/{clientId}/StopTransaction

Message Example

{"messageType":"Call","action":"StopTransaction","payload":{"lastChargeStart":null,"startPower":160,"power":152,"voltage":650,"startTimestamp":1710733474521,"endTimestamp":1710814724521,"timePercentage":1.000123076923077,"currentTimestamp":1710814734521,"pauseDuration":0,"isPaused":false,"counter":272,"meterStart":26987.776999999107,"meter":2634.7483999998626,"connectorId":"829f3fad-2cb0-4dc3-8dac-e703a9d74fe1","currentTemperature":94,"transactionId":"9f1c88c1-6bc5-4cf5-afdc-bd0bb63d861c","idTag":"No. 4","timestamp":1710814734521,"meterStop":29622.52539999897,"duration":81250,"reason":"SoftStop"}}

EMQX Rule SQL

EMQX will create a rule to process stop-charging message data, enabling the creation of orders in ClickHouse. You can also open http://localhost:18083 in your browser to access the EMQX Dashboard, then navigate to the Integration → Rules page to modify this rule, leveraging EMQX's built-in SQL functions for custom processing.

SELECT
  payload.payload as record
FROM
  "mqttx/simulate/charge/+/StopTransaction"

EMQX ClickHouse Data Integration

After processing the data through the rule, EMQX will use the rule action to write the orders to ClickHouse with the following template:

INSERT INTO charging_order (
  idTag, 
  connectorId,
  transactionId,
  startTimestamp,
  endTimestamp,
  duration,
  reservationId,
  stackLevel,
  meterStart,
  meterStop,
  meter,
  stopReason
) VALUES (
  '${record.idTag}', 
  '${record.connectorId}',
  '${record.transactionId}',
  toDateTime(${record.startTimestamp} / 1000),
  toDateTime(${record.endTimestamp} / 1000),
  ${record.duration},
  '',
  0,
  ${record.meterStart},
  ${record.meterStop},
  ${record.meter},
  '${record.reason}'
)

Subscribe to Data from EMQX

Docker Compose has included a subscriber to print all charging data. You can view the data with this command:

$ docker logs -f mqttx [3/20/2024] [4:05:01 AM] › topic: mqttx/simulate/charge/mqttx_0f8a625b_1/MeterValues payload: {"messageType":"Call","action":"MeterValues","payload":{"connectorId":"d04890fe-76ef-43de-a31b-4e6362bd872f","transactionId":"5a281373-6faa-4ab0-b6d5-56915716a528","timestamp":1710851484421,"meterValue":{"voltage":650,"currentInput":365.38,"power":237.5,"meter":0.6597,"currentTemperature":92}}}

You can subscribe and receive the data with any MQTT client, such as MQTTX CLI:

mqttx sub -t mqttx/simulate/charge/+/+

View Charging Station Data in Grafana

To view charging station data in the Grafana dashboard, please open http://localhost:3000/ in your browser and log in using the username admin and password public.

Once logged in successfully, click on the "Dashboards" page in the navigation bar of the homepage. Then, select the "ClickHouse - Charging Station" dashboard. This dashboard will display key metrics such as current orders at the charging station, revenue status, power consumption, statistics on disconnected charging guns, and hourly power consumption.

With these metrics, you can visually monitor the operational status of the charging station and make necessary adjustments and optimizations to improve its efficiency and profitability. Whether for real-time monitoring or viewing historical data, this dashboard will help you gain better insights into the operation of the charging station, providing strong support for your decision-making process.

Conclusion

The versatility and real-time capabilities of MQTT, combined with ClickHouse's data storage and analytics power, enable various industries to leverage IoT data for enhanced operational efficiency, cost savings, and data-driven decision-making. With the Data Integration capabilities of EMQX, organizations can benefit from a reliable, scalable, and feature-rich MQTT broker and easily achieve this powerful combination.

Related resources:

• MQTT with Kafka: Supercharging IoT Data Integration
• MQTT with TimescaleDB: An Efficient Solution for IoT Time-Series Data
• MQTT to MongoDB: A Beginner's Guide for IoT Data Integration
• MQTT and Redis: Creating a Real-Time Data Statistics Application for IoT
• Integrating MQTT Data into InfluxDB for a Time-Series IoT Application
• MQTT to MySQL: Building a Real-Time Data Monitoring Application for Oil Extraction

Talk to an Expert

Contact Us →