当前位置：首页 > news >正文

时序数据库详解

news 2026/5/10 22:58:08

时序数据库详解

1. 时序数据库概述

时序数据库（Time Series Database）专门存储和时间相关的数据，如监控指标、传感器数据、金融数据等。

2. InfluxDB

2.1 核心概念

Measurement：相当于表
Tag：索引字段，用于过滤
Field：数据字段
Timestamp：时间戳

2.2 InfluxQL

-- 插入数据 INSERT cpu,host=server1,region=us value=0.5 1672531200000000000 -- 查询数据 SELECT * FROM cpu WHERE time > now() - 1h -- 聚合查询 SELECT MEAN(value) FROM cpu WHERE time > now() - 1h GROUP BY time(5m) -- 连续查询 CREATE CONTINUOUS QUERY cpu_mean ON mydb BEGIN SELECT MEAN(value) INTO cpu_mean FROM cpu GROUP BY time(5m), * END

2.3 Go客户端

import "github.com/influxdata/influxdb-client-go/v2" client := influxdb2.NewClient("http://localhost:8086", "token") writeAPI := client.WriteAPIBlocking("my-org", "my-bucket") p := api.NewPoint("cpu", api.Tag("host", "server1"), api.Tag("region", "us"), api Fields("load", 0.5)) writeAPI.WritePoint(ctx, p) client.Close()

3. TimescaleDB

3.1 创建超表

-- 创建超表 CREATE TABLE conditions ( time TIMESTAMPTZ NOT NULL, location TEXT NOT NULL, temperature DOUBLE PRECISION ); SELECT create_hypertable('conditions', 'time'); -- 创建索引 CREATE INDEX ON conditions (location, time DESC);

3.2 写入数据

INSERT INTO conditions (time, location, temperature) VALUES ('2023-01-01 00:00:00', 'NYC', 5.0);

3.3 聚合查询

-- 时间聚合 SELECT time_bucket('1 hour', time) AS hour, AVG(temperature) FROM conditions WHERE location = 'NYC' GROUP BY hour ORDER BY hour DESC LIMIT 10; -- 降采样 SELECT time_bucket('1 day', time) AS day, MAX(temperature) - MIN(temperature) AS temp_range FROM conditions GROUP BY day ORDER BY day DESC;

4. QuestDB

4.1 快速写入

-- 创建表 CREATE TABLE sensors ( ts TIMESTAMP, sensor_id SYMBOL, temperature DOUBLE, humidity DOUBLE ) TIMESTAMP(ts) PARTITION BY DAY; -- 批量写入 INSERT INTO sensors VALUES ('2023-01-01T00:00:00.000Z', 'sensor_1', 22.5, 65.0), ('2023-01-01T00:00:00.000Z', 'sensor_2', 23.1, 63.5);

4.2 时间序列查询

-- 最近24小时数据 SELECT * FROM sensors WHERE ts > now() - 24h ORDER BY ts DESC; -- 按传感器聚合 SELECT sensor_id, AVG(temperature), MAX(temperature) FROM sensors WHERE ts > now() - 7d GROUP BY sensor_id;

5. Go客户端选择

5.1 InfluxDB Go客户端

import "github.com/influxdata/influxdb-client-go/v2" client := influxdb2.NewClient("http://localhost:8086", "token") defer client.Close() writeAPI := client.WriteAPIBlocking("my-org", "my-bucket") point := api.NewPoint("measurement"). AddTag("tag1", "value1"). AddField("field1", 10.5) writeAPI.WritePoint(ctx, point)

5.2 TimescaleDB Go客户端

import ( "github.com/jackc/pgx/v4" "github.com/timescale/timescaledb-go/sql" ) conn, _ := pgx.Connect(ctx, "postgres://user:pass@localhost:5432/db") defer conn.Close() // 创建超表 sql.CreateHypertable(ctx, conn, "conditions", "time")

6. 数据保留策略

6.1 InfluxDB保留策略

-- 创建保留策略 CREATE RETENTION POLICY "7d" ON "mydb" DURATION 7d REPLICATION 1 SHARD DURATION 1d DEFAULT; -- 修改保留策略 ALTER RETENTION POLICY "7d" ON "mydb" DURATION 30d;

6.2 TimescaleDB数据保留

-- 添加数据保留策略 SELECT add_retention_policy('conditions', INTERVAL '7 days');

7. 降采样与聚合

7.1 连续查询降采样

-- InfluxDB降采样 CREATE CONTINUOUS QUERY "1h_avg" ON "mydb" BEGIN SELECT MEAN(temperature) AS temp INTO "conditions_1h" FROM "conditions" GROUP BY time(1h), sensor_id END;

7.2 TimescaleDB降采样

-- 创建降采样表 CREATE TABLE conditions_1h ( time TIMESTAMPTZ NOT NULL, sensor_id TEXT NOT NULL, temp_avg DOUBLE PRECISION, temp_min DOUBLE PRECISION, temp_max DOUBLE PRECISION ); SELECT create_hypertable('conditions_1h', 'time'); -- 使用 refresh_continuous_aggregate 定期刷新

8. 数据压缩

8.1 TimescaleDB压缩

-- 启用压缩 ALTER TABLE conditions SET ( timescaledb.compress, timescaledb.compress_segmentby = 'sensor_id' ); -- 添加压缩策略 SELECT add_compression_policy('conditions', INTERVAL '7 days');