File Storage Connector

File Storage Overview

The FILE_STORAGE connector 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 CONNECTOR DataLogger WITH TYPE FILE_STORAGE
    ADD FILE_STORAGE_CONFIG
        WITH STORAGE_DIR "/data/logs"
        WITH FORMAT "CSV"
        WITH FILE_PATTERN "{0}.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

string

required

Directory path for storing files.

Storage Location Options

string

File name pattern. {0} is replaced with the table name. Default: {0}.csv for CSV and {0}.json for JSON.{0} is the only placeholder. Others, such as {date}, are not supported: the connector fails when it writes. For daily files, publish to a table name that includes the date.

Format

string

File format: CSV or JSON. Default: JSON.

string

CSV delimiter character. Default: comma (,).

Write Options

boolean

Append to existing files. Default: true.

boolean

Auto-add timestamp field. Default: false.

integer

JSON only: the most records a file keeps. Older records are dropped once the file is full. Default: 10000. Files are not rotated.

Timestamp

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 connector definitions:

WITH STORAGE_DIR GET ENV "STORAGE_PATH"

See Environment Variables & Secrets for setup and usage.

CLEAN Query Format

File storage connectors 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 CONNECTOR CSVLogger WITH TYPE FILE_STORAGE
    ADD FILE_STORAGE_CONFIG
        WITH STORAGE_DIR "/data/logs"
        WITH FORMAT "CSV"
        WITH FILE_PATTERN "{0}.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 CONNECTOR 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}}"

Connector to different files by data type:

DEFINE CONNECTOR MultiFileLogger WITH TYPE FILE_STORAGE
    ADD FILE_STORAGE_CONFIG
        WITH STORAGE_DIR "/data/logs"
        WITH FORMAT "CSV"
        WITH FILE_PATTERN "{0}.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 a rolling backup file that keeps the newest records:

DEFINE CONNECTOR BackupStorage WITH TYPE FILE_STORAGE
    ADD FILE_STORAGE_CONFIG
        WITH STORAGE_DIR "/backup/mqtt"
        WITH FORMAT "JSON"
        WITH FILE_PATTERN "{0}.json"
        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 connector. 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 CONNECTOR DataLogger WITH TYPE FILE_STORAGE
    ADD FILE_STORAGE_CONFIG
        WITH STORAGE_DIR "/data/logs"
        WITH FORMAT "CSV"
        WITH FILE_PATTERN "{0}.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`

{0} is the only supported placeholder. A pattern with others, such as {date}, fails when the connector writes.

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

Connector fails to write

FILE_PATTERN accepts only {0} (the table name). Remove other placeholders such as {date}.
For JSON, MAX_RECORDS caps the file and drops the oldest records. Files are not rotated.

Quick Start

Coreflux MQTT Broker

LoT Language

Coreflux HUB

VS Code - LoT Notebooks

Coreflux Portal

Using AI with Coreflux

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 Connectors Overview

PostgreSQL Connector

​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 Connectors Overview

PostgreSQL Connector

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