重新发现PostgreSQL之美 - 时序压缩算法之旋转门
digoal约 7 分钟背景
场景:
物联网、金融、游戏、社交、电商等时序场景.
挑战:
数据量超大, 即使采用列存、数据压缩方式还是很大.
PG 解决方案:
有损压缩, 自定义旋转门算法.
视频回放: https://www.bilibili.com/video/BV1Eg411g7Sx/
在物联网、监控、传感器、金融等应用领域，数据在时间维度上流式的产生，而且数据量非常庞大。
例如我们经常看到的性能监控视图，就是很多点在时间维度上描绘的曲线。
又比如金融行业的走势数据等等。
我们想象一下，如果每个传感器或指标每100毫秒产生1个点，一天就是864000个点。
而传感器或指标是非常多的，例如有100个传感器或指标，一天的量就接近一亿的量。
假设我们要描绘一个时间段的图形，这么多的点，渲染估计都要很久。
那么有没有好的压缩算法，即能保证失真度，又能很好的对数据进行压缩呢？
旋转门压缩原理
旋转门压缩算法（SDT）是一种直线趋势化压缩算法,其本质是通过一条由起点和终点确定的直线代替一系列连续数据点。
PostgreSQL旋转门压缩算法
通过旋转门算法的原理，可以了解到，有几个必要的输入项。
有x坐标和y坐标的点（如果是时间轴上的点，可以通过epoch转换成这种形式）
E，即门的宽度，起到了控制压缩失真度的作用
例子
创建测试表
create table t(id int, -- ID，可有可无
val numeric, -- 值（如传感器或金融行业的点值）
t timestamp -- 取值时间戳
);
插入10万条测试数据
insert into t select generate_series(1,100000), round((random()*100)::numeric, 2), clock_timestamp()+(generate_series(1,100000) || ' second')::interval ;
时间如何转换成X轴的数值，假设每1秒为X坐标的1个单位
test=> select (extract(epoch from t)-extract(epoch from first_value(t) over())) / 1 as x, -- 除以1秒为1个单位
val, t from t limit 100;
x | val | t
------------------+-------+----------------------------
0 | 31.79 | 2016-08-12 23:22:27.530318
1.00012493133545 | 18.23 | 2016-08-12 23:22:28.530443
2.00013494491577 | 5.14 | 2016-08-12 23:22:29.530453
3.00014090538025 | 90.25 | 2016-08-12 23:22:30.530459
4.00014686584473 | 8.17 | 2016-08-12 23:22:31.530465
5.00015187263489 | 97.43 | 2016-08-12 23:22:32.53047
6.00015807151794 | 17.41 | 2016-08-12 23:22:33.530476
7.00016307830811 | 0.23 | 2016-08-12 23:22:34.530481
8.00016903877258 | 84.67 | 2016-08-12 23:22:35.530487
编写实现螺旋门算法的函数
create or replace function f (
i_radius numeric, -- 压缩半径
i_time timestamp, -- 开始时间
i_interval_s numeric, -- 时间转换间隔 (秒，例如每5秒在坐标上表示1个单位间隔，则这里使用5)
query text, -- 需要进行旋转门压缩的数据, 例子 'select t, val from t where t>=%L order by t limit 100' ， select 子句必须固定， 必须按t排序
OUT o_val numeric, -- 值，纵坐标 y (跳跃点y)
OUT o_time timestamp, -- 时间，横坐标 x (跳跃点x)
OUT o_x numeric -- 跳跃点x, 通过 o_time 转换
)
returns setof record as $$
declare
v_time timestamp; -- 时间变量
v_x numeric; -- v_time 转换为v_x
v_val numeric; -- y坐标
v1_time timestamp; -- 前一点 时间变量
v1_x numeric; -- 前一点 v_time 转换为v_x
v1_val numeric; -- 前一点 y坐标
v_start_time numeric; -- 记录第一条的时间坐标, 用于计算x偏移量
v_rownum int8 := 0; -- 用于标记是否第一行
v_max_angle1 numeric; -- 最大上门夹角角度
v_max_angle2 numeric; -- 最大下门夹角角度
v_angle1 numeric; -- 上门夹角角度
v_angle2 numeric; -- 下门夹角角度
begin
for v_time , v_val in execute format(query, i_time)
LOOP
-- 第一行，第一个点，是实际要记录的点位
v_rownum := v_rownum + 1;
if v_rownum=1 then
v_start_time := extract(epoch from v_time);
v_x := 0;
o_val := v_val;
o_time := v_time;
o_x := v_x;
-- raise notice 'rownum=1 %, %', o_val,o_time;
return next; -- 返回第一个点
else
v_x := (extract(epoch from v_time) - v_start_time) / i_interval_s; -- 生成X坐标
SELECT 180-ST_Azimuth(
ST_MakePoint(o_x, o_val+i_radius), -- 门上点
ST_MakePoint(v_x, v_val) -- next point
)/(2*pi())*360 as degAz, -- 上夹角
ST_Azimuth(
ST_MakePoint(o_x, o_val-i_radius), -- 门下点
ST_MakePoint(v_x, v_val) -- next point
)/(2*pi())*360 As degAzrev -- 下夹角
INTO v_angle1, v_angle2; select GREATEST(v_angle1, v_max_angle1), GREATEST(v_angle2, v_max_angle2) into v_max_angle1, v_max_angle2;
if (v_max_angle1 + v_max_angle2) >= 180 then -- 找到四边形外的点位，输出上一个点，并从上一个点开始重新计算四边形
-- raise notice 'max1 %, max2 %', v_max_angle1 , v_max_angle2;
-- 复原
v_angle1 := 0;
v_max_angle1 := 0;
v_angle2 := 0;
v_max_angle2 := 0;
-- 门已完全打开，输出前一个点的值
o_val := v1_val;
o_time := v1_time;
v1_x := (extract(epoch from v1_time) - v_start_time) / i_interval_s; -- 生成前一个点的X坐标
o_x := v1_x;
-- 用新的门，与当前点计算新的夹角
SELECT 180-ST_Azimuth(
ST_MakePoint(o_x, o_val+i_radius), -- 门上点
ST_MakePoint(v_x, v_val) -- next point
)/(2*pi())*360 as degAz, -- 上夹角
ST_Azimuth(
ST_MakePoint(o_x, o_val-i_radius), -- 门下点
ST_MakePoint(v_x, v_val) -- next point
)/(2*pi())*360 As degAzrev -- 下夹角
INTO v_angle1, v_angle2;
select GREATEST(v_angle1, v_max_angle1), GREATEST(v_angle2, v_max_angle2) into v_max_angle1, v_max_angle2;
-- raise notice 'new max %, new max %', v_max_angle1 , v_max_angle2;
-- raise notice 'rownum<>1 %, %', o_val, o_time;
return next;
end if;
-- 记录当前值，保存作为下一个点的前点
v1_val := v_val;
v1_time := v_time;
end if;
END LOOP;
end;
$$ language plpgsql strict;
压缩测试
门宽为15，起始时间为'2016-08-12 23:22:27.530318'，每1秒表示1个X坐标单位。
test=>
select * from f (
15, -- 门宽度=15
'2016-08-12 23:22:27.530318', -- 开始时间
1, -- 时间坐标换算间隔，1秒
'select t, val from t where t>=%L order by t limit 100' -- query
); o_val | o_time | o_x
-------+----------------------------+------------------
18.23 | 2016-08-12 23:22:28.530443 | 0
5.14 | 2016-08-12 23:22:29.530453 | 1.00001287460327
90.25 | 2016-08-12 23:22:30.530459 | 2.00001883506775
......
83.21 | 2016-08-12 23:24:05.531001 | 97.0005609989166
(71 rows)
100个点，压缩成了71个点。
使用excel绘图，进行压缩前后的对比
上面是压缩后的数据绘图，下面是压缩前的数据绘图
红色标记的位置，就是通过旋转门算法压缩掉的数据。
失真度是可控的。
流式压缩的实现
本文略，其实也很简单，这个函数改一下，创建一个以数组为输入参数的函数。
以lambda的方式，实时的从流式输入的管道取数，并执行即可。
参考我的github.
方法1，阅后即焚：
流计算结果表（即有效位点），增加2个字段：PK于明细表关联，时间戳代表最后一条记录。下次阅后即焚从最后一个有效位点开始，从明细表继续消费。
方法2，直接在明细表上更新状态（点、当前记录是否可见）。
其他，所有涉及到中间计算结果的，都可以用类似方法实现：
计算当前记录时，更新计算结果到当前记录上（也就是通过直接更新明细表的流计算方法）。如果计算时需要用到上一条或者上若干条流计算结果，通过递归，或者使用UDF调用都很容易得到。
例子
create table 明细表 (
上报内容字段定义....,
中间结果字段....,
可见性字段....
);
小结
通过旋转门算法，对IT监控、金融、电力、水利等监控、物联网、等流式数据进行实时的压缩。
数据不需要从数据库LOAD出来即可在库内完成运算和压缩。
用户也可以根据实际的需求，进行流式的数据压缩，同样数据也不需要从数据库LOAD出来，在数据库端即可完成。
PostgreSQL的功能一如既往的强大，好用，快用起来吧.
欢迎关注我的github (https://github.com/digoal/blog) , 学习数据库不迷路.
近期正在写公开课材料, 未来将通过视频号推出.
12.16在杭州举办的<是否需要专业的时序数据库, PK活动>视频回放: