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

本文涉及的产品
云原生数据库 PolarDB PostgreSQL 版,企业版 4核16GB
推荐场景:
HTAP混合负载
云数据库 RDS MySQL,集群版 2核4GB 100GB
推荐场景:
搭建个人博客
云原生数据库 PolarDB MySQL 版,Serverless 5000PCU 100GB
简介:

标签

PostgreSQL , 时序数据 , rrd , rrdtool , round robin database , 自动压缩 , CTE , dml returning , 环比 , 同比 , KNN


背景

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

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

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

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

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

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

设计

pic

明细表

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,直接查明细表,即可得到所有数据(包括压缩数据)。

相关案例

《超时流式处理 - 没有消息流入的数据异常监控》

《阿里云RDS PostgreSQL varbitx实践 - 流式标签 (阅后即焚流式批量计算) - 万亿级,任意标签圈人,毫秒响应》

《PostgreSQL 流式统计 - insert on conflict 实现 流式 UV(distinct), min, max, avg, sum, count ...》

《HTAP数据库 PostgreSQL 场景与性能测试之 32 - (OLTP) 高吞吐数据进出(堆存、行扫、无需索引) - 阅后即焚(JSON + 函数流式计算)》

《HTAP数据库 PostgreSQL 场景与性能测试之 31 - (OLTP) 高吞吐数据进出(堆存、行扫、无需索引) - 阅后即焚(读写大吞吐并测)》

《HTAP数据库 PostgreSQL 场景与性能测试之 27 - (OLTP) 物联网 - FEED日志, 流式处理 与 阅后即焚 (CTE)》

《基于PostgreSQL的流式PipelineDB, 1000万/s实时统计不是梦》

参考

1、https://en.wikipedia.org/wiki/RRDtool

2、https://oss.oetiker.ch/rrdtool/

RRD is round robin database.

RRDtool is the OpenSource industry standard, high performance data logging and graphing system for time series data.

RRDtool can be easily integrated in shell scripts, perl, python, ruby, lua or tcl applications.

3、https://www.postgresql.org/docs/10/static/sql-expressions.html#SYNTAX-WINDOW-FUNCTIONS

https://www.postgresql.org/docs/10/static/functions-window.html

相关实践学习
使用PolarDB和ECS搭建门户网站
本场景主要介绍基于PolarDB和ECS实现搭建门户网站。
阿里云数据库产品家族及特性
阿里云智能数据库产品团队一直致力于不断健全产品体系,提升产品性能,打磨产品功能,从而帮助客户实现更加极致的弹性能力、具备更强的扩展能力、并利用云设施进一步降低企业成本。以云原生+分布式为核心技术抓手,打造以自研的在线事务型(OLTP)数据库Polar DB和在线分析型(OLAP)数据库Analytic DB为代表的新一代企业级云原生数据库产品体系, 结合NoSQL数据库、数据库生态工具、云原生智能化数据库管控平台,为阿里巴巴经济体以及各个行业的企业客户和开发者提供从公共云到混合云再到私有云的完整解决方案,提供基于云基础设施进行数据从处理、到存储、再到计算与分析的一体化解决方案。本节课带你了解阿里云数据库产品家族及特性。
相关文章
|
17天前
|
关系型数据库 MySQL 分布式数据库
PolarDB操作报错合集之在进行批量导出数据时,如何过滤掉视图并只导出表
在使用阿里云的PolarDB(包括PolarDB-X)时,用户可能会遇到各种操作报错。下面汇总了一些常见的报错情况及其可能的原因和解决办法:1.安装PolarDB-X报错、2.PolarDB安装后无法连接、3.PolarDB-X 使用rpm安装启动卡顿、4.PolarDB执行UPDATE/INSERT报错、5.DDL操作提示“Lock conflict”、6.数据集成时联通PolarDB报错、7.编译DN报错(RockyLinux)、8.CheckStorage报错(源数据库实例被删除)、9.嵌套事务错误(TDDL-4604)。
|
28天前
|
消息中间件 Java 关系型数据库
实时计算 Flink版操作报错合集之从 PostgreSQL 读取数据并写入 Kafka 时,遇到 "initial slot snapshot too large" 的错误,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
923 0
|
1月前
|
DataWorks 安全 关系型数据库
DataWorks产品使用合集之使用Flink CDC读取PostgreSQL数据时如何指定编码格式
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
|
12天前
|
关系型数据库 MySQL 数据库
关系型数据库mysql数据增量恢复
【7月更文挑战第3天】
126 2
|
12天前
|
关系型数据库 MySQL Shell
关系型数据库mysql数据完全恢复
【7月更文挑战第3天】
83 2
|
13天前
|
存储 关系型数据库 MySQL
|
19天前
|
关系型数据库 MySQL 分布式数据库
PolarDB产品使用问题之从MySQL迁移数据到PolarDB-X时,自定义函数不会自动迁移,该怎么办
PolarDB产品使用合集涵盖了从创建与管理、数据管理、性能优化与诊断、安全与合规到生态与集成、运维与支持等全方位的功能和服务,旨在帮助企业轻松构建高可用、高性能且易于管理的数据库环境,满足不同业务场景的需求。用户可以通过阿里云控制台、API、SDK等方式便捷地使用这些功能,实现数据库的高效运维与持续优化。
|
19天前
|
关系型数据库 MySQL 分布式数据库
PolarDB产品使用问题之要验证MySQL迁移后的数据库数据与迁移前的数据一致性,该怎么办
PolarDB产品使用合集涵盖了从创建与管理、数据管理、性能优化与诊断、安全与合规到生态与集成、运维与支持等全方位的功能和服务,旨在帮助企业轻松构建高可用、高性能且易于管理的数据库环境,满足不同业务场景的需求。用户可以通过阿里云控制台、API、SDK等方式便捷地使用这些功能,实现数据库的高效运维与持续优化。
PolarDB产品使用问题之要验证MySQL迁移后的数据库数据与迁移前的数据一致性,该怎么办
|
21天前
|
关系型数据库 MySQL 数据库
关系型数据库MySQL开发要点之多表设计案例详解代码实现
关系型数据库MySQL开发要点之多表设计案例详解代码实现
30 2
|
24天前
|
存储 关系型数据库 MySQL
关系型数据库mysql数据文件存储
【6月更文挑战第15天】
22 4

相关产品

  • 云原生数据库 PolarDB