时序数据库应用：高效IoT数据存储与查询¶
学习目标¶
完成本文章后，你将能够：
- 理解时序数据库的核心概念和应用场景
- 掌握时序数据的特点和存储挑战
- 了解主流时序数据库（InfluxDB、TimescaleDB）
- 掌握时序数据压缩技术
- 理解时序数据查询优化策略
- 了解降采样和数据聚合技术
- 掌握时序数据库在嵌入式系统中的应用
- 理解时序数据的可视化方法
前置要求¶
在开始学习之前，建议你具备：
知识要求： - 熟悉基本的数据库概念 - 了解SQL查询语言 - 理解数据结构和算法 - 掌握数据压缩基础 - 了解IoT系统架构
技能要求： - 能够使用SQLite或其他数据库 - 熟悉数据采集和处理 - 了解网络通信协议 - 能够进行性能分析 - 掌握数据可视化基础
时序数据库概述¶
什么是时序数据库¶
时序数据库（Time Series Database, TSDB）是专门用于存储和查询时间序列数据的数据库系统。时间序列数据是按时间顺序排列的数据点序列，每个数据点包含时间戳和对应的测量值。
典型应用场景：
- IoT设备监控
- 传感器数据采集（温度、湿度、压力）
- 设备状态监控
- 能耗数据记录
-
环境参数追踪
-
工业自动化
- 生产线数据采集
- 设备运行参数监控
- 质量控制数据
-
预测性维护
-
智能家居
- 环境数据监测
- 能源消耗统计
- 设备使用记录
-
用户行为分析
-
车联网
- 车辆位置追踪
- 行驶数据记录
- 车况监控
- 驾驶行为分析
时序数据的特点¶
核心特征：
- 时间戳为主键
- 每条记录必须包含时间戳
- 数据按时间顺序写入
- 查询通常基于时间范围
-
时间是最重要的索引维度
-
写多读少
- 数据持续高频写入
- 读取操作相对较少
- 写入性能至关重要
-
批量写入优化
-
数据不可变
- 历史数据通常不修改
- 只有追加操作
- 简化并发控制
-
优化存储结构
-
数据量大
- 持续产生海量数据
- 需要高效压缩
- 存储成本敏感
- 需要数据保留策略
时序数据库架构¶
典型架构图：
┌─────────────────────────────────────────────────┐
│ 数据采集层 │
│ 传感器、设备、应用程序 │
└─────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────┐
│ 数据写入层 │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 批量缓冲 │ │ 数据验证 │ │
│ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────┐
│ 存储引擎 │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 时间分区 │ │ 数据压缩 │ │
│ └──────────────┘ └──────────────┘ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 索引管理 │ │ 数据保留 │ │
│ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────┐
│ 查询引擎 │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 时间范围查询 │ │ 聚合计算 │ │
│ └──────────────┘ └──────────────┘ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 降采样 │ │ 查询优化 │ │
│ └──────────────┘ └──────────────┘ │
└─────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────┐
│ 应用层 │
│ 数据可视化、告警、分析 │
└─────────────────────────────────────────────────┘
时序数据库 vs 传统数据库¶
| 特性 | 时序数据库 | 传统关系型数据库 |
|---|---|---|
| 数据模型 | 时间序列 | 表、行、列 |
| 主要操作 | 追加写入 | CRUD全操作 |
| 查询模式 | 时间范围查询 | 任意条件查询 |
| 数据压缩 | 高度优化 | 通用压缩 |
| 写入性能 | 极高 | 中等 |
| 存储效率 | 10-100倍压缩 | 较低压缩 |
| 聚合查询 | 高度优化 | 通用实现 |
| 数据保留 | 自动策略 | 手动管理 |
| 适用场景 | 监控、IoT | 业务数据 |
主流时序数据库介绍¶
InfluxDB¶
特点： - 开源时序数据库 - 专为时序数据设计 - 高性能写入和查询 - 内置HTTP API - 支持SQL-like查询语言（InfluxQL） - 自动数据保留策略 - 连续查询功能
数据模型：
measurement（测量）
├── tags（标签，索引）
│ ├── sensor_id
│ ├── location
│ └── device_type
├── fields（字段，实际数据）
│ ├── temperature
│ ├── humidity
│ └── pressure
└── timestamp（时间戳）
示例数据：
measurement: sensor_data
tags: sensor_id=sensor01, location=room1
fields: temperature=23.5, humidity=65.2
timestamp: 2024-01-15T10:30:00Z
优势： - 写入性能优秀（每秒数十万点） - 查询速度快 - 内存占用合理 - 易于部署和使用 - 丰富的生态系统
局限： - 集群版本为商业版 - 内存消耗较大 - 不适合嵌入式系统直接使用
TimescaleDB¶
特点： - 基于PostgreSQL的时序数据库扩展 - 完全兼容SQL - 支持关系型和时序数据 - 自动分区（hypertables） - 压缩和数据保留策略 - 支持连续聚合
数据模型：
CREATE TABLE sensor_data (
time TIMESTAMPTZ NOT NULL,
sensor_id TEXT NOT NULL,
temperature DOUBLE PRECISION,
humidity DOUBLE PRECISION,
PRIMARY KEY (time, sensor_id)
);
-- 转换为hypertable
SELECT create_hypertable('sensor_data', 'time');
优势： - 标准SQL支持 - 强大的查询能力 - 支持复杂关联查询 - 成熟的PostgreSQL生态 - 开源且功能完整
局限： - 资源占用较大 - 不适合嵌入式系统 - 配置相对复杂
嵌入式时序数据库方案¶
对于资源受限的嵌入式系统，可以使用轻量级方案：
1. SQLite + 时序优化
-- 创建时序数据表
CREATE TABLE sensor_readings (
timestamp INTEGER NOT NULL,
sensor_id TEXT NOT NULL,
value REAL NOT NULL,
PRIMARY KEY (timestamp, sensor_id)
);
-- 创建时间索引
CREATE INDEX idx_timestamp ON sensor_readings(timestamp);
-- 创建传感器索引
CREATE INDEX idx_sensor ON sensor_readings(sensor_id);
优势： - 轻量级，适合嵌入式 - 无需额外依赖 - 标准SQL支持 - 可靠性高
2. 自定义轻量级TSDB
基于循环缓冲区的简单实现：
// 时序数据点
typedef struct {
uint32_t timestamp;
float value;
} tsdb_point_t;
// 时序数据序列
typedef struct {
char name[32];
tsdb_point_t *points;
uint32_t capacity;
uint32_t head;
uint32_t count;
} tsdb_series_t;
优势： - 极低资源占用 - 可定制化 - 实时性好 - 适合简单场景
时序数据压缩技术¶
为什么需要压缩¶
时序数据具有以下特点，使其非常适合压缩：
- 时间戳规律性
- 采样间隔通常固定
- 可以使用差分编码
-
压缩率可达90%以上
-
数值相关性
- 相邻数据点变化小
- 可以使用增量编码
-
适合预测编码
-
数据冗余
- 重复模式多
- 可以使用字典编码
- 适合游程编码
常用压缩算法¶
1. Delta编码（差分编码）
存储相邻值的差值而不是绝对值：
实现示例：
// Delta编码
void delta_encode(int32_t *data, int32_t *encoded, size_t count) {
encoded[0] = data[0]; // 第一个值不变
for (size_t i = 1; i < count; i++) {
encoded[i] = data[i] - data[i-1]; // 存储差值
}
}
// Delta解码
void delta_decode(int32_t *encoded, int32_t *data, size_t count) {
data[0] = encoded[0];
for (size_t i = 1; i < count; i++) {
data[i] = data[i-1] + encoded[i]; // 累加差值
}
}
2. Delta-of-Delta编码
对差值再次进行差分：
原始数据: 1000, 1001, 1002, 1003, 1004
Delta: 1000, +1, +1, +1, +1
DoD: 1000, +1, 0, 0, 0
压缩效果更好，特别适合固定采样率的数据
3. Gorilla压缩算法
Facebook开发的时序数据压缩算法，用于InfluxDB等：
时间戳压缩： - 使用Delta-of-Delta编码 - 变长整数编码 - 压缩率可达96%
数值压缩： - XOR编码（相邻值XOR） - 前导零和尾随零压缩 - 压缩率可达90%
示例：
// 简化的XOR压缩
typedef struct {
uint64_t prev_value;
uint8_t prev_leading_zeros;
uint8_t prev_trailing_zeros;
} xor_compressor_t;
void xor_compress_value(xor_compressor_t *comp, double value,
uint8_t *output, size_t *output_len) {
uint64_t current = *(uint64_t*)&value;
uint64_t xor_val = current ^ comp->prev_value;
if (xor_val == 0) {
// 值相同，只需1位
output[0] = 0;
*output_len = 1;
} else {
// 计算前导零和尾随零
uint8_t leading = __builtin_clzll(xor_val);
uint8_t trailing = __builtin_ctzll(xor_val);
// 编码XOR值（简化版本）
// 实际实现需要更复杂的位操作
*output_len = 64 - leading - trailing;
}
comp->prev_value = current;
}
4. 游程编码（RLE）
适合有大量重复值的场景：
压缩效果对比¶
| 算法 | 压缩率 | CPU开销 | 适用场景 |
|---|---|---|---|
| Delta | 50-70% | 低 | 单调递增数据 |
| Delta-of-Delta | 70-90% | 低 | 固定采样率 |
| Gorilla | 85-96% | 中 | 通用时序数据 |
| RLE | 30-80% | 低 | 重复值多 |
| LZ4 | 40-60% | 中 | 通用压缩 |
| Zstd | 60-80% | 高 | 高压缩率需求 |
时序数据查询优化¶
时间范围查询¶
时序数据库最常见的查询模式：
1. 基本时间范围查询
-- 查询最近1小时的数据
SELECT timestamp, temperature, humidity
FROM sensor_data
WHERE timestamp >= NOW() - INTERVAL '1 hour'
ORDER BY timestamp DESC;
-- 查询特定时间段
SELECT *
FROM sensor_data
WHERE timestamp BETWEEN '2024-01-15 00:00:00'
AND '2024-01-15 23:59:59'
AND sensor_id = 'sensor01';
优化策略： - 在时间戳字段上创建索引 - 使用分区表按时间分区 - 限制返回的数据量 - 使用覆盖索引
2. 时间分区
将数据按时间分区可以大幅提升查询性能：
-- 按天分区
CREATE TABLE sensor_data_2024_01_15 (
CHECK (timestamp >= '2024-01-15' AND timestamp < '2024-01-16')
) INHERITS (sensor_data);
-- 自动分区（PostgreSQL）
CREATE TABLE sensor_data (
timestamp TIMESTAMPTZ NOT NULL,
sensor_id TEXT,
value DOUBLE PRECISION
) PARTITION BY RANGE (timestamp);
CREATE TABLE sensor_data_2024_01
PARTITION OF sensor_data
FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');
优势： - 查询只扫描相关分区 - 可以独立管理和删除旧分区 - 提升并发性能 - 简化数据归档
聚合查询优化¶
1. 时间窗口聚合
-- 按小时聚合
SELECT
DATE_TRUNC('hour', timestamp) as hour,
sensor_id,
AVG(temperature) as avg_temp,
MAX(temperature) as max_temp,
MIN(temperature) as min_temp,
COUNT(*) as sample_count
FROM sensor_data
WHERE timestamp >= NOW() - INTERVAL '24 hours'
GROUP BY hour, sensor_id
ORDER BY hour DESC;
-- 按5分钟聚合
SELECT
TIME_BUCKET('5 minutes', timestamp) as bucket,
AVG(value) as avg_value
FROM sensor_data
WHERE timestamp >= NOW() - INTERVAL '1 hour'
GROUP BY bucket
ORDER BY bucket;
2. 连续聚合（Continuous Aggregates）
预计算聚合结果，提升查询性能：
-- TimescaleDB连续聚合
CREATE MATERIALIZED VIEW sensor_data_hourly
WITH (timescaledb.continuous) AS
SELECT
TIME_BUCKET('1 hour', timestamp) as hour,
sensor_id,
AVG(temperature) as avg_temp,
MAX(temperature) as max_temp,
MIN(temperature) as min_temp
FROM sensor_data
GROUP BY hour, sensor_id;
-- 自动刷新策略
SELECT add_continuous_aggregate_policy('sensor_data_hourly',
start_offset => INTERVAL '3 hours',
end_offset => INTERVAL '1 hour',
schedule_interval => INTERVAL '1 hour');
优势： - 查询预计算结果，速度快 - 自动更新 - 减少实时计算开销 - 适合仪表板和报表
降采样（Downsampling）¶
降采样是减少数据点数量的技术，保留数据趋势的同时减少存储和查询开销：
1. 时间窗口降采样
-- 将1秒采样降为1分钟采样
INSERT INTO sensor_data_1min
SELECT
TIME_BUCKET('1 minute', timestamp) as timestamp,
sensor_id,
AVG(value) as value
FROM sensor_data_1sec
WHERE timestamp >= NOW() - INTERVAL '1 hour'
GROUP BY TIME_BUCKET('1 minute', timestamp), sensor_id;
2. 多级降采样策略
实现示例：
-- 创建不同精度的表
CREATE TABLE sensor_data_raw (
timestamp TIMESTAMPTZ NOT NULL,
sensor_id TEXT,
value DOUBLE PRECISION
);
CREATE TABLE sensor_data_1min (
timestamp TIMESTAMPTZ NOT NULL,
sensor_id TEXT,
avg_value DOUBLE PRECISION,
min_value DOUBLE PRECISION,
max_value DOUBLE PRECISION,
sample_count INTEGER
);
-- 定期降采样任务
CREATE OR REPLACE FUNCTION downsample_to_1min()
RETURNS void AS $$
BEGIN
INSERT INTO sensor_data_1min
SELECT
TIME_BUCKET('1 minute', timestamp) as timestamp,
sensor_id,
AVG(value) as avg_value,
MIN(value) as min_value,
MAX(value) as max_value,
COUNT(*) as sample_count
FROM sensor_data_raw
WHERE timestamp >= NOW() - INTERVAL '1 hour'
AND timestamp < NOW() - INTERVAL '5 minutes'
GROUP BY TIME_BUCKET('1 minute', timestamp), sensor_id
ON CONFLICT (timestamp, sensor_id) DO NOTHING;
END;
$$ LANGUAGE plpgsql;
索引策略¶
1. 复合索引
-- 时间戳 + 传感器ID复合索引
CREATE INDEX idx_sensor_time ON sensor_data(sensor_id, timestamp DESC);
-- 适合的查询
SELECT * FROM sensor_data
WHERE sensor_id = 'sensor01'
AND timestamp >= NOW() - INTERVAL '1 hour'
ORDER BY timestamp DESC;
2. 部分索引
只为常用查询创建索引：
-- 只索引最近30天的数据
CREATE INDEX idx_recent_data ON sensor_data(timestamp)
WHERE timestamp >= NOW() - INTERVAL '30 days';
3. BRIN索引（Block Range Index）
适合时序数据的轻量级索引：
优势： - 索引大小极小（通常<1%数据大小） - 适合顺序写入的数据 - 查询性能良好 - 维护成本低
数据保留策略¶
为什么需要数据保留策略¶
时序数据持续增长，需要自动清理旧数据：
- 存储空间限制
- 嵌入式设备存储有限
- 避免磁盘空间耗尽
-
控制存储成本
-
查询性能
- 数据量越大，查询越慢
- 保持数据集在合理范围
-
提升索引效率
-
合规要求
- 数据保留期限规定
- 隐私保护要求
- 审计需求
保留策略实现¶
1. 基于时间的保留
-- 删除30天前的数据
DELETE FROM sensor_data
WHERE timestamp < NOW() - INTERVAL '30 days';
-- 使用分区表，直接删除旧分区（更快）
DROP TABLE sensor_data_2023_12;
2. 基于数据量的保留
-- 保留最新的100万条记录
DELETE FROM sensor_data
WHERE id NOT IN (
SELECT id FROM sensor_data
ORDER BY timestamp DESC
LIMIT 1000000
);
3. 分级保留策略
-- 原始数据保留7天
DELETE FROM sensor_data_raw
WHERE timestamp < NOW() - INTERVAL '7 days';
-- 1分钟聚合保留30天
DELETE FROM sensor_data_1min
WHERE timestamp < NOW() - INTERVAL '30 days';
-- 1小时聚合保留1年
DELETE FROM sensor_data_1hour
WHERE timestamp < NOW() - INTERVAL '1 year';
4. 自动保留策略
-- TimescaleDB自动保留策略
SELECT add_retention_policy('sensor_data', INTERVAL '30 days');
-- InfluxDB保留策略
CREATE RETENTION POLICY "30_days" ON "mydb"
DURATION 30d
REPLICATION 1
DEFAULT;
嵌入式系统实现¶
/**
* @brief 清理旧数据
* @param db: 数据库连接
* @param retention_days: 保留天数
* @retval 删除的记录数
*/
int cleanup_old_data(sqlite3 *db, int retention_days) {
char sql[256];
char *err_msg = NULL;
int rc;
// 计算截止时间戳
uint32_t cutoff_time = time(NULL) - (retention_days * 24 * 3600);
// 删除旧数据
snprintf(sql, sizeof(sql),
"DELETE FROM sensor_data WHERE timestamp < %u;",
cutoff_time);
rc = sqlite3_exec(db, sql, NULL, NULL, &err_msg);
if (rc != SQLITE_OK) {
printf("Failed to cleanup: %s\n", err_msg);
sqlite3_free(err_msg);
return -1;
}
// 获取删除的行数
int deleted = sqlite3_changes(db);
// 清理碎片
sqlite3_exec(db, "VACUUM;", NULL, NULL, NULL);
printf("Cleaned up %d old records\n", deleted);
return deleted;
}
/**
* @brief 定期清理任务（在主循环中调用）
*/
void periodic_cleanup_task(void) {
static uint32_t last_cleanup = 0;
uint32_t current_time = HAL_GetTick();
// 每24小时清理一次
if (current_time - last_cleanup >= 24 * 3600 * 1000) {
cleanup_old_data(db, 30); // 保留30天
last_cleanup = current_time;
}
}
嵌入式系统应用实例¶
案例1：环境监测系统¶
需求： - 每秒采集温度、湿度、气压数据 - 保留原始数据7天 - 提供小时/天级别的统计查询 - 存储空间限制在100MB
方案设计：
// 数据结构
typedef struct {
uint32_t timestamp;
int16_t temperature; // 温度 * 10（节省空间）
uint16_t humidity; // 湿度 * 10
uint16_t pressure; // 气压 - 900（偏移压缩）
} __attribute__((packed)) sensor_record_t;
// 数据库表设计
CREATE TABLE sensor_data_raw (
timestamp INTEGER PRIMARY KEY,
temperature INTEGER,
humidity INTEGER,
pressure INTEGER
);
CREATE TABLE sensor_data_hourly (
hour INTEGER PRIMARY KEY,
avg_temperature REAL,
min_temperature REAL,
max_temperature REAL,
avg_humidity REAL,
avg_pressure REAL,
sample_count INTEGER
);
数据流程：
性能优化：
/**
* @brief 批量写入传感器数据
*/
int batch_insert_sensor_data(sqlite3 *db, sensor_record_t *records,
int count) {
sqlite3_stmt *stmt;
int rc;
// 开始事务
sqlite3_exec(db, "BEGIN TRANSACTION;", NULL, NULL, NULL);
// 准备语句
const char *sql =
"INSERT INTO sensor_data_raw VALUES (?, ?, ?, ?);";
sqlite3_prepare_v2(db, sql, -1, &stmt, NULL);
// 批量插入
for (int i = 0; i < count; i++) {
sqlite3_bind_int(stmt, 1, records[i].timestamp);
sqlite3_bind_int(stmt, 2, records[i].temperature);
sqlite3_bind_int(stmt, 3, records[i].humidity);
sqlite3_bind_int(stmt, 4, records[i].pressure);
sqlite3_step(stmt);
sqlite3_reset(stmt);
}
sqlite3_finalize(stmt);
// 提交事务
sqlite3_exec(db, "COMMIT;", NULL, NULL, NULL);
return count;
}
/**
* @brief 计算小时聚合数据
*/
int calculate_hourly_aggregates(sqlite3 *db) {
const char *sql =
"INSERT OR REPLACE INTO sensor_data_hourly "
"SELECT "
" timestamp / 3600 * 3600 as hour, "
" AVG(temperature) / 10.0 as avg_temperature, "
" MIN(temperature) / 10.0 as min_temperature, "
" MAX(temperature) / 10.0 as max_temperature, "
" AVG(humidity) / 10.0 as avg_humidity, "
" AVG(pressure) + 900 as avg_pressure, "
" COUNT(*) as sample_count "
"FROM sensor_data_raw "
"WHERE timestamp >= ? AND timestamp < ? "
"GROUP BY hour;";
// 执行聚合（实现略）
return 0;
}
案例2：设备状态监控¶
需求： - 监控多个设备的运行状态 - 记录状态变化事件 - 统计设备运行时长 - 生成设备健康报告
数据模型：
-- 设备状态事件表
CREATE TABLE device_events (
timestamp INTEGER NOT NULL,
device_id TEXT NOT NULL,
event_type TEXT NOT NULL, -- 'start', 'stop', 'error'
state TEXT,
message TEXT,
PRIMARY KEY (timestamp, device_id)
);
-- 设备运行统计表
CREATE TABLE device_stats (
date INTEGER PRIMARY KEY,
device_id TEXT,
total_runtime INTEGER, -- 总运行时长（秒）
error_count INTEGER,
start_count INTEGER
);
查询示例：
-- 查询设备最近的状态
SELECT * FROM device_events
WHERE device_id = 'device01'
ORDER BY timestamp DESC
LIMIT 10;
-- 统计设备今日运行时长
SELECT
device_id,
SUM(CASE WHEN event_type = 'start' THEN 1 ELSE 0 END) as starts,
SUM(CASE WHEN event_type = 'error' THEN 1 ELSE 0 END) as errors
FROM device_events
WHERE timestamp >= strftime('%s', 'now', 'start of day')
GROUP BY device_id;
数据可视化¶
时序数据可视化方法¶
1. 折线图（Line Chart）
最常用的时序数据可视化方式：
// 使用Chart.js示例
const chartData = {
labels: timestamps, // 时间戳数组
datasets: [{
label: '温度',
data: temperatures, // 温度数组
borderColor: 'rgb(255, 99, 132)',
tension: 0.1
}]
};
适用场景： - 连续变化的数据 - 趋势分析 - 多指标对比
2. 面积图（Area Chart）
强调数据量的累积效果：
const chartData = {
datasets: [{
label: '能耗',
data: powerConsumption,
fill: true,
backgroundColor: 'rgba(75, 192, 192, 0.2)',
borderColor: 'rgb(75, 192, 192)'
}]
};
适用场景： - 累积数据 - 占比分析 - 容量监控
3. 热力图（Heatmap）
展示时间和另一维度的关系：
00:00 06:00 12:00 18:00 24:00
Mon ████ ██ ████ ████ ██
Tue ██ ████ ██ ████ ████
Wed ████ ██ ████ ██ ████
适用场景： - 周期性模式分析 - 异常检测 - 资源使用分析
4. 仪表盘（Dashboard）
综合展示多个指标：
┌─────────────────────────────────────────┐
│ 实时监控仪表盘 │
├─────────────────────────────────────────┤
│ 温度: 23.5°C │ 湿度: 65% │ 在线设备: 12 │
├─────────────────────────────────────────┤
│ [温度趋势图] │
│ [湿度趋势图] │
├─────────────────────────────────────────┤
│ [告警列表] │ [设备状态] │
└─────────────────────────────────────────┘
实时数据推送¶
WebSocket实时更新：
// 前端WebSocket连接
const ws = new WebSocket('ws://device-ip:8080/realtime');
ws.onmessage = function(event) {
const data = JSON.parse(event.data);
updateChart(data.timestamp, data.value);
};
// 后端推送（伪代码）
void send_realtime_data(void) {
sensor_data_t data = read_sensor();
char json[128];
snprintf(json, sizeof(json),
"{\"timestamp\":%lu,\"temperature\":%.2f,\"humidity\":%.2f}",
data.timestamp, data.temperature, data.humidity);
websocket_broadcast(json);
}
数据导出¶
CSV导出：
/**
* @brief 导出数据为CSV格式
*/
int export_to_csv(sqlite3 *db, const char *filename,
uint32_t start_time, uint32_t end_time) {
FILE *fp = fopen(filename, "w");
if (!fp) return -1;
// 写入CSV头
fprintf(fp, "Timestamp,Temperature,Humidity,Pressure\n");
// 查询数据
sqlite3_stmt *stmt;
const char *sql =
"SELECT timestamp, temperature, humidity, pressure "
"FROM sensor_data_raw "
"WHERE timestamp BETWEEN ? AND ? "
"ORDER BY timestamp;";
sqlite3_prepare_v2(db, sql, -1, &stmt, NULL);
sqlite3_bind_int(stmt, 1, start_time);
sqlite3_bind_int(stmt, 2, end_time);
// 写入数据行
while (sqlite3_step(stmt) == SQLITE_ROW) {
uint32_t timestamp = sqlite3_column_int(stmt, 0);
float temperature = sqlite3_column_int(stmt, 1) / 10.0f;
float humidity = sqlite3_column_int(stmt, 2) / 10.0f;
float pressure = sqlite3_column_int(stmt, 3) + 900.0f;
fprintf(fp, "%u,%.1f,%.1f,%.1f\n",
timestamp, temperature, humidity, pressure);
}
sqlite3_finalize(stmt);
fclose(fp);
return 0;
}
性能优化最佳实践¶
写入优化¶
1. 批量写入
// 不推荐：逐条插入
for (int i = 0; i < 1000; i++) {
insert_single_record(data[i]); // 每次都提交
}
// 推荐：批量插入
BEGIN TRANSACTION;
for (int i = 0; i < 1000; i++) {
insert_single_record(data[i]);
}
COMMIT;
// 性能提升：100-1000倍
2. 预编译语句
// 不推荐：每次都编译SQL
for (int i = 0; i < 1000; i++) {
sprintf(sql, "INSERT INTO ... VALUES (%d, %f)", ...);
sqlite3_exec(db, sql, ...);
}
// 推荐：预编译语句
sqlite3_prepare_v2(db, "INSERT INTO ... VALUES (?, ?)", ...);
for (int i = 0; i < 1000; i++) {
sqlite3_bind_int(...);
sqlite3_step(...);
sqlite3_reset(...);
}
sqlite3_finalize(...);
3. 异步写入
// 使用缓冲区异步写入
#define BUFFER_SIZE 100
typedef struct {
sensor_record_t buffer[BUFFER_SIZE];
int count;
pthread_mutex_t mutex;
} write_buffer_t;
void async_write_data(sensor_record_t *data) {
pthread_mutex_lock(&buffer.mutex);
buffer.buffer[buffer.count++] = *data;
if (buffer.count >= BUFFER_SIZE) {
// 触发批量写入
flush_buffer_to_db();
buffer.count = 0;
}
pthread_mutex_unlock(&buffer.mutex);
}
查询优化¶
1. 限制返回数据量
-- 不推荐：返回所有数据
SELECT * FROM sensor_data;
-- 推荐：限制数量和时间范围
SELECT * FROM sensor_data
WHERE timestamp >= NOW() - INTERVAL '1 hour'
LIMIT 1000;
2. 使用覆盖索引
-- 创建覆盖索引
CREATE INDEX idx_sensor_time_value
ON sensor_data(sensor_id, timestamp, value);
-- 查询只需要索引，不需要访问表
SELECT sensor_id, timestamp, value
FROM sensor_data
WHERE sensor_id = 'sensor01'
AND timestamp >= ?;
3. 避免全表扫描
-- 不推荐：没有WHERE条件
SELECT AVG(temperature) FROM sensor_data;
-- 推荐：限制时间范围
SELECT AVG(temperature) FROM sensor_data
WHERE timestamp >= NOW() - INTERVAL '1 day';
存储优化¶
1. 数据类型选择
-- 不推荐：使用TEXT存储数值
CREATE TABLE sensor_data (
timestamp TEXT,
value TEXT
);
-- 推荐：使用合适的数值类型
CREATE TABLE sensor_data (
timestamp INTEGER, -- Unix时间戳
value REAL -- 浮点数
);
2. 数据压缩
// 使用整数存储浮点数（精度损失可接受）
int16_t temperature_compressed = (int16_t)(temperature * 10);
// 存储时
INSERT INTO sensor_data VALUES (timestamp, temperature_compressed);
// 读取时
float temperature = temperature_compressed / 10.0f;
3. 定期维护
常见问题与解决方案¶
问题1：写入性能不足¶
症状： - 数据写入延迟高 - 缓冲区溢出 - 数据丢失
解决方案： 1. 使用批量写入和事务 2. 增加写入缓冲区大小 3. 使用异步写入 4. 优化磁盘I/O（使用SSD） 5. 考虑使用内存数据库
问题2：查询速度慢¶
症状： - 查询响应时间长 - 系统资源占用高 - 用户体验差
解决方案： 1. 创建合适的索引 2. 使用时间分区 3. 限制查询范围 4. 使用预聚合数据 5. 实施缓存策略
问题3：存储空间不足¶
症状： - 磁盘空间耗尽 - 写入失败 - 系统崩溃
解决方案： 1. 实施数据保留策略 2. 使用数据压缩 3. 实施降采样 4. 定期清理旧数据 5. 监控存储使用情况
问题4：数据丢失¶
症状： - 断电后数据丢失 - 数据不完整 - 数据损坏
解决方案： 1. 使用WAL模式（Write-Ahead Logging） 2. 定期同步到持久存储 3. 实施数据备份 4. 使用事务保证一致性 5. 添加数据校验
问题5：时间同步问题¶
症状： - 时间戳不准确 - 数据顺序混乱 - 聚合计算错误
解决方案： 1. 使用NTP同步时间 2. 使用RTC硬件时钟 3. 记录时间偏移 4. 使用单调递增的序列号 5. 实施时间校准机制
总结¶
核心要点¶
- 时序数据库特点
- 专为时间序列数据设计
- 高效的写入和查询性能
- 优秀的数据压缩能力
-
自动化的数据管理
-
关键技术
- Delta和Gorilla压缩算法
- 时间分区和索引优化
- 降采样和聚合查询
-
数据保留策略
-
嵌入式应用
- SQLite + 时序优化
- 自定义轻量级方案
- 批量写入和异步处理
-
资源受限环境优化
-
性能优化
- 批量操作和事务
- 合适的索引策略
- 数据压缩和分区
- 定期维护和清理
选型建议¶
选择时序数据库的考虑因素：
| 场景 | 推荐方案 | 理由 |
|---|---|---|
| 云端大规模IoT | InfluxDB/TimescaleDB | 高性能、易扩展 |
| 边缘计算网关 | SQLite + 优化 | 轻量级、可靠 |
| 嵌入式设备 | 自定义轻量级 | 资源占用最小 |
| 混合架构 | 边缘+云端 | 本地缓存+云端存储 |
决策树：
最佳实践清单¶
设计阶段： - [ ] 明确数据采集频率和数据量 - [ ] 规划数据保留策略 - [ ] 设计合理的数据模型 - [ ] 评估存储和性能需求 - [ ] 选择合适的数据库方案
实现阶段： - [ ] 使用批量写入和事务 - [ ] 创建合适的索引 - [ ] 实施数据压缩 - [ ] 实现降采样机制 - [ ] 添加数据校验
运维阶段： - [ ] 监控存储使用情况 - [ ] 定期清理旧数据 - [ ] 执行数据库维护 - [ ] 备份重要数据 - [ ] 性能调优和优化
进阶学习路径¶
- 深入学习
- 研究InfluxDB源码
- 学习Gorilla压缩算法
- 掌握时序数据分析
-
了解流式计算
-
实践项目
- 构建完整的IoT监控系统
- 实现自定义时序数据库
- 开发数据可视化平台
-
集成机器学习预测
-
相关技术
- 流式处理（Kafka、Flink）
- 数据分析（Pandas、NumPy）
- 可视化（Grafana、Kibana）
- 机器学习（时序预测）
参考资源¶
官方文档¶
- InfluxDB: https://docs.influxdata.com/
- TimescaleDB: https://docs.timescale.com/
- SQLite: https://www.sqlite.org/docs.html
- Prometheus: https://prometheus.io/docs/
开源项目¶
- InfluxDB: https://github.com/influxdata/influxdb
- TimescaleDB: https://github.com/timescale/timescaledb
- Prometheus: https://github.com/prometheus/prometheus
- Grafana: https://github.com/grafana/grafana
学习资源¶
论文： - "Gorilla: A Fast, Scalable, In-Memory Time Series Database" (Facebook) - "The Log-Structured Merge-Tree (LSM-Tree)" (O'Neil et al.)
书籍： - 《Time Series Databases: New Ways to Store and Access Data》 - 《Database Internals》 by Alex Petrov
在线课程： - Coursera: Time Series Analysis - Udemy: InfluxDB and Time Series Data
工具和库¶
数据库： - InfluxDB - 专业时序数据库 - TimescaleDB - PostgreSQL扩展 - Prometheus - 监控和时序数据库 - OpenTSDB - 基于HBase的时序数据库
可视化： - Grafana - 数据可视化平台 - Kibana - Elasticsearch可视化 - Chronograf - InfluxDB可视化
客户端库： - influxdb-client-python - timescaledb-python - prometheus-client
社区资源¶
- InfluxDB Community: https://community.influxdata.com/
- TimescaleDB Forum: https://www.timescale.com/forum
- Stack Overflow: 搜索 "time-series-database"
- Reddit: r/timeseries, r/database
练习与思考¶
基础练习¶
- 数据压缩实验
- 实现Delta编码算法
- 对比压缩前后的存储空间
-
测量压缩和解压缩性能
-
查询优化
- 创建时序数据表
- 插入100万条测试数据
-
对比有无索引的查询性能
-
降采样实现
- 实现1秒到1分钟的降采样
- 保留最大值、最小值、平均值
- 验证数据准确性
进阶项目¶
- 环境监测系统
- 采集温湿度数据
- 实现数据存储和查询
- 开发Web可视化界面
-
添加告警功能
-
设备健康监控
- 记录设备运行状态
- 统计运行时长和故障次数
- 生成健康报告
-
实现预测性维护
-
能耗分析系统
- 采集电力消耗数据
- 按小时/天/月聚合
- 分析用电模式
- 提供节能建议
思考题¶
-
为什么时序数据库的写入性能通常比传统数据库高？
-
在资源受限的嵌入式系统中，如何平衡数据精度和存储空间？
-
如何设计一个既能满足实时查询又能长期存储的时序数据系统？
-
降采样会丢失信息，如何判断哪些信息可以丢弃？
-
在分布式IoT系统中，如何处理设备时间不同步的问题？
下一步学习¶
完成本文章的学习后，建议继续学习：
- 数据同步与备份策略
- 学习数据同步机制
- 掌握增量备份技术
-
了解云端同步方案
-
分布式数据管理系统
- 学习分布式架构
- 掌握数据分片技术
-
了解一致性协议
-
流式数据处理
- 学习Kafka、Flink等流式处理框架
- 掌握实时数据分析
-
了解复杂事件处理（CEP）
-
机器学习与时序预测
- 学习时序分析方法
- 掌握ARIMA、LSTM等预测模型
- 实现异常检测算法
作者: 嵌入式知识平台
最后更新: 2024-01-15
版本: 1.0
如有问题或建议，欢迎通过以下方式联系我们： - 邮箱: support@embedded-platform.com - 论坛: https://forum.embedded-platform.com - GitHub: https://github.com/embedded-platform