IoT Data Collection and Storage
IoT devices generate continuous streams of time-stamped data. Choosing the right storage strategy depends on your query patterns, retention requirements, and scale. This topic covers the full spectrum — from a simple CSV file on a Raspberry Pi to a production InfluxDB cluster.
Time-Series Data Characteristics
IoT sensor data is fundamentally different from typical application data:
python
# Typical application record — rarely changes
user = {
"id": 42,
"name": "Alice",
"email": "alice@example.com"
}
# IoT time-series record — one per measurement, always new
reading = {
"timestamp": 1704067200, # Every record has a time
"device_id": "sensor-01", # Tags (low-cardinality metadata)
"location": "Room A", # Tag
"temperature": 24.5, # Field (measured value)
"humidity": 61.2, # Field
}
# Characteristics:
# ✓ Append-only (old data never updated)
# ✓ High write throughput (millions of points/second at scale)
# ✓ Time-range queries are most common
# ✓ Old data can be compressed or downsampled (1s → 1min → 1hr averages)
# ✓ High data volume (1 sensor × 1 reading/min × 1 year = 525,600 rows)CSV Files — Simple Local Logging
Perfect for a Raspberry Pi with an SD card, or for prototyping before a database.
python
# Python — robust CSV logger with rotation
import csv
import os
import time
from datetime import datetime, date
class CSVLogger:
def __init__(self, directory="./data", max_rows_per_file=86400):
"""
max_rows_per_file: 86400 = one reading/sec for 24 hours.
Creates a new file each day.
"""
self.directory = directory
self.max_rows = max_rows_per_file
self._row_count = 0
self._file = None
self._writer = None
os.makedirs(directory, exist_ok=True)
self._open_file()
def _get_filename(self):
today = date.today().isoformat() # e.g. "2024-01-15"
return os.path.join(self.directory, f"sensors_{today}.csv")
def _open_file(self):
if self._file:
self._file.close()
filename = self._get_filename()
is_new = not os.path.exists(filename)
self._file = open(filename, 'a', newline='')
self._writer = csv.writer(self._file)
if is_new:
self._writer.writerow(
["timestamp", "device_id", "temperature", "humidity", "battery"]
)
self._row_count = 0
print(f"Logging to: {filename}")
def log(self, device_id, temperature, humidity, battery=None):
ts = datetime.utcnow().isoformat()
self._writer.writerow([ts, device_id, temperature, humidity, battery])
self._file.flush() # Ensure data survives power loss
self._row_count += 1
# Rotate file daily or when it gets too large
current_file = self._get_filename()
if (self._row_count >= self.max_rows or
not self._file.name.endswith(current_file.split('/')[-1])):
self._open_file()
def close(self):
if self._file:
self._file.close()
# Usage
logger = CSVLogger(directory="/home/pi/sensor_data")
while True:
logger.log("sensor-01", temperature=24.5, humidity=61.2, battery=87)
time.sleep(60)SQLite — Structured Local Storage
SQLite is built into Python and perfect for moderate data volumes (< 1M rows) on a local device.
python
import sqlite3
import time
from datetime import datetime, timedelta
class SensorDatabase:
def __init__(self, db_path="sensors.db"):
self.db_path = db_path
self._setup_db()
def _setup_db(self):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS readings (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
device_id TEXT NOT NULL,
temperature REAL,
humidity REAL,
battery INTEGER,
pressure REAL
)
""")
# Index on timestamp + device_id for fast queries
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_device_time
ON readings (device_id, timestamp DESC)
""")
def insert(self, device_id, temperature, humidity, battery=None, pressure=None):
ts = datetime.utcnow().isoformat()
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
INSERT INTO readings (timestamp, device_id, temperature, humidity, battery, pressure)
VALUES (?, ?, ?, ?, ?, ?)
""", (ts, device_id, temperature, humidity, battery, pressure))
return ts
def get_latest(self, device_id):
with sqlite3.connect(self.db_path) as conn:
row = conn.execute("""
SELECT timestamp, temperature, humidity, battery
FROM readings
WHERE device_id = ?
ORDER BY timestamp DESC
LIMIT 1
""", (device_id,)).fetchone()
return row
def get_average(self, device_id, hours=24):
since = (datetime.utcnow() - timedelta(hours=hours)).isoformat()
with sqlite3.connect(self.db_path) as conn:
row = conn.execute("""
SELECT
AVG(temperature) as avg_temp,
MIN(temperature) as min_temp,
MAX(temperature) as max_temp,
AVG(humidity) as avg_hum,
COUNT(*) as count
FROM readings
WHERE device_id = ? AND timestamp >= ?
""", (device_id, since)).fetchone()
return row
def purge_old(self, days=30):
"""Delete records older than N days."""
cutoff = (datetime.utcnow() - timedelta(days=days)).isoformat()
with sqlite3.connect(self.db_path) as conn:
deleted = conn.execute(
"DELETE FROM readings WHERE timestamp < ?", (cutoff,)
).rowcount
print(f"Purged {deleted} records older than {days} days")
return deleted
# Usage
db = SensorDatabase()
db.insert("sensor-01", 24.5, 61.2, battery=87, pressure=1013.2)
db.insert("sensor-01", 24.7, 60.8, battery=87)
db.insert("sensor-02", 22.1, 55.0, battery=92)
latest = db.get_latest("sensor-01")
print(f"Latest: {latest}")
avg = db.get_average("sensor-01", hours=24)
print(f"24h avg: T={avg[0]:.2f}°C H={avg[3]:.2f}% ({avg[4]} readings)")InfluxDB — Purpose-Built Time-Series Database
InfluxDB is the standard choice for production IoT time-series data. It handles millions of data points and natively understands time-series concepts.
Key Concepts
InfluxDB data model:
Bucket = database (with retention policy)
Measurement = table name (e.g. "environment")
Tag = indexed metadata (device_id, location) — strings only
Field = actual measurements (temperature, humidity) — numeric or string
Timestamp = nanosecond precision Unix time
Line Protocol (how data is written):
measurement,tag1=val1,tag2=val2 field1=val1,field2=val2 timestamp
environment,device_id=sensor-01,room=A temperature=24.5,humidity=61.2 1704067200000000000
^ ^^^^^^^^^^^^^^^^^ ^^^^^^^^^^^^^^^^^^^^^^^^ ^^^^^^^^^^^^^^^^^^^^^^^^^^^
measurement tags (indexed) fields (values) timestamp (nanoseconds)python
# Python → InfluxDB 2.x (pip install influxdb-client)
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS
from datetime import datetime, timezone
import time
INFLUX_URL = "http://localhost:8086"
INFLUX_TOKEN = "your-api-token-here"
ORG = "bigxstar"
BUCKET = "iot_sensors"
class InfluxLogger:
def __init__(self):
self._client = InfluxDBClient(
url=INFLUX_URL, token=INFLUX_TOKEN, org=ORG
)
self._write = self._client.write_api(write_options=SYNCHRONOUS)
self._query = self._client.query_api()
def write(self, device_id, location, temperature, humidity,
battery=None, pressure=None):
"""Write a single sensor reading."""
point = (
Point("environment")
.tag("device_id", device_id)
.tag("location", location)
.field("temperature", float(temperature))
.field("humidity", float(humidity))
)
if battery is not None: point.field("battery", int(battery))
if pressure is not None: point.field("pressure", float(pressure))
point.time(datetime.now(tz=timezone.utc), WritePrecision.SECONDS)
self._write.write(bucket=BUCKET, record=point)
def write_batch(self, records):
"""Write multiple records efficiently."""
points = []
for r in records:
p = (Point("environment")
.tag("device_id", r["device_id"])
.tag("location", r.get("location", "unknown"))
.field("temperature", float(r["temperature"]))
.field("humidity", float(r["humidity"])))
points.append(p)
self._write.write(bucket=BUCKET, record=points)
def query_average(self, device_id, hours=1):
"""Get average readings for the last N hours."""
query = f'''
from(bucket: "{BUCKET}")
|> range(start: -{hours}h)
|> filter(fn: (r) => r._measurement == "environment")
|> filter(fn: (r) => r.device_id == "{device_id}")
|> filter(fn: (r) => r._field == "temperature" or r._field == "humidity")
|> mean()
'''
tables = self._query.query(query, org=ORG)
results = {}
for table in tables:
for record in table.records:
results[record.get_field()] = round(record.get_value(), 2)
return results
def query_latest(self, device_id):
"""Get the most recent reading for a device."""
query = f'''
from(bucket: "{BUCKET}")
|> range(start: -1h)
|> filter(fn: (r) => r.device_id == "{device_id}")
|> last()
|> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
'''
tables = self._query.query(query, org=ORG)
for table in tables:
for record in table.records:
return record.values
return None
def close(self):
self._client.close()
# Usage
db = InfluxLogger()
# Write single point
db.write("sensor-01", "building-a", 24.5, 61.2, battery=87)
# Write batch (more efficient for high-frequency data)
batch = [
{"device_id": "sensor-01", "location": "room-a",
"temperature": 24.5, "humidity": 61.2},
{"device_id": "sensor-02", "location": "room-b",
"temperature": 22.1, "humidity": 55.0},
]
db.write_batch(batch)
avg = db.query_average("sensor-01", hours=24)
print(f"24h avg: {avg}") # {'temperature': 24.2, 'humidity': 60.8}
db.close()Data Downsampling (Retention Policies)
Raw data strategy:
0–7 days: 1 reading per minute (raw)
7–30 days: 5-minute averages (downsampled)
30d–1year: 1-hour averages
1year+: Delete or archive to S3
InfluxDB Flux query for downsampling:// Aggregate raw data to 5-minute averages
from(bucket: "iot_sensors")
|> range(start: -7d)
|> filter(fn: (r) => r._measurement == "environment")
|> aggregateWindow(every: 5m, fn: mean, createEmpty: false)
|> to(bucket: "iot_sensors_5min")Summary
- CSV files are perfect for local logging on a Raspberry Pi — simple, human-readable, no setup
- SQLite provides SQL queries for moderate data volumes and is built into Python
- InfluxDB is purpose-built for time-series data — handles millions of points and has built-in downsampling
- The line protocol format (
measurement,tags fields timestamp) is InfluxDB's write format - Tags are indexed metadata (device_id, location); fields are the measured values
- Always implement data retention — purge or downsample old data to keep storage manageable
- Batch writes are 10–100× more efficient than writing one point at a time
Next, we'll build dashboards to visualize this data with Grafana and Node-RED.