本文共 8423 字，大约阅读时间需要 28 分钟。
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 -- 状态上传时间 );
类似的用法如下
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达到10万级别时，查询性能会下降到250毫秒。
如果传感器ID特别多，例如有百万以上，那么会下降到2.5秒。就不适合每秒查询一次了。
因此传感器数量特别多时，如何优化？
有一个比较好的方法是数据按传感器ID进行哈希分布，例如每张分区表负责1万个传感器ID。在查询在线数时，并发的查询所有的分区表，从而降低RT。
使用本文提到的方法(递归查询)，我们可以实现非常细粒度的，大量被跟踪物的状态实时统计。
用于绘制被跟踪物的实时状态图，例如：
1、实时热力图
2、实时传感器(或用户)在线、离线数，任意滑动窗口的最大最小在线、离线值。
转载地址：http://efufm.baihongyu.com/