标签
PostgreSQL , 物联网 , feed , 网游 , 热力 , 商场驻留 , 人群分析 , 实时热力图 , 实时在线图 , 实时分段最大最小区间图 , 任意滑动窗口实时最高、最低在线数
背景
在现实生活中，经常有聚集分析的需求。
例如：
某个商场，每个时间点，商场的每个商铺位置的人群驻留数量。（有技术手段可以感知人的驻留位置，当走进某个区域时，将写入一条记录，表示你进入了这个区域，离开时记录一条离开的记录，如果长时间不动，则定时写心跳记录）。
某个网游，每个时间点，在线人数。（上线时写一条上线记录，下线时写一条下线记录。）
某个共享单车公司，每个时间点，在线和不在线的车辆数量。（借车时写一条上线记录，还车时写一条下线记录。同时每隔一段时间询问车辆状态。）
某个物联网企业，每个分钟单位内，最小、最大在线传感器的数量。（传感器上线时写一条上线记录，下线时写一条下线记录，同时每隔一段时间询问传感器状态。）
这种属于非常典型的FEED应用，要求输出每个时间点这个世界（系统）的在线数。（如果按时间段输出，则输出每个时间段内的最大，最小在线数，实际上就是取range的边界）。
设计
场景：
某个物联网企业，有一些传感器，传感器上线时写一条上线记录，下线时写一条下线记录，同时每隔小时询问传感器状态，也就是说1小时内没有记录的传感器视为不在线。
企业需要统计每个分钟单位内，最小、最大在线传感器的数量。
1、表结构
create table sensor_stat(
sid int, -- 传感器ID
state boolean, -- 传感器状态，true在线，false离线
crt_time timestamp -- 状态上传时间
);
2、索引
create index idx_sensor_stat_1 on sensor_stat(sid, crt_time desc);
写入1.101亿测试数据（我们假设这是1小时的数据写入量，全天写入26.424亿记录），1001个传感器ID。
insert into sensor_stat select random()*1000, (random()*1)::int::boolean, clock_timestamp() from generate_series(1,110100000);
3、数据TTL，确保表比较瘦，只包含心跳时间范围内的数据。
由于每小时接收心跳，所以1小时内，必有数据，没有数据的传感器不计状态。因此我们保留1小时内的状态即可。
一种保留方法是pipelinedb，用法如下。
另一种保留方法，使用两张表，轮询使用即可。
create table sensor_stat1 (
sid int, -- 传感器ID
state boolean, -- 传感器状态，true在线，false离线
crt_time timestamp -- 状态上传时间
);
create table sensor_stat2 (
sid int, -- 传感器ID
state boolean, -- 传感器状态，true在线，false离线
crt_time timestamp -- 状态上传时间
);
类似的用法如下
《PostgreSQL 数据rotate用法介绍 - 按时间覆盖历史数据》
4、使用递归查询，高效查询传感器的最终状态
with recursive t as
(
(
select sensor_stat as sensor_stat from sensor_stat order by sid, crt_time desc limit 1
)
union all
(
select (select t1 from sensor_stat AS t1 where t1.sid>(t.sensor_stat).sid order by sid, crt_time desc limit 1) from t where (t.sensor_stat).sid is not null
)
)
select (t.sensor_stat).* from t where t.* is not null;
执行计划如下
explain (analyze,verbose,timing,costs,buffers) with recursive t as
(
(
select sensor_stat as sensor_stat from sensor_stat where state is true order by sid, crt_time desc limit 1
)
union all
(
select (select t1 from sensor_stat AS t1 where t1.sid>(t.sensor_stat).sid and t1.state is true order by sid, crt_time desc limit 1) from t where (t.sensor_stat).sid is not null
)
)
select (t.sensor_stat).* from t where t.* is not null;
QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
CTE Scan on t (cost=70.86..72.88 rows=100 width=13) (actual time=0.037..10.975 rows=1001 loops=1)
Output: (t.sensor_stat).sid, (t.sensor_stat).state, (t.sensor_stat).crt_time
Filter: (t.* IS NOT NULL)
Rows Removed by Filter: 1
Buffers: shared hit=5926
CTE t
-> Recursive Union (cost=0.57..70.86 rows=101 width=37) (actual time=0.030..10.293 rows=1002 loops=1)
Buffers: shared hit=5926
-> Subquery Scan on "*SELECT* 1" (cost=0.57..0.63 rows=1 width=37) (actual time=0.029..0.029 rows=1 loops=1)
Output: "*SELECT* 1".sensor_stat
Buffers: shared hit=5
-> Limit (cost=0.57..0.62 rows=1 width=49) (actual time=0.028..0.028 rows=1 loops=1)
Output: sensor_stat.*, sensor_stat.sid, sensor_stat.crt_time
Buffers: shared hit=5
-> Index Scan using idx_sensor_stat_1 on public.sensor_stat (cost=0.57..3180100.70 rows=55369290 width=49) (actual time=0.027..0.027 rows=1 loops=1)
Output: sensor_stat.*, sensor_stat.sid, sensor_stat.crt_time
Filter: (sensor_stat.state IS TRUE)
Buffers: shared hit=5
-> WorkTable Scan on t t_1 (cost=0.00..6.82 rows=10 width=32) (actual time=0.010..0.010 rows=1 loops=1002)
Output: (SubPlan 1)
Filter: ((t_1.sensor_stat).sid IS NOT NULL)
Rows Removed by Filter: 0
Buffers: shared hit=5921
SubPlan 1
-> Limit (cost=0.57..0.66 rows=1 width=49) (actual time=0.009..0.009 rows=1 loops=1001)
Output: t1.*, t1.sid, t1.crt_time
Buffers: shared hit=5921
-> Index Scan using idx_sensor_stat_1 on public.sensor_stat t1 (cost=0.57..1746916.71 rows=18456430 width=49) (actual time=0.009..0.009 rows=1 loops=1001)
Output: t1.*, t1.sid, t1.crt_time
Index Cond: (t1.sid > (t_1.sensor_stat).sid)
Filter: (t1.state IS TRUE)
Rows Removed by Filter: 1
Buffers: shared hit=5921
Planning time: 0.180 ms
Execution time: 11.083 ms
(35 rows)
样例
sid | state | crt_time
------+-------+----------------------------
0 | t | 2017-07-05 10:29:09.470687
1 | f | 2017-07-05 10:29:09.465721
2 | t | 2017-07-05 10:29:09.474216
3 | f | 2017-07-05 10:29:09.473176
4 | t | 2017-07-05 10:29:09.473179
5 | t | 2017-07-05 10:29:09.473842
......
996 | t | 2017-07-05 10:29:09.469787
997 | f | 2017-07-05 10:29:09.470983
998 | t | 2017-07-05 10:29:09.47268
999 | t | 2017-07-05 10:29:09.469192
1000 | t | 2017-07-05 10:29:09.472195
(1001 rows)
Time: 11.067 ms
效率很高，1.101亿数据，11毫秒获取最终在线状态。
在线的设备为state=t的。
with recursive t as
(
(
select sensor_stat as sensor_stat from sensor_stat order by sid, crt_time desc limit 1
)
union all
(
select (select t1 from sensor_stat AS t1 where t1.sid>(t.sensor_stat).sid order by sid, crt_time desc limit 1) from t where (t.sensor_stat).sid is not null
)
)
select count(*) from t where t.* is not null and (t.sensor_stat).state is true;
count
-------
491
(1 row)
Time: 10.182 ms
5、统计任意时间点的传感器在线数量，如果每个设备上线的时间精确到秒(crt_time精确到秒)，那么不管有多少条记录，一天最多需要统计86400个时间点的传感器在线数量。
例如统计 2017-07-05 10:29:09 时间点的传感器在线数量，加一个时间限制即可。
with recursive t as
(
(
select sensor_stat as sensor_stat from sensor_stat where crt_time <= '2017-07-05 10:29:09' order by sid, crt_time desc limit 1
)
union all
(
select (select t1 from sensor_stat AS t1 where t1.crt_time <= '2017-07-05 10:29:09' and t1.sid>(t.sensor_stat).sid order by sid, crt_time desc limit 1) from t where (t.sensor_stat).sid is not null
)
)
select count(*) from t where t.* is not null and (t.sensor_stat).state is true;
count
-------
501
(1 row)
Time: 20.743 ms
新增这个时间限制，会带来一定的性能影响，特别是如果这个时间是过去很久以前的时间，过滤会越多，性能下降越严重。
因此，建议实时，每秒发起一次查询请求，就不要加这个时间限制了。
6、一次性生成过去每一秒的在线数。
使用窗口查询的帧查询技术。（帧表示按时间排序，截止到当前记录的区间。）
7、统计每分钟内，最高在线数、最低在线数。
每秒查询一次，将数据写入结果表。
create table result (crt_time timestamp(0) default now(), state boolean, cnt int);
create index idx_result_1 on result using brin (crt_time);
insert into result (state,cnt)
with recursive t as
(
(
select sensor_stat as sensor_stat from sensor_stat order by sid, crt_time desc limit 1
)
union all
(
select (select t1 from sensor_stat AS t1 where t1.sid>(t.sensor_stat).sid order by sid, crt_time desc limit 1) from t where (t.sensor_stat).sid is not null
)
)
select (t.sensor_stat).state, count(*) from t where t.* is not null group by 1;
INSERT 0 2
Time: 12.061 ms
postgres=# select * from result ;
crt_time | state | cnt
---------------------+-------+-----
2017-07-05 11:11:03 | f | 510
2017-07-05 11:11:03 | t | 491
(2 rows)
Time: 0.274 ms
由于每次查询仅需12毫秒，每秒调用一次没有问题。
统计某一分钟内，最高在线数、最低在线数。
select '2017-07-05 11:11:00', min(cnt), max(cnt) from result where crt_time between '2017-07-05 11:11:00' and '2017-07-05 11:12:00';
or
select to_char(crt_time, 'yyyy-mm-dd hh24:mi:00'), min(cnt), max(cnt) from result where crt_time between ? and ? group by 1;
传感器ID很多很多时，如何优化
当传感器ID达到10万级别时，查询性能会下降到250毫秒。
如果传感器ID特别多，例如有百万以上，那么会下降到2.5秒。就不适合每秒查询一次了。
因此传感器数量特别多时，如何优化？
有一个比较好的方法是数据按传感器ID进行哈希分布，例如每张分区表负责1万个传感器ID。在查询在线数时，并发的查询所有的分区表，从而降低RT。
小结
使用本文提到的方法(递归查询)，我们可以实现非常细粒度的，大量被跟踪物的状态实时统计。
用于绘制被跟踪物的实时状态图，例如：
1、实时热力图
2、实时传感器(或用户)在线、离线数，任意滑动窗口的最大最小在线、离线值。
