Data Storage Routes Overview

Data Storage Integration

Data storage routes enable automatic persistence of MQTT messages to databases and files. Store sensor data for analytics, create historical archives, aggregate logs, and build time-series datasets—all configured through LoT syntax.

Like a filing cabinet with an assistant. You can hand your assistant a form (STORE IN with models) and they file it automatically, or you can write specific filing instructions yourself (EVENTs with queries). Either way, your data ends up organized in the right drawer.

Supported Databases

PostgreSQL

Open-source relational database for structured data storage.

MySQL

Popular relational database with wide ecosystem support.

MariaDB

MySQL-compatible database with enhanced features.

SQL Server

Microsoft SQL Server for enterprise environments.

MongoDB

Document database for flexible JSON storage.

OpenSearch

Real-time analytics, search, and visualization.

CrateDB

Distributed SQL optimized for time-series data.

Snowflake

Cloud data warehouse for large-scale analytics.

File Storage

CSV and JSON file storage for simple persistence.

Database Comparison

Database	Type	Best For	Query Format
PostgreSQL	Relational	Structured data, complex queries	SQL
MySQL	Relational	Web applications, general purpose	SQL
MariaDB	Relational	MySQL compatibility with extras	SQL
SQL Server	Relational	Enterprise, Windows environments	SQL
MongoDB	Document	Flexible schemas, JSON data	CLEAN
OpenSearch	Search/Analytics	Full-text search, dashboards	CLEAN
CrateDB	Time-Series	High-volume sensor data	SQL
Snowflake	Cloud Warehouse	Large-scale analytics	SQL
File Storage	File	Simple logging, backups	CLEAN

SQL databases use standard SQL queries. CLEAN is a simplified document-style syntax used by non-relational databases and file storage — see Query Formats below for details.

How to Build a Data Storage Route

A data storage route needs two things: a connection to your database and at least one rule for what to store. You define the connection once, then add events that listen for MQTT messages and run queries — or use a model with STORE IN to save data automatically without writing queries at all.

Quick Example

This complete route connects to a PostgreSQL database and automatically stores every sensor reading that arrives on sensors/+/reading:

DEFINE ROUTE SensorDB WITH TYPE POSTGRESQL
    ADD SQL_CONFIG
        WITH SERVER "postgres.example.com"
        WITH PORT '5432'
        WITH DATABASE "iot_data"
        WITH USERNAME "iot_user"
        WITH PASSWORD "secure_password"
    ADD EVENT StoreSensorReading
        WITH SOURCE_TOPIC "sensors/+/reading"
        WITH DESTINATION_TOPIC "db/status"
        WITH QUERY "INSERT INTO readings (ts, sensor_id, value) VALUES (NOW(), '{sensor_id}', '{value.json}')"

That’s it — the broker handles the rest. The sections below break down each part in detail.

Step 1: Define the Route Connection

Every data storage route starts with a connection configuration. Define the route type and provide the connection details for your database:

DEFINE ROUTE SensorDB WITH TYPE POSTGRESQL
    ADD SQL_CONFIG
        WITH SERVER "postgres.example.com"
        WITH PORT '5432'
        WITH DATABASE "iot_data"
        WITH USERNAME "iot_user"
        WITH PASSWORD "secure_password"

Each database type has its own config block (SQL_CONFIG, MONGODB_CONFIG, OPENSEARCH_CONFIG, etc.). See the specific database page for exact parameters.

The examples on this page use hardcoded credentials for clarity. In production, use environment variables and encrypted secrets to keep sensitive values out of your LoT definitions:

DEFINE ROUTE SensorDB WITH TYPE POSTGRESQL
    ADD SQL_CONFIG
        WITH SERVER GET ENV "DB_HOST"
        WITH PORT GET ENV "DB_PORT"
        WITH DATABASE GET ENV "DB_NAME"
        WITH USERNAME GET ENV "DB_USER"
        WITH PASSWORD GET SECRET "DB_PASSWORD"

See Environment Variables and Secrets for setup instructions across Windows, Linux, Docker, and Kubernetes.

Step 2: Store Data

There are two ways to write data into your database. Choose the approach that fits your use case:

With Models (STORE IN)
With Events (INSERT)

The STORE IN approach ties a model directly to a database route. Every time an action publishes this model, the record is automatically inserted into the specified table — no EVENT or query needed.1. Define a model with STORE IN:

DEFINE MODEL ProductionCount
    ADD "f_event_time" WITH TIMESTAMP "UNIX"
    ADD STRING "machine_id"
    ADD STRING "product_type"
    ADD INT "count"
    STORE IN "ProductionDB"
        WITH TABLE "production_counts"

2. Define the route the model stores into:

DEFINE ROUTE ProductionDB WITH TYPE POSTGRESQL
    ADD SQL_CONFIG
        WITH SERVER "localhost"
        WITH PORT '5432'
        WITH DATABASE "factory"
        WITH USERNAME "postgres"
        WITH PASSWORD "secure_password"

3. Publish the model from an action:

DEFINE ACTION RecordProduction
ON TOPIC "factory/+/production" DO
    SET "machine" WITH TOPIC POSITION 2
    SET "product" WITH (GET JSON "product_type" IN PAYLOAD AS STRING)
    SET "qty" WITH (GET JSON "count" IN PAYLOAD AS INT)

    PUBLISH MODEL ProductionCount TO "factory/production/recorded" WITH
        machine_id = {machine}
        product_type = {product}
        count = {qty}

Every PUBLISH MODEL call automatically inserts a row into the production_counts table with all the model fields as columns.

The EVENT approach uses explicit SQL or CLEAN queries triggered by MQTT messages arriving on a source topic. You write the query yourself, giving full control over column mapping and transformations.Define an EVENT inside the route:

DEFINE ROUTE SensorDB WITH TYPE POSTGRESQL
    ADD SQL_CONFIG
        WITH SERVER "postgres.example.com"
        WITH PORT '5432'
        WITH DATABASE "iot_data"
        WITH USERNAME "iot_user"
        WITH PASSWORD "secure_password"
    ADD EVENT StoreSensorReading
        WITH SOURCE_TOPIC "sensors/+/reading"
        WITH DESTINATION_TOPIC "db/status"
        WITH QUERY "INSERT INTO readings (ts, sensor_id, value) VALUES (NOW(), '{sensor_id}', '{value.json}')"

When a message arrives on any topic matching sensors/+/reading, the INSERT query runs with placeholders replaced by actual values from the message.

Step 3: Read Data (SELECT)

EVENTs also support SELECT queries to read data back from the database. Publish a message to the event’s SOURCE_TOPIC with your query parameters, and the result is published to DESTINATION_TOPIC:

DEFINE ROUTE ProductionDB WITH TYPE POSTGRESQL
    ADD SQL_CONFIG
        WITH SERVER "localhost"
        WITH PORT '5432'
        WITH DATABASE "factory"
        WITH USERNAME "postgres"
        WITH PASSWORD "secure_password"
    ADD EVENT GetLatestCount
        WITH QUERY "SELECT MAX(count) as count FROM production_counts WHERE machine_id = '{payload}'"
        WITH SOURCE_TOPIC "factory/production/getLatestCount"
        WITH DESTINATION_TOPIC "factory/production/latest_count_from_db"

To trigger this query, publish the machine ID to the source topic:

Topic:   factory/production/getLatestCount
Payload: MACHINE-001

The query result is published to factory/production/latest_count_from_db, where your actions or external clients can consume it.

Common Query Placeholders

These placeholders work across all database routes:

Placeholder	Description
`{payload}`	Raw MQTT payload (useful in SELECT queries as a parameter)
`{value}`	Full MQTT payload as string
`{value.json}`	Parsed JSON payload
`{value.json.field}`	Specific field from JSON
`{timestamp}`	Message timestamp
`{source_topic}`	Original MQTT topic
`{field}`	Field extracted from topic path

Extracting Fields from Topics

For topic sensors/temp001/reading, use these placeholders:

Placeholder	Value
`{source_topic}`	`sensors/temp001/reading`
`{sensor_id}`	`temp001` (position 2 in topic)

Query Formats

SQL Format

For relational databases (PostgreSQL, MySQL, MariaDB, SQL Server, CrateDB, Snowflake):

WITH QUERY "INSERT INTO sensor_readings (timestamp, sensor_id, value) VALUES ('{timestamp}', '{sensor_id}', '{value.json}')"

CLEAN Format

For document databases and search engines (MongoDB, OpenSearch, File Storage):

WITH QUERY "CLEAN:{collection: readings, document: {sensor_id: {sensor_id}, value: {value.json}, timestamp: {timestamp}}}"

The CLEAN format provides simplified JSON syntax without requiring proper escaping.

Common EVENT Parameters

Trigger & Routing

SOURCE_TOPIC

string

required

MQTT topic pattern that triggers the query. For INSERT operations, incoming message data is used to populate placeholders. For SELECT operations, the payload can be used as a query parameter via {payload}.

DESTINATION_TOPIC

string

MQTT topic to publish query results. Required for SELECT queries so the result can be consumed by actions or external clients. Optional for INSERT queries (publishes operation status).

Query

QUERY

string

required

SQL statement or CLEAN format query with placeholders. Supports both write operations (INSERT INTO ...) and read operations (SELECT ... FROM ...).

Best Practices

Use SSL and Certificates in Production

Always enable SSL for database connections. Most database routes support SSL configuration through the config block:

WITH USE_SSL "true"
WITH TRUST_SERVER_CERTIFICATE "false"

When using certificate-based authentication, store certificate passphrases as encrypted secrets rather than in plain text. See Environment Variables and Secrets for details.

Secure Your Credentials

Never hardcode passwords or API keys in route definitions. Use GET SECRET for sensitive values and GET ENV for non-sensitive configuration like hostnames and ports. Create dedicated database users with minimal permissions for each route.See Environment Variables and Secrets for full setup and deployment instructions.

Monitor Database Size

Implement data retention policies:

Index lifecycle management
Time-based partitioning
Automatic cleanup jobs

Use STORE IN for Automatic Persistence

When your data has a consistent schema, use STORE IN on a model to automatically insert records every time the model is published. This eliminates the need to write INSERT queries manually:

DEFINE MODEL SensorReading
    ADD STRING "sensor_id"
    ADD DOUBLE "value"
    ADD STRING "timestamp" WITH TIMESTAMP "UTC"
    STORE IN "SensorDB"
        WITH TABLE "sensor_readings"

Use Connection Pooling

For high-throughput scenarios, configure connection pools where supported.

Next Steps

Choose the database that matches your requirements:

PostgreSQL

Start with the most popular open-source database.

MongoDB

Store flexible JSON documents.

Quick Start

Developing and Managing Coreflux

Coreflux MQTT Broker

LoT Language

Product Updates

Data Storage Integration

Supported Databases

PostgreSQL

MySQL

MariaDB

SQL Server

MongoDB

OpenSearch

CrateDB

Snowflake

File Storage

Database Comparison

How to Build a Data Storage Route

Quick Example

Step 1: Define the Route Connection

Step 2: Store Data

Step 3: Read Data (SELECT)

Common Query Placeholders

Extracting Fields from Topics

Query Formats

SQL Format

CLEAN Format

Common EVENT Parameters

Best Practices

Next Steps

PostgreSQL

MongoDB

​Data Storage Integration

​Supported Databases

PostgreSQL

MySQL

MariaDB

SQL Server

MongoDB

OpenSearch

CrateDB

Snowflake

File Storage

​Database Comparison

​How to Build a Data Storage Route

​Quick Example

​Step 1: Define the Route Connection

​Step 2: Store Data

​Step 3: Read Data (SELECT)

​Common Query Placeholders

​Extracting Fields from Topics

​Query Formats

​SQL Format

​CLEAN Format

​Common EVENT Parameters

​Best Practices

​Next Steps

PostgreSQL

MongoDB

Data Storage Integration

Supported Databases

Database Comparison

How to Build a Data Storage Route

Quick Example

Step 1: Define the Route Connection

Step 2: Store Data

Step 3: Read Data (SELECT)

Common Query Placeholders

Extracting Fields from Topics

Query Formats

SQL Format

CLEAN Format

Common EVENT Parameters

Best Practices

Next Steps