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

PostgreSQL 时序数据案例 - 时间流逝, 自动压缩, 同比\环比

news 来源：原创 2024/5/10 21:40:10

2019独角兽企业重金招聘Python工程师标准>>>

背景

时序数据库一个重要的特性是时间流逝压缩，例如1天前压缩为5分钟一个点，7天前压缩为30分钟一个点。

PostgreSQL 压缩算法可定制。例如简单的平均值、最大值、最小值压缩，或者基于旋转门压缩算法的压缩。

《[未完待续] SQL流式案例 - 旋转门压缩(前后计算相关滑窗处理例子)》

《旋转门数据压缩算法在PostgreSQL中的实现 - 流式压缩在物联网、监控、传感器等场景的应用》

本文介绍一种简单压缩的场景，类似RRD数据库，按时间维度，压缩为平均值、最大值、最小值、总和、记录数等维度。

本文还介绍了窗口查询、同比、环比UDF(含KNN计算)、按时间分组均匀写入、等高级SQL用法。

设计

明细表

create table tbl (  
  id serial8 primary key,  -- 主键  
  sid int,                 -- 传感器ID  
  hid int,                 -- 指标ID  
  val float8,              -- 采集值  
  ts timestamp             -- 采集时间  
);   
  
create index idx_tbl on tbl(ts);

压缩表

1、5分钟级压缩表

create table tbl_5min (  
  id serial8 primary key,  -- 主键  
  sid int,                 -- 传感器ID  
  hid int,                 -- 指标ID  
  val float8,              -- 继承，平均值，方便做环比分析  
  ts timestamp,            -- 继承，开始时间，方便做环比分析  
  val_min float8,              -- 最小值  
  val_max float8,              -- 最大值  
  val_sum float8,              -- 和  
  val_count float8,            -- 采集次数  
  ts_start timestamp,      -- 区间开始时间  
  ts_end timestamp         -- 区间结束时间  
);   
  
alter table tbl_5min inherit tbl;

2、30分钟级压缩表

create table tbl_30min (  
  id serial8 primary key,  -- 主键  
  sid int,                 -- 传感器ID  
  hid int,                 -- 指标ID  
  val float8,              -- 继承，平均值，方便做环比分析  
  ts timestamp,            -- 继承，开始时间，方便做环比分析  
  val_min float8,              -- 最小值  
  val_max float8,              -- 最大值  
  val_sum float8,              -- 和  
  val_count float8,            -- 采集次数  
  ts_start timestamp,      -- 区间开始时间  
  ts_end timestamp         -- 区间结束时间  
);   
  
alter table tbl_30min inherit tbl;

3、5分钟级压缩语句

with tmp1 as (  
  delete from only tbl where ts <= now()-interval '1 day' returning *  
)  
insert into tbl_5min  
  (sid, hid, val, ts, val_min, val_max, val_sum, val_count, ts_start, ts_end)   
select sid, hid, avg(val) as val, min(ts) as ts, min(val) as val_min, max(val) as val_max, sum(val) as val_sum, count(*) as val_count, min(ts) as ts_start, max(ts) as ts_end from   
tmp1  
group by sid, hid, substring(to_char(ts, 'yyyymmddhh24mi'), 1, 10) || lpad(((substring(to_char(ts, 'yyyymmddhh24mi'), 11, 2)::int / 5) * 5)::text, 2, '0');

4、30分钟级压缩语句

with tmp1 as (  
  delete from only tbl_5min where ts_start <= now()-interval '1 day' returning *  
)  
insert into tbl_30min  
  (sid, hid, val_min, val_max, val_sum, val_count, ts_start, ts_end)  
select sid, hid, min(val_min) as val_min, max(val_max) as val_max, sum(val_sum) as val_sum, sum(val_count) as val_count, min(ts_start) as ts_start, max(ts_end) as ts_end from   
tmp1     
group by sid, hid, substring(to_char(ts_start, 'yyyymmddhh24mi'), 1, 10) || lpad(((substring(to_char(ts_start, 'yyyymmddhh24mi'), 11, 2)::int / 30) * 30)::text, 2, '0');

DEMO

1、写入明细测试数据，1亿条，分布到10天。

insert into tbl (sid, hid, val, ts) select random()*1000, random()*5, random()*100,   -- 1000个传感器，每个传感器5个指标。  
  now()-interval '10 day' + (id * ((10*24*60*60/100000000.0)||' sec')::interval)   -- 倒推10天为起点 + (id * 每条记录的耗时)  
from generate_series(1,100000000) t(id);

2、5分钟压缩调度，1天前的数据，每隔1小时调度一次以下SQL。

with tmp1 as (  
  delete from only tbl where ts <= now()-interval '1 day' returning *  
)  
insert into tbl_5min  
  (sid, hid, val, ts, val_min, val_max, val_sum, val_count, ts_start, ts_end)   
select sid, hid, avg(val) as val, min(ts) as ts, min(val) as val_min, max(val) as val_max, sum(val) as val_sum, count(*) as val_count, min(ts) as ts_start, max(ts) as ts_end from   
tmp1  
group by sid, hid, substring(to_char(ts, 'yyyymmddhh24mi'), 1, 10) || lpad(((substring(to_char(ts, 'yyyymmddhh24mi'), 11, 2)::int / 5) * 5)::text, 2, '0');

3、30分钟压缩调度，7天前的数据，每隔1天调度一次以下SQL。

with tmp1 as (  
  delete from only tbl_5min where ts_start <= now()-interval '1 day' returning *  
)  
insert into tbl_30min  
  (sid, hid, val_min, val_max, val_sum, val_count, ts_start, ts_end)  
select sid, hid, min(val_min) as val_min, max(val_max) as val_max, sum(val_sum) as val_sum, sum(val_count) as val_count, min(ts_start) as ts_start, max(ts_end) as ts_end from   
tmp1     
group by sid, hid, substring(to_char(ts_start, 'yyyymmddhh24mi'), 1, 10) || lpad(((substring(to_char(ts_start, 'yyyymmddhh24mi'), 11, 2)::int / 30) * 30)::text, 2, '0');

小结

1、根据interval取时间分组，用整型除法+乘法。

例子：

5分钟：

substring(to_char(ts, 'yyyymmddhh24mi'), 1, 10) || lpad(((substring(to_char(ts, 'yyyymmddhh24mi'), 11, 2)::int / 5) * 5)::text, 2, '0');

30分钟：

substring(to_char(ts_start, 'yyyymmddhh24mi'), 1, 10) || lpad(((substring(to_char(ts_start, 'yyyymmddhh24mi'), 11, 2)::int / 30) * 30)::text, 2, '0')

2、生成均匀分布的时序数据，使用PG的interval和generate_series，可以将写入时间均匀分配到对应区间。

insert into tbl (sid, hid, val, ts) select random()*1000, random()*5, random()*100,   -- 1000个传感器，每个传感器5个指标。  
  now()-interval '10 day' + (id * ((10*24*60*60/100000000.0)||' sec')::interval)   -- 倒推10天为起点 + (id * 每条记录的耗时)  
from generate_series(1,100000000) t(id);

3、时序数据库一个重要的特性是时间流逝压缩，例如1天前压缩为5分钟一个点，7天前压缩为30分钟一个点。

PostgreSQL 压缩算法可定制。例如简单的平均值、最大值、最小值压缩，或者基于旋转门压缩算法的压缩。

本文介绍了一种简单压缩的场景，类似RRD数据库，按时间维度，压缩为平均值、最大值、最小值、总和、记录数等维度。

加上调度即可：

《PostgreSQL 定时任务方法2》

《PostgreSQL Oracle 兼容性之 - DBMS_JOBS - Daily Maintenance - Timing Tasks(pgagent)》

4、压缩后包含区间、最大值、最小值、平均值、点数等值，可以用于绘制图形。

5、结合PG的窗口函数，很容易绘制同比、环比的图形，SQL例句：

索引，加速

create index idx_tbl_2 on tbl using btree (sid, hid, ts);  
create index idx_tbl_5min_2 on tbl_5min using btree (sid, hid, ts);  
create index idx_tbl_30min_2 on tbl_30min using btree (sid, hid, ts);

复合类型，返回环比值

create type tp as (id int8, sid int, hid int, val float8, ts timestamp);

获取环比值函数，返回指定SID,HID在某个时间点附近的一条记录，含KNN算法

create or replace function get_val(v_sid int, v_hid int, v_ts timestamp) returns tp as $$  
select t.tp from 
(
select 
(select (id, sid, hid, val, ts)::tp tp from only tbl where sid=1 and hid=1 and ts>= now() limit 1) 
union all 
select 
(select (id, sid, hid, val, ts)::tp tp from only tbl where sid=1 and hid=1 and ts< now() limit 1)
) t
order by (t.tp).ts limit 1;
$$ language sql strict;

同比、周环比、月环比(这些值也可以自动生成，避免每次查询时计算)：

select   
sid,   
hid,   
val,   
lag(val) over w1,                           -- 同比  
get_val(sid, hid, ts-interval '1 week'),    -- 周环比  
get_val(sid, hid, ts-interval '1 month')    -- 月环比  
  from tbl         -- where ...  ，时间区间打点。   
window w1 as (partition by sid, hid order by ts)   
;

6、结合PG的线性回归，可以绘制预测指标。以下为详细介绍的例子：

《PostgreSQL 多元线性回归 - 2 股票预测》

《在PostgreSQL中用线性回归分析linear regression做预测 - 例子2, 预测未来数日某股收盘价》

《PostgreSQL 线性回归 - 股价预测 1》

《在PostgreSQL中用线性回归分析(linear regression) - 实现数据预测》

7、将压缩表继承到明细表，方便开发的使用，不需要再写UNION的SQL，直接查明细表，即可得到所有数据(包括压缩数据)。