Sink data from RisingWave to MQTT

The Message Queuing Telemetry Transport (MQTT) protocol is ideal for connecting remote devices with a small code footprint and minimal network bandwidth. MQTT today is used in a wide variety of industries, such as automotive, manufacturing, telecommunications, oil and gas, etc.

​

Prerequisites

Before sinking data from RisingWave to an MQTT topic, please ensure the following:

• The RisingWave cluster is running.
• An MQTT broker is running and accessible from your RisingWave cluster.
• Create an MQTT topic that you want to sink data to.
• You have permission to publish data to the MQTT topic.

For example, we have an iot_sensor_data table in RisingWave that stores data from various IoT devices at a given timestamp, including temperature and humidity readings, along with a status field indicating whether the device is in a normal or abnormal state. For more information to learn about MQTT and get started with it, refer to the MQTT guide.

​

Syntax

To sink data from RisingWave to an MQTT topic, create a sink using the syntax below:

CREATE SINK [ IF NOT EXISTS ] sink_name
[FROM sink_from | AS select_query]
WITH (
   connector='mqtt',
   url = '<your MQTT server>:<port>',
   topic = '<topic>',
   qos =  '<qos_level>',
   type = '<append-only>'
   username = '<your user name>',
   password = '<your password>')
FORMAT PLAIN ENCODE data_encode -- Format options: plain (encode BYTES and JSON) (
    force_append_only='true',
);

This query sets up an MQTT sink mqtt_sink to forward data from iot_sensor_data to an MQTT server. It configures the MQTT connector, server URL, target topic, data type, message retention, quality of service, and JSON encoding.

CREATE SINK mqtt_sink
FROM iot_sensor_data
WITH
(
    connector='mqtt',
    url='tcp://mqtt-server',
    topic= 'sink_iot_data',
    type = 'append-only',
    retain = 'true',
    qos = 'at_least_once',
) FORMAT PLAIN ENCODE JSON (
    force_append_only='true',
);

After the sink is created, you will continuously consume the data in the MQTT topic from RisingWave in append-only mode.

​

Parameters