TimescaleDB 核心语法与使用示例全集1. 超表 (Hypertable) 管理1.1 创建超表-- 基础创建将普通表转换为超表 CREATE TABLE conditions ( time TIMESTAMPTZ NOT NULL, location TEXT NOT NULL, temperature DOUBLE PRECISION ); SELECT create_hypertable(conditions, time); -- 创建时指定分区间隔1天 SELECT create_hypertable(conditions, time, chunk_time_interval INTERVAL 1 day); -- 创建时指定空间分区按location字段 SELECT create_hypertable(conditions, time, partitioning_column location, number_partitions 4); -- 创建带索引的超表 CREATE TABLE metrics ( ts TIMESTAMPTZ NOT NULL, device_id INT NOT NULL, value FLOAT NOT NULL ); CREATE INDEX ON metrics (device_id, ts DESC); SELECT create_hypertable(metrics, ts);1.2 超表信息查询-- 查看所有超表 SELECT hypertable_name, num_chunks, table_size FROM timescaledb_information.hypertables; -- 查看特定超表的块信息 SELECT chunk_name, range_start, range_end, is_compressed FROM timescaledb_information.chunks WHERE hypertable_name conditions; -- 查看超表维度信息 SELECT * FROM timescaledb_information.dimensions WHERE hypertable_name conditions; -- 查看超表大小 SELECT hypertable_size(conditions) as total_size; SELECT * FROM hypertable_detailed_size(conditions);1.3 超表维护操作-- 修改块时间间隔 SELECT set_chunk_time_interval(conditions, INTERVAL 2 days); -- 添加新维度列SELECT add_dimension(conditions, location, number_partitions 4); -- 手动创建新块 SELECT create_chunk(conditions, chunk_time_interval INTERVAL 1 day); -- 重新排序块SELECT reorder_chunk(_timescaledb_internal._hyper_1_chunk_1);2. 时间桶函数 (Time Bucket)2.1 基础时间桶-- 按1小时分组 SELECT time_bucket(1 hour, time) AS hour, AVG(temperature) as avg_temp FROM conditions GROUP BY hour ORDER BY hour; -- 按1天分组对齐到周一 SELECT time_bucket(1 day, time, Monday) AS day, COUNT(*) as readings FROM conditions GROUP BY day; -- 按5分钟分组带时区转换 SELECT time_bucket(5 minutes, time AT TIME ZONE UTC) AS bucket, MAX(temperature) as max_temp FROM conditions GROUP BY bucket;2.2 时间桶与偏移-- 带偏移的时间桶对齐到特定时间点 SELECT time_bucket(1 day, time, INTERVAL 6 hours) AS day_start, location, AVG(temperature) as avg_temp FROM conditions GROUP BY day_start, location; -- 可变时间桶基于数据量 SELECT time_bucket_gapfill(1 hour, time, start NOW() - INTERVAL 7 days, finish NOW()) AS hour, locf(AVG(temperature)) as avg_temp -- 使用最后观测值向前填充 FROM conditions WHERE time NOW() - INTERVAL 7 days GROUP BY hour;3. 连续聚合 (Continuous Aggregates)3.1 创建连续聚合-- 基础连续聚合 CREATE MATERIALIZED VIEW conditions_summary_hourly WITH (timescaledb.continuous) AS SELECT time_bucket(1 hour, time) AS bucket, location, AVG(temperature) as avg_temp, MAX(temperature) as max_temp, MIN(temperature) as min_temp, COUNT(*) as readings FROM conditions GROUP BY bucket, location; -- 带WHERE条件的连续聚合 CREATE MATERIALIZED VIEW conditions_daily WITH (timescaledb.continuous) AS SELECT time_bucket(1 day, time) AS day, location, PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY temperature) as p95_temp FROM conditions WHERE temperature IS NOT NULL GROUP BY day, location; -- 带JOIN的连续聚合 CREATE MATERIALIZED VIEW device_metrics_daily WITH (timescaledb.continuous) AS SELECT time_bucket(1 day, m.ts) AS day, d.device_name, AVG(m.value) as avg_value, SUM(m.value) as total_value FROM metrics m JOIN devices d ON m.device_id d.id GROUP BY day, d.device_name;3.2 连续聚合策略管理-- 添加刷新策略 SELECT add_continuous_aggregate_policy( conditions_summary_hourly, start_offset INTERVAL 3 days, end_offset INTERVAL 1 hour, schedule_interval INTERVAL 30 minutes ); -- 查看连续聚合信息 SELECT view_name, materialized_only, enabledFROM timescaledb_information.continuous_aggregates; -- 手动刷新连续聚合 CALL refresh_continuous_aggregate( conditions_summary_hourly, window_start NOW() - INTERVAL 1 day, window_end NOW() ); -- 暂停/恢复策略 SELECT alter_job(job_id, scheduled false) FROM timescaledb_information.jobs WHERE proc_name policy_refresh_continuous_aggregate; SELECT alter_job(job_id, scheduled true) FROM timescaledb_information.jobs WHERE proc_name policy_refresh_continuous_aggregate;4. 压缩 (Compression)4.1 启用压缩-- 基础压缩配置 ALTER TABLE conditions SET ( timescaledb.compress, timescaledb.compress_segmentby location, timescaledb.compress_orderby time DESC ); -- 多字段分段压缩 ALTER TABLE metrics SET ( timescaledb.compress, timescaledb.compress_segmentby device_id, timescaledb.compress_orderby ts DESC, value ); -- 查看压缩设置 SELECT * FROM timescaledb_information.compression_settings WHERE hypertable_name conditions;4.2 压缩策略-- 添加压缩策略SELECT add_compression_policy(conditions, compress_after INTERVAL 7 days); -- 手动压缩数据SELECT compress_chunk(_timescaledb_internal._hyper_1_chunk_1); -- 解压缩数据 SELECT decompress_chunk(_timescaledb_internal._hyper_1_chunk_1); -- 查看压缩状态 SELECT chunk_name, is_compressed, compressed_size FROM timescaledb_information.chunks WHERE hypertable_name conditions;5. 数据保留策略 (Retention)5.1 保留策略管理-- 添加保留策略SELECT add_retention_policy(conditions, drop_after INTERVAL 90 days); -- 带间隔的保留策略 SELECT add_retention_policy(metrics, drop_after INTERVAL 180 days, schedule_interval INTERVAL 1 day); -- 查看保留策略 SELECT * FROM timescaledb_information.jobs WHERE proc_name policy_retention; -- 手动删除旧数据 SELECT drop_chunks(conditions, older_than INTERVAL 90 days); -- 删除特定时间范围的数据 SELECT drop_chunks(conditions, newer_than INTERVAL 30 days, older_than INTERVAL 90 days);6. 分布式超表 (Distributed Hypertable)6.1 创建分布式超表-- 启用分布式扩展 CREATE EXTENSION IF NOT EXISTS timescaledb_toolkit; -- 创建分布式超表 CREATE TABLE distributed_metrics ( time TIMESTAMPTZ NOT NULL, metric_name TEXT NOT NULL, value DOUBLE PRECISION, region TEXT NOT NULL ); SELECT create_distributed_hypertable( distributed_metrics, time, region, number_partitions 4, replication_factor 2 ); -- 添加数据节点 SELECT add_data_node(node1, host 192.168.1.101); SELECT add_data_node(node2, host 192.168.1.102); -- 查看数据节点SELECT * FROM timescaledb_information.data_nodes;7. 高级查询函数7.1 窗口函数-- 移动平均 SELECT time, temperature, AVG(temperature) OVER ( ORDER BY time ROWS BETWEEN 6 PRECEDING AND CURRENT ROW ) as moving_avg_7FROM conditions WHERE location room1; -- 时间加权平均 SELECT time_bucket(1 hour, time) as hour, time_weight(Linear, time, temperature) as time_weighted_temp FROM conditions GROUP BY hour; -- 百分位数计算 SELECT time_bucket(1 day, time) as day, approx_percentile(0.95, percentile_agg(temperature)) as p95_temp FROM conditions GROUP BY day;7.2 间隙填充-- 时间序列间隙填充 SELECT time_bucket_gapfill(5 minutes, time, start 2024-01-01, finish 2024-01-02) as bucket, locf(AVG(temperature)) as temp_filled, interpolate(AVG(humidity)) as humidity_filled FROM conditions WHERE time BETWEEN 2024-01-01 AND 2024-01-02 GROUP BY bucket; -- 多重间隙填充 SELECT time_bucket_gapfill(1 hour, time) as hour, device_id, locf(AVG(value), prev (SELECT AVG(value) FROM metrics WHERE device_id m.device_id AND time m.time - INTERVAL 1 day) ) as value_filled FROM metrics m GROUP BY hour, device_id;8. 性能优化函数8.1 查询优化-- 使用更快的近似函数 SELECT time_bucket(1 hour, time) as hour, approx_count_distinct(device_id) as unique_devices, approx_percentile(0.99, percentile_agg(value)) as p99_value FROM metrics GROUP BY hour; -- 分层聚合 WITH hourly AS ( SELECT time_bucket(1 hour, time) as hour, device_id, AVG(value) as avg_value FROM metrics GROUP BY hour, device_id ) SELECT time_bucket(1 day, hour) as day, device_id, AVG(avg_value) as daily_avg FROM hourly GROUP BY day, device_id;8.2 索引优化-- 创建默认索引 CREATE INDEX ON conditions (time DESC, location); -- 创建多列索引 CREATE INDEX ON metrics (device_id, ts DESC) INCLUDE (value); -- 创建表达式索引 CREATE INDEX ON conditions ((time::date), location); -- 查看索引使用情况 SELECT * FROM timescaledb_information.hypertable_index_stats WHERE hypertable_name conditions;9. 监控与管理函数9.1 系统监控-- 查看作业状态 SELECT job_id, proc_name, scheduled, runs_successful, runs_failed FROM timescaledb_information.jobs; -- 查看作业统计 SELECT * FROM timescaledb_information.job_stats; -- 查看锁信息 SELECT * FROM timescaledb_information.locks; -- 查看压缩统计 SELECT * FROM timescaledb_information.compression_stats;9.2 维护操作-- 手动执行维护任务 CALL run_maintenance(); -- 重新排序所有块 SELECT reorder_chunks(conditions); -- 清理旧数据 SELECT drop_chunks_older_than(conditions, INTERVAL 365 days); -- 重新压缩所有块 SELECT compress_chunks(conditions, older_than INTERVAL 30 days);10. 完整示例物联网监控系统-- 1. 创建设备表 CREATE TABLE devices ( id SERIAL PRIMARY KEY, device_name TEXT NOT NULL, device_type TEXT NOT NULL, location TEXT, installed_date TIMESTAMPTZ DEFAULT NOW() ); -- 2. 创建遥测数据超表 CREATE TABLE telemetry ( time TIMESTAMPTZ NOT NULL, device_id INT NOT NULL REFERENCES devices(id), metric_name TEXT NOT NULL, value DOUBLE PRECISION NOT NULL, status_code INT DEFAULT 200 ); SELECT create_hypertable(telemetry, time, chunk_time_interval INTERVAL 1 day, create_default_indexes false); CREATE INDEX ON telemetry (device_id, time DESC, metric_name); -- 3. 启用压缩 ALTER TABLE telemetry SET ( timescaledb.compress, timescaledb.compress_segmentby device_id, metric_name, timescaledb.compress_orderby time DESC ); -- 4. 创建连续聚合 CREATE MATERIALIZED VIEW telemetry_hourly WITH (timescaledb.continuous) AS SELECT time_bucket(1 hour, time) AS bucket, device_id, metric_name, COUNT(*) as sample_count, AVG(value) as avg_value, MAX(value) as max_value, MIN(value) as min_value, STDDEV(value) as std_value FROM telemetry WHERE status_code 200 GROUP BY bucket, device_id, metric_name; -- 5. 添加策略 -- 压缩策略3天前的数据压缩 SELECT add_compression_policy(telemetry, INTERVAL 3 days); -- 保留策略保留1年数据 SELECT add_retention_policy(telemetry, INTERVAL 365 days); -- 连续聚合刷新策略SELECT add_continuous_aggregate_policy(telemetry_hourly, start_offset INTERVAL 7 days, end_offset INTERVAL 5 minutes, schedule_interval INTERVAL 15 minutes); -- 6. 查询示例 -- 实时查询最近数据 SELECT * FROM telemetry WHERE device_id 1 AND metric_name temperature AND time NOW() - INTERVAL 1 hour ORDER BY time DESC; -- 聚合查询 SELECT time_bucket(1 day, bucket) as day, metric_name, AVG(avg_value) as daily_avg, MAX(max_value) as daily_max FROM telemetry_hourly WHERE device_id 1 AND bucket NOW() - INTERVAL 30 days GROUP BY day, metric_name ORDER BY day DESC; -- 异常检测 SELECT time_bucket(5 minutes, time) as interval, COUNT(*) as readings, AVG(value) as avg_value, CASE WHEN AVG(value) (SELECT AVG(value) 3*STDDEV(value) FROM telemetry WHERE metric_name temperature AND time NOW() - INTERVAL 7 days) THEN HIGH_ALERT WHEN AVG(value) (SELECT AVG(value)3*STDDEV(value) FROM telemetry WHERE metric_name temperature AND time NOW() - INTERVAL 7 days) THEN LOW_ALERT ELSE NORMAL END as status FROM telemetry WHERE metric_name temperature AND time NOW() - INTERVAL 1 hour GROUP BY interval;11. 故障排除与诊断-- 查看查询计划 EXPLAIN (ANALYZE, BUFFERS) SELECT * FROM conditions WHERE time NOW() - INTERVAL 1 day; -- 查看块状态 SELECT chunk_name, range_start, range_end, is_compressed, chunk_size FROM chunks_detailed_size(conditions) ORDER BY range_start; -- 检查数据分布 SELECT time_bucket(1 day, time) as day, COUNT(*) as row_count, pg_size_pretty(SUM(pg_column_size(temperature))) as temp_size FROM conditions GROUP BY day ORDER BY day DESC; -- 验证扩展状态 SELECT extname, extversion FROM pg_extension WHERE extname LIKE timescale%;参考来源Linux下用Docker安全部署OpenClawOpenClaw Skill 编写规范 与示例如何安全部署你的 AI 助手 OpenClawOpenClaw语法基础龙虾智能体核心命令快速上手附常用命令汇总OpenClaw - 免费跑通、接入钉钉到使用 ClawHub