Real-Time IoT Device Monitoring Using RisingWave and HiveMQ

In the world of IoT (Internet of Things), where countless devices are interconnected, real-time monitoring is essential for ensuring smooth operations and prompt response to potential issues. The ability to detect anomalies and unusual behavior as they happen allows organizations to proactively address problems, maintain system health, and minimize downtime in IoT networks.

In this blog, we delve into the realm of IoT device monitoring using the dynamic duo of RisingWave and HiveMQ. Data from IoT devices are published to an MQTT broker on HiveMQ Cloud. We ingest the data into RisingWave and process it in real time to find anomalies. We’ll also visualize the data in RisingWave Cloud.

HiveMQ is a leading enterprise MQTT platform known for its reliability, scalability, and flexibility. It extends MQTT, the industry standard for IoT messaging, and offers a comprehensive solution for IoT deployments. HiveMQ is the preferred choice for renowned brands like Air France-KLM Group, BMW, Liberty Global, Mercedes-Benz, and ZF in various industries including automotive, energy, logistics, smart manufacturing, transportation, Industry 4.0, and connected IoT products. At its core, HiveMQ provides a high-performance MQTT-compliant broker, ensuring fast and reliable data transfer.

RisingWave is a streaming database that allows users to ingest, transform, and manage streaming data using Postgres-style SQL. Data in RisingWave is available to serve queries from applications and BI tools.

Setting up a cluster in HiveMQ Cloud

Sign up for an account and sign in to console.hivemq.cloud.

After creating your HiveMQ Cloud account, you can easily set up a free HiveMQ Cloud cluster. Within moments, your cluster will be up and running.

Once your HiveMQ Cloud cluster is created, you can start sending IoT device health-related data to the MQTT broker in HiveMQ Cloud. Each entry represents a data point from a specific IoT device at a given timestamp, including temperature and humidity readings, as well as a status field indicating whether the device is in a normal or abnormal state.

For more detailed instructions, please refer to the HiveMQ Quick Start Guide.

[
{'device_id': 'device_001', 'ts': '2024-05-03T14:41:24Z', 'temperature': 94, 'humidity': 68, 'device_status': 'normal'}
{'device_id': 'device_002', 'ts': '2024-05-03T14:41:27Z', 'temperature': 58, 'humidity': 90, 'device_status': 'normal'}
{'device_id': 'device_002', 'ts': '2024-05-03T14:41:26Z', 'temperature': 50, 'humidity': 5, 'device_status': 'abnormal'}
{'device_id': 'device_003', 'ts': '2024-05-03T14:41:31Z', 'temperature': 50, 'humidity': 39, 'device_status': 'normal'}
{'device_id': 'device_001', 'ts': '2024-05-03T14:41:26Z', 'temperature': 92, 'humidity': 18, 'device_status': 'normal'}
{'device_id': 'device_001', 'ts': '2024-05-03T14:41:29Z', 'temperature': 94, 'humidity': 92, 'device_status': 'normal'}
{'device_id': 'device_002', 'ts': '2024-05-03T14:41:29Z', 'temperature': 39, 'humidity': 25, 'device_status': 'normal'}
]

Ingest data from HiveMQ into RisingWave

For ingesting and processing streaming data, there are two options available: the open-source RisingWave and the managed service, RisingWave Cloud. In this blog, we will focus on using RisingWave Cloud, which provides a user-friendly experience and simplifies the operational aspects of managing and utilizing RisingWave for our IoT device monitoring solution.

RisingWave Cloud also offers built-in visualization features that enable us to generate tables and charts. These visualizations help us gain deeper insights from real-time analysis, allowing us to make more informed data-driven decisions.

Create a RisingWave cluster

To create a RisingWave cluster in RisingWave Cloud and explore the various features it offers, you can sign up for the free plan available. The free plan allows you to test the functionalities of RisingWave without any cost. For detailed instructions on how to create a RisingWave cluster and get started, you can refer to the official RisingWave documentation. It will provide you with step-by-step guidance to set up and explore the features of RisingWave. If you need additional help with setting up this integration, join our active Slack community.

Create a source in RisingWave to ingest data streams

This SQL query sets up an IoT data source called iot_sensor_data. It includes columns for device ID, timestamp, temperature, humidity, and device status. The query connects to an MQTT broker using specific settings for the MQTT broker, topic, and credentials. It reads data in real-time from the earliest available records, assuming the data is already in JSON format as sent by the MQTT clients. The purpose is to ingest the data into RisingWave. Since no transformation is applied in the HiveMQ Cloud, the data is in the required JSON format before being sent by the MQTT clients.

Please note that a source in RisingWave establishes the connection with the stream, but it does not persist the streaming data. If you need more details about the MQTT source, refer to RisingWave documentation.

CREATE SOURCE iot_sensor_data(
  device_id VARCHAR,
  ts TIMESTAMP,
  temperature INTEGER,
  humidity VARCHAR,
  device_status INTEGER
  )
WITH (
  connector='mqtt', 
  url = 'ssl://xxxxxxxxx:8883',
  topic = 'iot_topic',
  username = 'xxxxxx',
  password = 'xxxxxx',
  qos = 'at_least_once',
  scan.startup.mode = 'earliest'
) FORMAT PLAIN ENCODE JSON;

Analyze real-time data using RisingWave

In RisingWave, we create various materialized views to analyze IoT data related to the health of IoT devices. This analysis allows us to calculate the average, maximum and minimum temperature and humidity, as well as the device status, within specific time windows.

As a streaming database, RisingWave has leveraged materialized views in a unique way to power always-on analytics and data transformations for latency-sensitive applications such as alerting, monitoring and trading. They are automatically refreshed and incrementally computed whenever a new event is received.

Calculate the average temperature and humidity for each device in a time window

CREATE MATERIALIZED VIEW device_avg_mv AS
SELECT device_id, 
AVG(temperature) AS avg_temperature,
AVG(humidity) AS avg_humidity,
window_start, window_end
FROM TUMBLE (iot_sensor_data, ts, INTERVAL '1 MINUTES')
GROUP BY device_id,window_start, window_end;

Counts device status as either Normal or Abnormal within a time window

CREATE MATERIALIZED VIEW device_status_count_mv AS
SELECT  device_id,
COUNT(CASE WHEN device_status = 'normal' THEN 1 ELSE NULL END) AS normal_count,
COUNT(CASE WHEN device_status = 'abnormal' THEN 1 ELSE NULL END) AS abnormal_count, 
window_start, window_end
FROM TUMBLE (iot_sensor_data, ts, INTERVAL '1 MINUTES')
GROUP BY device_id, window_start, window_end;

Calculate the max and min temperature and humidity for each device in a time window

CREATE MATERIALIZED VIEW device_max_min_mv AS
SELECT device_id, 
MAX(temperature) AS max_temperature,
MAX(humidity) AS max_humidity,
MIN(temperature) AS min_temperature,
MIN(humidity) AS min_humidity,
window_start, window_end
FROM TUMBLE (iot_sensor_data, ts, INTERVAL '1 MINUTES')
GROUP BY device_id,window_start, window_end;

Visualization using RisingWave Cloud

RisingWave Cloud also provides powerful built-in tools for data visualization, making it easy for users to effectively visualize real-time data. With RisingWave, you can choose from various visualization formats such as tables, line charts, area charts, and bar charts, allowing you to derive valuable insights from your processed data.

Refreshing Results

To update your data views, you have the option to manually rerun the last executed query by clicking the designated button. Alternatively, you can set a timer for automatic refreshes, ensuring that your visualizations stay up to date.

Downloading Results

If you need to perform offline analysis or keep records, RisingWave offers a convenient feature that allows you to easily download your results as a CSV file. Simply click the corresponding button to initiate the download.

Comprehensive overview of IoT devices

The table below provides a detailed overview of IoT devices. It is generated using the iot_sensor_data table, which captures data from various IoT devices. The table includes device identifiers (device_id), timestamps (ts), temperature readings (temperature), humidity levels (humidity), and device statuses (device_status). This information gives you a comprehensive understanding of the data captured from the IoT devices.

Analysis of Device Status from IoT Devices

The chart displayed below, generated from the device_status_count_mv materialized view, provides minute-by-minute counts of normal and abnormal statuses for each device. This visualization helps identify patterns in device performance and potential issues. By analyzing this chart, you can gain valuable insights into the status of your IoT devices.

Analysis of Average Temperature and Humidity

The chart below is generated using the device_avg_mv view. It provides a visual representation of the average temperature and humidity measurements for different devices. These measurements are calculated every minute, allowing you to observe trends or variations in environmental conditions over time. This visualization helps you gain insights into the average temperature and humidity levels across your devices.

Analysis of Highest and Lowest Sensor Values

The chart below is generated using the device_max_min_mv view. It displays the highest and lowest temperature and humidity values recorded per minute across devices. This visualization helps identify environmental fluctuations and extremes in sensor readings. By analyzing this chart, you can gain insights into the range of temperature and humidity values observed across your devices.