# Ingesting MQTT messages into CrateDB using Node-RED

**URL:** <https://community.cratedb.com/t/ingesting-mqtt-messages-into-cratedb-using-node-red/803>\
**Category:** Tutorials\
**Tags:** integration, data-ingestion\
**Created:** [September 13, 2021, 2:10pm UTC](https://community.cratedb.com/t/ingesting-mqtt-messages-into-cratedb-using-node-red/803 "2021-09-13T14:10:36Z")\
**Posts on this page:** 1\
**Page:** 1

<div class="post-metadata">

**Author:** ![hammerhead](https://sea2.discourse-cdn.com/flex020/user_avatar/community.cratedb.com/hammerhead/32/270_2.png) [@hammerhead](https://community.cratedb.com/u/hammerhead)\
**Post date:** [September 13, 2021, 2:10pm UTC](https://community.cratedb.com/t/ingesting-mqtt-messages-into-cratedb-using-node-red/803/1 "2021-09-13T14:10:36Z")

</div>

[Node-RED](https://nodered.org/) is a workflow automation tool allowing to orchestrate message flows and transformations via a comfortable web interface.  
In this article, we will demonstrate the usage of Node-RED with CrateDB at the example of reading messages from an MQTT broker and inserting them into CrateDB.

# Prerequisites

To follow this article, you will need:

1. A running [Node-RED installation](https://nodered.org/#get-started)
2. An installed [node-red-contrib-postgresql](https://github.com/alexandrainst/node-red-contrib-postgresql) module
3. A running MQTT broker. We are using [HiveMQ Cloud](https://www.hivemq.com/) for this test setup.

# Producing data

First, we will generate data that populates the MQTT topic using Node-RED. If you already have an MQTT topic with regular messages, you can skip this part.

 ![Screenshot 2026-03-19 at 15.49.42](https://us1.discourse-cdn.com/flex020/uploads/crate/original/2X/9/9e423c7fd7b85313a23961c62afddb6db07afce8.png)

The inject node is creating a JSON payload with three different attributes:

 ![image](https://us1.discourse-cdn.com/flex020/uploads/crate/original/2X/e/e473a5b00aa697d696b69a9abdabc8d2b024732f.png)

For the sake of this example, two of them are static and only the timestamp will change.  
Here is the full workflow definition: [flows-producer.json](https://community.cratedb.com/uploads/short-url/4DNU2Oaz1RHH6nLhkfDHf1O6NWb.json) (2.0 KB)

# Consuming and ingesting data

First of all, we create the target table in CrateDB:

```sql
CREATE TABLE nodered_target (
   ts TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL,
   payload OBJECT(DYNAMIC)
);

```

The payload makes use of CrateDB’s [OBJECT data type](https://crate.io/docs/crate/reference/en/4.6/general/ddl/data-types.html#object) that is a perfect fit for payloads with a schema that will likely evolve over time. For productive usage, [partitioning and sharding](https://community.cratedb.com/t/sharding-and-partitioning-guide-for-time-series-data/737) should be considered as well.

To deal with high amounts of data in an efficient manner, messages retrieved from the MQTT topic are grouped into batches before ingestion. Batching messages into [multi-value INSERT statements](https://crate.io/docs/crate/howtos/en/latest/performance/inserts/methods.html#inserts-multiple-values) will reduce the overhead of generating single INSERT statements for each MQTT message and is set up as follows:

 ![image](https://us1.discourse-cdn.com/flex020/uploads/crate/original/2X/2/2894f29958a74571f692062e9d398092e377216c.png)

1. HiveMQ Cloud: Consumes messages from the configured MQTT topic

2. join: Merges a given number of messages into one array. The length of the array equals the number of rows that will later be inserted into CrateDB in one multi-value statement.

3. function: This node receives a message with an array of payloads from the join node and performs a `reduce` operation. That operation transforms the array into a list of values represented as a string (`(p1), (p2), ...`). The aim is to produce a string that is already in a proper SQL-compatible syntax for usage as part of an INSERT query.

4. postgresql: As a final step, the postgresql node contains the connection information to the CrateDB cluster and the INSERT SQL query. The values are interpolated from the payload provided by the function node.

Here is the full workflow definition: [flows-consumer.json](https://community.cratedb.com/uploads/short-url/w90X07WgNyuZg1NInQd1cga4TGS.json) (2.3 KB)

# Testing

To test the workflow, you can click on the rectangle left to the timestamp node ( ![image](https://us1.discourse-cdn.com/flex020/uploads/crate/original/2X/b/b170c531bd1ab35b3b4ce7fca7271d6a1e72f3b2.png)) to inject a message. In the provided configuration, an INSERT statement will be triggered after two messages have been injected or ten seconds after the first message, if no second message follows.

Run a SELECT statement on your CrateDB cluster to see the inserted rows:

```sql
SELECT *
FROM nodered_target;

```

![Screenshot 2021-09-13 at 16.05.33](https://sea2.discourse-cdn.com/flex020/images/transparent.png)