File Storage Route

File Storage Overview

The FILE_STORAGE route writes MQTT messages to local CSV or JSON files. It provides simple persistence for logging, backups, and data export without requiring a database server.

File storage is ideal for simple logging, local backups, and exporting data for offline analysis. Use it when you don’t need a full database but want persistent records.

Basic Syntax

DEFINE ROUTE DataLogger WITH TYPE FILE_STORAGE
    ADD FILE_STORAGE_CONFIG
        WITH STORAGE_DIR "/data/logs"
        WITH FORMAT "CSV"
        WITH FILE_PATTERN "{0}_{date}.csv"
        WITH APPEND "true"
    ADD EVENT LogSensorData
        WITH SOURCE_TOPIC "sensors/#"
        WITH QUERY "CLEAN:{table: sensor_data, data: {value.json}}"

Connection Configuration

FILE_STORAGE_CONFIG Parameters

STORAGE_DIR

string

required

Directory path for storing files.

Storage Location Options

FILE_PATTERN

string

File name pattern. Use {0} for table name, {date} for current date. Default: {0}.csv

Format

FORMAT

string

File format: CSV or JSON. Default: CSV.

CSV_DELIMITER

string

CSV delimiter character. Default: comma (,).

Write Options

APPEND

boolean

Append to existing files. Default: false (overwrite).

INCLUDE_TIMESTAMP

boolean

Auto-add timestamp field. Default: false.

MAX_RECORDS

integer

Maximum records per file before rotation. Default: 10000.

Timestamp

DATE_FORMAT

string

Timestamp format string. Default: yyyy-MM-dd HH:mm:ss.

Use environment variables for paths in production. Keep deployment-specific values out of your route definitions:

WITH STORAGE_DIR GET ENV "STORAGE_PATH"

See Environment Variables & Secrets for setup and usage.

CLEAN Query Format

File storage routes use the CLEAN format:

WITH QUERY "CLEAN:{table: <name>, data: {value.json}}"

The table name is used in the file pattern with {0}.

Writing Data

CSV Logging
JSON Storage
Multiple Files
Backup Storage

Log sensor data to daily CSV files:

DEFINE ROUTE CSVLogger WITH TYPE FILE_STORAGE
    ADD FILE_STORAGE_CONFIG
        WITH STORAGE_DIR "/data/logs"
        WITH FORMAT "CSV"
        WITH FILE_PATTERN "{0}_{date}.csv"
        WITH APPEND "true"
        WITH CSV_DELIMITER ","
        WITH INCLUDE_TIMESTAMP "true"
        WITH MAX_RECORDS "10000"
    ADD EVENT LogSensorData
        WITH SOURCE_TOPIC "sensors/#"
        WITH DESTINATION_TOPIC "file/status"
        WITH QUERY "CLEAN:{table: sensor_data, data: {value.json}}"

Creates files like: sensor_data_2025-01-15.csv

Store events as JSON lines:

DEFINE ROUTE JSONLogger WITH TYPE FILE_STORAGE
    ADD FILE_STORAGE_CONFIG
        WITH STORAGE_DIR "/data/json"
        WITH FORMAT "JSON"
        WITH FILE_PATTERN "{0}.json"
        WITH APPEND "true"
        WITH INCLUDE_TIMESTAMP "true"
        WITH MAX_RECORDS "5000"
    ADD EVENT LogEvents
        WITH SOURCE_TOPIC "events/#"
        WITH DESTINATION_TOPIC "file/json/status"
        WITH QUERY "CLEAN:{table: events, data: {value.json}}"

Route to different files by data type:

DEFINE ROUTE MultiFileLogger WITH TYPE FILE_STORAGE
    ADD FILE_STORAGE_CONFIG
        WITH STORAGE_DIR "/data/logs"
        WITH FORMAT "CSV"
        WITH FILE_PATTERN "{0}_{date}.csv"
        WITH APPEND "true"
        WITH INCLUDE_TIMESTAMP "true"
    
    ADD EVENT LogTemperature
        WITH SOURCE_TOPIC "sensors/+/temperature"
        WITH QUERY "CLEAN:{table: temperature, data: {value.json}}"
    
    ADD EVENT LogPressure
        WITH SOURCE_TOPIC "sensors/+/pressure"
        WITH QUERY "CLEAN:{table: pressure, data: {value.json}}"
    
    ADD EVENT LogAlerts
        WITH SOURCE_TOPIC "alerts/#"
        WITH QUERY "CLEAN:{table: alerts, data: {value.json}}"

Creates: temperature_2025-01-15.csv, pressure_2025-01-15.csv, alerts_2025-01-15.csv

Create backup files with rotation:

DEFINE ROUTE BackupStorage WITH TYPE FILE_STORAGE
    ADD FILE_STORAGE_CONFIG
        WITH STORAGE_DIR "/backup/mqtt"
        WITH FORMAT "JSON"
        WITH FILE_PATTERN "{0}_{date}.jsonl"
        WITH APPEND "true"
        WITH MAX_RECORDS "100000"
        WITH INCLUDE_TIMESTAMP "true"
    ADD EVENT BackupAll
        WITH SOURCE_TOPIC "#"
        WITH QUERY "CLEAN:{table: mqtt_backup, data: {value.json}}"

Alternative: STORE IN with Models — Instead of writing EVENT queries, you can bind a model directly to this route. Every PUBLISH MODEL call automatically inserts a row — no query needed:

DEFINE MODEL SensorReading
    ADD STRING "sensor_id"
    ADD DOUBLE "value"
    STORE IN "DataLogger"
        WITH TABLE "sensor_data"

See Data Storage Overview for the full STORE IN workflow.

Reading Data

EVENTs also support read queries to retrieve data from stored files. Publish a message to the event’s SOURCE_TOPIC, and the query result is published to DESTINATION_TOPIC. File Storage read queries use the CLEAN format with filter, sort, and limit instead of SQL:

DEFINE ROUTE DataLogger WITH TYPE FILE_STORAGE
    ADD FILE_STORAGE_CONFIG
        WITH STORAGE_DIR "/data/logs"
        WITH FORMAT "CSV"
        WITH FILE_PATTERN "{0}_{date}.csv"
        WITH APPEND "true"
    ADD EVENT GetLatestEntry
        WITH SOURCE_TOPIC "file/query/latest"
        WITH DESTINATION_TOPIC "file/result/latest"
        WITH QUERY "CLEAN:{table: sensor_data, filter: {\"sensor_id\": \"{payload}\"}, sort: {\"timestamp\": -1}, limit: 1}"

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

Topic:   file/query/latest
Payload: temp001

The matching record is published to file/result/latest, where your actions or external clients can consume it.

CLEAN Read Operators

SQL Equivalent	File Storage CLEAN Syntax
`SELECT *`	`{"table": "x"}`
`WHERE a = 1`	`filter: {"a": 1}`
`WHERE a > 10`	`filter: {"a": {"$gt": 10}}`
`ORDER BY a DESC`	`sort: {"a": -1}`
`LIMIT 10`	`limit: 10`
`OFFSET 5`	`skip: 5`
`SELECT a, b`	`projection: {"a": 1, "b": 1}`

File Pattern Placeholders

Placeholder	Description	Example
`{0}`	Table name from CLEAN query	`sensor_data`
`{date}`	Current date (yyyy-MM-dd)	`2025-01-15`

Output Examples

CSV Output

timestamp,sensor_id,temperature,humidity
2025-01-15 10:30:00,temp001,23.5,65
2025-01-15 10:30:05,temp002,24.1,62

JSON Output (JSON Lines format)

{"timestamp":"2025-01-15T10:30:00Z","sensor_id":"temp001","temperature":23.5,"humidity":65}
{"timestamp":"2025-01-15T10:30:05Z","sensor_id":"temp002","temperature":24.1,"humidity":62}

Troubleshooting

Permission Denied

Verify STORAGE_DIR exists
Check write permissions on directory
Ensure Coreflux process has access

File Not Created

Verify STORAGE_DIR path is correct
Check disk space is available
Ensure directory is writable

Data Not Appearing

Verify SOURCE_TOPIC matches incoming messages
Check QUERY syntax is correct
Verify APPEND is set correctly

File Rotation Issues

Check MAX_RECORDS is set appropriately
Verify FILE_PATTERN includes {date} for daily rotation

Next Steps

Data Storage Routes Overview

Compare all storage options.

PostgreSQL Route

Configure PostgreSQL database storage.

Quick Start

Developing and Managing Coreflux

Coreflux MQTT Broker

LoT Language

Product Updates

File Storage Overview

Basic Syntax

Connection Configuration

FILE_STORAGE_CONFIG Parameters

CLEAN Query Format

Writing Data

Reading Data

CLEAN Read Operators

File Pattern Placeholders

Output Examples

CSV Output

JSON Output (JSON Lines format)

Troubleshooting

Next Steps

Data Storage Routes Overview

PostgreSQL Route

​File Storage Overview

​Basic Syntax

​Connection Configuration

​FILE_STORAGE_CONFIG Parameters

​CLEAN Query Format

​Writing Data

​Reading Data

​CLEAN Read Operators

​File Pattern Placeholders

​Output Examples

​CSV Output

​JSON Output (JSON Lines format)

​Troubleshooting

​Next Steps

Data Storage Routes Overview

PostgreSQL Route

File Storage Overview

Basic Syntax

Connection Configuration

FILE_STORAGE_CONFIG Parameters

CLEAN Query Format

Writing Data

Reading Data

CLEAN Read Operators

File Pattern Placeholders

Output Examples

CSV Output

JSON Output (JSON Lines format)

Troubleshooting

Next Steps