403
奔跑吧，大屏 - 時間+空間 實時四維數據透視
標簽
PostgreSQL , 遞歸查詢 , 大屏播報 , 最擁堵的路口 , 最旺的商鋪 , 某個區域最擁堵的廁所 , 數據透視 , 流式計算 , 時序數據
背景
隨著物聯網的發展，數據的產生越來越快。比如馬路的汽車通過數據，用戶的駐留數據，水紋傳感器的數據，電商的FEED數據，網約車的軌跡數據 等等。
這麼多的數據，用途也不一樣，有需要流式實時統計的，也有時序處理相關需求的，還有全量分析需求的。
這些都有對應的解決方案。
《流計算風雲再起 - PostgreSQL攜PipelineDB力挺IoT》
《時序數據庫有哪些特點? TimescaleDB時序數據庫介紹》
《ApsaraDB的左右互搏(PgSQL+HybridDB+OSS) - 解決OLTP+OLAP混合需求》
以指揮中心的大屏為例，有一些需求就很有意思，比如
1. 展示最新的傳感器TOP VALUE數據。
2. 選擇時間軸，按區域，展示人流量。
例子
1 輸出所有傳感器上報的最新值
這個需要和我之前寫的這個例子很相似。
《時序數據合並場景加速分析和實現 - 複合索引，窗口分組查詢加速，變態遞歸加速》
設計表結構，gid表示傳感器ID，val是上傳的值，crt_time是時間。
假設有1萬個傳感器，插入1000萬條記錄。
create unlogged table sort_test(
id serial8 primary key, -- 自增主鍵
s_id int, -- 傳感器ID
val numeric(10,2), -- 傳感器值
crt_time timestamp default clock_timestamp() -- 上報時間
);
寫入1000萬傳感器測試數據
postgres=# insert into sort_test (s_id,val) select random()*10000, random()*100 from generate_series(1,10000000);
INSERT 0 10000000
創建索引
postgres=# create index idx_test on sort_test using btree(s_id,id desc);
使用遞歸調用的方法，獲取所有傳感器的最新值(以每個傳感器的最大的自增ID為最新上報標記)
create type r as (s_id int, val numeric(10,2)); -- 複合類型
with recursive skip as (
(
select (s_id,val)::r as r from sort_test where id in (select id from sort_test where s_id is not null order by s_id,id desc limit 1)
)
union all
(
select (
select (s_id,val)::r as r from sort_test where id in (select id from sort_test t where t.s_id>(s.r).s_id and t.s_id is not null order by s_id,id desc limit 1)
) from skip s where (s.r).s_id is not null
) -- 這裏的where (s.r).s_id is not null 一定要加, 否則就死循環了.
)
select (t.r).s_id, (t.r).val from skip t where t.* is not null;
1000萬條記錄，篩選1萬條最新記錄，耗費時間：129毫秒。
為什麼能這麼快？因為用了遞歸，減少了掃描量和運算量。
s_id | val
-------+-------
0 | 83.55
1 | 91.62
2 | 72.70
3 | 45.46
4 | 99.97
5 | 17.04
6 | 8.96
7 | 25.83
8 | 28.10
9 | 26.19
10 | 83.03
11 | 1.30
......
Time: 128.779 ms
使用遊標則更快，一次獲取10條，僅花費0.36毫秒。
postgres=# begin;
BEGIN
Time: 0.095 ms
postgres=# declare cur cursor for with recursive skip as (
(
select (s_id,val)::r as r from sort_test where id in (select id from sort_test where s_id is not null order by s_id,id desc limit 1)
)
union all
(
select (
select (s_id,val)::r as r from sort_test where id in (select id from sort_test t where t.s_id>(s.r).s_id and t.s_id is not null order by s_id,id desc limit 1)
) from skip s where (s.r).s_id is not null
) -- 這裏的where (s.r).s_id is not null 一定要加, 否則就死循環了.
)
select (t.r).s_id, (t.r).val from skip t where t.* is not null;
DECLARE CURSOR
Time: 0.841 ms
postgres=# fetch 10 from cur;
s_id | val
------+-------
0 | 83.55
1 | 91.62
2 | 72.70
3 | 45.46
4 | 99.97
5 | 17.04
6 | 8.96
7 | 25.83
8 | 28.10
9 | 26.19
(10 rows)
Time: 0.364 ms
2 輸出某個城市的車流TOP 10路口
相比第一個例子，做了一次收斂，按VALUE排序，輸出最大的。
假設每個路口有傳感器不斷上報路口通過的車流數量。大屏展示通過量最大的10個路口。
為了測試方便，我這裏依舊使用第一個例子的數據，末尾加上。
postgres=# with recursive skip as (
(
select (s_id,val)::r as r from sort_test where id in (select id from sort_test where s_id is not null order by s_id,id desc limit 1)
)
union all
(
select (
select (s_id,val)::r as r from sort_test where id in (select id from sort_test t where t.s_id>(s.r).s_id and t.s_id is not null order by s_id,id desc limit 1)
) from skip s where (s.r).s_id is not null
) -- 這裏的where (s.r).s_id is not null 一定要加, 否則就死循環了.
)
select (t.r).s_id, (t.r).val from skip t where t.* is not null order by 2 desc limit 10;
s_id | val
------+-------
997 | 99.99
2233 | 99.97
610 | 99.97
4 | 99.97
6735 | 99.96
545 | 99.93
2992 | 99.91
4747 | 99.90
543 | 99.89
7229 | 99.88
(10 rows)
Time: 126.052 ms
1000萬條記錄，篩選1萬條最新記錄，輸出TOP 10，耗費時間：126毫秒。
3 某個區域，某個時間段，按鈕人流量輸出TOP 商鋪
相比前兩個例子，多了兩個維度：
一個是時間維度，用戶可以勾選時間段進行分析。另一個是區域維度，用戶要勾選地區，輸出地區內的數據。
思考：
空間索引不像B-TREE索引是有序存儲的，空間索引是GIST索引，使用了類似聚類分區的結構，因此在進行多列複合時，GIST的空間查詢結合索引排序輸出第一條，是行不通的，會引入顯示的SORT。
原理參考
《從難纏的模煳查詢聊開 - PostgreSQL獨門絕招之一 GIN , GiST , SP-GiST , RUM 索引原理與技術背景》
同時查詢條件包含了時間區間作為條件，索引非驅動列（子段gid+VAL）的排序也是行不通的。
什麼時候能使用複合索引的查詢+排序？
僅僅當排序列前麵的所有列都是等值查詢時，才能使用隱式排序，並且索引的順序要和排序的順序一致。例如index(a,b,c)支持where a=? and b=? order by c，但是不支持where a> ? and b=? order by c等等。
重新規劃測試數據，為了測試方便, 以point取代經緯度，真實業務可以使用geometry類型。
create table test (
id serial8 primary key, -- 自增序列
gid int, -- 商鋪ID
val int, -- 商鋪人流
pos point, -- 商鋪位置, 為了測試方便, 以point取代經緯度
crt_time timestamp -- 上傳時間
);
插入1000萬測試數據，1萬個店鋪ID，1億的點陣範圍中的隨機point。
postgres=# insert into test (gid,val,pos,crt_time) select random()*10000, random()*100000, point(random()*10000, random()*10000), clock_timestamp() from generate_series(1,10000000);
postgres=# select min(crt_time),max(crt_time) from test;
min | max
----------------------------+----------------------------
2017-04-13 20:04:18.969268 | 2017-04-13 20:04:54.578339
(1 row)
時間+空間 的快速傳感器最大值篩選怎麼加速呢？
分兩種情況優化
1. 總的傳感器（店鋪）不多（例如1萬個店鋪）
利用索引快速搜索每個GID的最大VAL，使用partial index，規避時間問題；使用CPU完成點麵判斷。
例子，
例如我們允許用戶勾選的最小時間範圍是2小時，可以每2小時建一個partial index。（使用這麼多partial index很變態，也不優雅。建議10.0的分區表優化後，每2小時切一個分區。）
create index idx_test_1 on test (gid, val desc) where crt_time between '2017-04-13 20:04:18.969268' and '2017-04-13 20:04:30.969268';
這個區間的總數據量, 約350萬。
postgres=# select count(*) from test where crt_time between '2017-04-13 20:04:18.969268' and '2017-04-13 20:04:30.969268';
count
---------
3461005
(1 row)
使用這個partial index，以及遞歸調用，取出該區間的所有店鋪的最大值。然後根據點麵判斷，得到某個區域的數據，再排序輸出TOP 10。
with recursive skip as (
(
select t0 from test t0 where id in
(select id from test where gid is not null and crt_time between '2017-04-13 20:04:18.969268' and '2017-04-13 20:04:30.969268' order by gid,val desc limit 1) -- 時間參數，取出最小GID的最大val。作為啟動記錄
)
union all
(
select (
select t1 from test t1 where id in (select id from test t where t.gid > (s.t0).gid and t.gid is not null
and crt_time between '2017-04-13 20:04:18.969268' and '2017-04-13 20:04:30.969268' -- 時間參數
order by gid,val desc limit 1)
) from skip s where (s.t0).gid is not null
) -- 這裏的where (s.t0).gid is not null 一定要加, 否則就死循環了.
)
select (t.t0).* from skip t where t.* is not null
and circle '((5000,5000), 1000)' @> (t.t0).pos -- 區域參數
order by (t.t0).val desc limit 10; -- 取出前十的店鋪
135毫秒返回
id | gid | val | pos | crt_time
---------+------+-------+-------------------------------------+----------------------------
1754353 | 4001 | 99997 | (4755.64117543399,5253.53815406561) | 2017-04-13 20:04:24.563999
600729 | 5874 | 99996 | (5507.96090625226,4394.04523000121) | 2017-04-13 20:04:20.851141
1137330 | 4248 | 99995 | (4332.14340358973,4383.84034205228) | 2017-04-13 20:04:22.575639
2609044 | 7209 | 99995 | (5809.22217573971,4967.18854177743) | 2017-04-13 20:04:27.328745
1330926 | 2834 | 99994 | (4153.9505450055,4986.64934188128) | 2017-04-13 20:04:23.197925
208578 | 3439 | 99994 | (4186.14753056318,5103.39797474444) | 2017-04-13 20:04:19.598547
703010 | 5736 | 99993 | (4913.89285307378,4628.21466382593) | 2017-04-13 20:04:21.178653
298380 | 7680 | 99992 | (4539.91844784468,4454.29485291243) | 2017-04-13 20:04:19.884725
996318 | 7658 | 99992 | (4462.14715018868,5504.16304729879) | 2017-04-13 20:04:22.122626
3120169 | 3261 | 99991 | (4814.33014851063,4505.81138487905) | 2017-04-13 20:04:28.98197
(10 rows)
Time: 135.480 ms
執行計劃如下
------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Limit (cost=937.82..937.83 rows=1 width=40) (actual time=147.241..147.243 rows=10 loops=1)
Output: ((t.t0).id), ((t.t0).gid), ((t.t0).val), ((t.t0).pos), ((t.t0).crt_time)
Buffers: shared hit=80066
CTE skip
-> Recursive Union (cost=1.00..935.54 rows=101 width=64) (actual time=0.037..141.284 rows=10002 loops=1)
Buffers: shared hit=80066
-> Nested Loop (cost=1.00..9.03 rows=1 width=64) (actual time=0.036..0.036 rows=1 loops=1)
Output: t0.*
Inner Unique: true
Buffers: shared hit=8
-> HashAggregate (cost=0.57..0.58 rows=1 width=8) (actual time=0.022..0.023 rows=1 loops=1)
Output: test.id
Group Key: test.id
Buffers: shared hit=4
-> Limit (cost=0.43..0.55 rows=1 width=16) (actual time=0.017..0.018 rows=1 loops=1)
Output: test.id, test.gid, test.val
Buffers: shared hit=4
-> Index Scan using idx_test_1 on public.test (cost=0.43..431864.13 rows=3461209 width=16) (actual time=0.017..0.017 rows=1 loops=1)
Output: test.id, test.gid, test.val
Index Cond: (test.gid IS NOT NULL)
Buffers: shared hit=4
-> Index Scan using test_pkey on public.test t0 (cost=0.43..8.45 rows=1 width=72) (actual time=0.012..0.012 rows=1 loops=1)
Output: t0.*, t0.id
Index Cond: (t0.id = test.id)
Buffers: shared hit=4
-> WorkTable Scan on skip s (cost=0.00..92.45 rows=10 width=32) (actual time=0.014..0.014 rows=1 loops=10002)
Output: (SubPlan 1)
Filter: ((s.t0).gid IS NOT NULL)
Rows Removed by Filter: 0
Buffers: shared hit=80058
SubPlan 1
-> Nested Loop (cost=1.20..9.22 rows=1 width=64) (actual time=0.013..0.013 rows=1 loops=10001)
Output: t1.*
Inner Unique: true
Buffers: shared hit=80058
-> HashAggregate (cost=0.76..0.77 rows=1 width=8) (actual time=0.009..0.009 rows=1 loops=10001)
Output: t_1.id
Group Key: t_1.id
Buffers: shared hit=40033
-> Limit (cost=0.43..0.75 rows=1 width=16) (actual time=0.008..0.008 rows=1 loops=10001)
Output: t_1.id, t_1.gid, t_1.val
Buffers: shared hit=40033
-> Index Scan using idx_test_1 on public.test t_1 (cost=0.43..369056.35 rows=1153736 width=16) (actual time=0.008..0.008 rows=1 loops=10001)
Output: t_1.id, t_1.gid, t_1.val
Index Cond: ((t_1.gid > (s.t0).gid) AND (t_1.gid IS NOT NULL))
Buffers: shared hit=40033
-> Index Scan using test_pkey on public.test t1 (cost=0.43..8.45 rows=1 width=72) (actual time=0.003..0.003 rows=1 loops=10000)
Output: t1.*, t1.id
Index Cond: (t1.id = t_1.id)
Buffers: shared hit=40025
-> Sort (cost=2.28..2.29 rows=1 width=40) (actual time=147.240..147.241 rows=10 loops=1)
Output: ((t.t0).id), ((t.t0).gid), ((t.t0).val), ((t.t0).pos), ((t.t0).crt_time)
Sort Key: ((t.t0).val) DESC
Sort Method: top-N heapsort Memory: 26kB
Buffers: shared hit=80066
-> CTE Scan on skip t (cost=0.00..2.27 rows=1 width=40) (actual time=0.252..147.138 rows=317 loops=1)
Output: (t.t0).id, (t.t0).gid, (t.t0).val, (t.t0).pos, (t.t0).crt_time
Filter: ((t.* IS NOT NULL) AND ('<(5000,5000),1000>'::circle @> (t.t0).pos))
Rows Removed by Filter: 9685
Buffers: shared hit=80066
Planning time: 0.508 ms
Execution time: 147.505 ms
(62 rows)
2. 店鋪很多，但是時間+空間收斂後，記錄數不多(比如幾百萬)
這種情況，可以考慮使用時間分區表。然後構建空間索引。
通過時間條件，定位到指定的分區，通過空間索引，篩選數據。對篩選後的數據，通過少量CPU計算得到TOP店鋪。
例子
2.1 將表按時間分區（例如每2小時一個分區，前麵有介紹為什麼這麼做）
略，我這裏假設每兩小時約1千萬數據。
2.2 創建空間索引
postgres=# create index idx_test_gist on test using gist(pos);
CREATE INDEX
2.3 透視
SQL中輸入時間條件時，PostgreSQL會自動鎖定到分區表，我這裏為了簡便，直接寫TEST表。
使用窗口查詢，得到TOP SQL
select * from
(
select row_number() over(partition by gid order by val desc) as rn, * from test
where
circle '((5000,5000), 1000)' @> pos -- 區域參數
) t
where rn = 1 -- 取出該區間內每個店鋪的最大值
order by val desc limit 10; -- 取出前十的店鋪
效率
rn | id | gid | val | pos | crt_time
----+---------+------+-------+-------------------------------------+----------------------------
1 | 7859807 | 2311 | 99999 | (4900.04640072584,4950.79724118114) | 2017-04-13 20:04:46.013424
1 | 4658616 | 3699 | 99999 | (5625.03716442734,5338.90711143613) | 2017-04-13 20:04:35.467025
1 | 1754353 | 4001 | 99997 | (4755.64117543399,5253.53815406561) | 2017-04-13 20:04:24.563999
1 | 6076598 | 4610 | 99997 | (5679.03681658208,4793.08029171079) | 2017-04-13 20:04:40.09587
1 | 6139261 | 4069 | 99997 | (5225.87833926082,4101.83480009437) | 2017-04-13 20:04:40.301817
1 | 600729 | 5874 | 99996 | (5507.96090625226,4394.04523000121) | 2017-04-13 20:04:20.851141
1 | 4281282 | 9720 | 99996 | (5036.95292398334,4731.64941649884) | 2017-04-13 20:04:34.237957
1 | 5579952 | 1503 | 99996 | (4271.09604235739,5250.28191972524) | 2017-04-13 20:04:38.469311
1 | 5310205 | 1317 | 99995 | (4439.0160869807,4796.70224711299) | 2017-04-13 20:04:37.590451
1 | 1137330 | 4248 | 99995 | (4332.14340358973,4383.84034205228) | 2017-04-13 20:04:22.575639
(10 rows)
Time: 633.342 ms
執行計劃
Limit (cost=39265.88..39265.91 rows=10 width=48) (actual time=730.704..730.706 rows=10 loops=1)
Output: t.rn, t.id, t.gid, t.val, t.pos, t.crt_time
Buffers: shared hit=317037, temp read=1921 written=1928
-> Sort (cost=39265.88..39266.01 rows=50 width=48) (actual time=730.702..730.703 rows=10 loops=1)
Output: t.rn, t.id, t.gid, t.val, t.pos, t.crt_time
Sort Key: t.val DESC
Sort Method: top-N heapsort Memory: 26kB
Buffers: shared hit=317037, temp read=1921 written=1928
-> Subquery Scan on t (cost=38939.80..39264.80 rows=50 width=48) (actual time=520.846..728.927 rows=10001 loops=1)
Output: t.rn, t.id, t.gid, t.val, t.pos, t.crt_time
Filter: (t.rn = 1)
Rows Removed by Filter: 303477
Buffers: shared hit=317037, temp read=1921 written=1928
-> WindowAgg (cost=38939.80..39139.80 rows=10000 width=48) (actual time=520.844..703.933 rows=313478 loops=1)
Output: row_number() OVER (?), test.id, test.gid, test.val, test.pos, test.crt_time
Buffers: shared hit=317037, temp read=1921 written=1928
-> Sort (cost=38939.80..38964.80 rows=10000 width=40) (actual time=520.837..594.505 rows=313478 loops=1)
Output: test.gid, test.val, test.id, test.pos, test.crt_time
Sort Key: test.gid, test.val DESC
Sort Method: external merge Disk: 15368kB
Buffers: shared hit=317037, temp read=1921 written=1928
-> Index Scan using idx_test_gist on public.test (cost=0.42..38275.42 rows=10000 width=40) (actual time=0.240..336.235 rows=313478 loops=1)
Output: test.gid, test.val, test.id, test.pos, test.crt_time
Index Cond: ('<(5000,5000),1000>'::circle @> test.pos)
Buffers: shared hit=317037
Planning time: 0.140 ms
Execution time: 734.226 ms
(27 rows)
內核層麵優化(空間GRID分區表的支持)
讓PostgreSQL支持空間GRID分區(實際上你現在就可以使用繼承來實現，觸發器中使用grid+mod判斷應該插入哪個分區)。
參考如下
《蜂巢的藝術與技術價值 - PostgreSQL PostGIS's hex-grid》
對於時間+空間維度的數據透視，可以創建空間grid分區 + 時間分區 二級分區。
檢索時，通過分區表直接過濾到目標子分區表。再通過btree索引，遞歸調用，篩選出每個店鋪在候選區間的峰值數據，最後加上少量CPU運算，得到TOP店鋪。
使用這種方法，時間+空間的四維數據透視，查詢效率可以進入100毫秒以內。
業務優化方法
1. 對於例子1和2，由於業務層麵取的都是最近的數據，曆史數據並不關心。除了使用遞歸優化，還有2種方法。
方法1，不記錄曆史，將插入換成插入或更新。使用這種方法，查詢sort_test得到的始終是最新的值。
create unlogged table sort_test(
s_id int primary key, -- 傳感器ID
val numeric(10,2), -- 傳感器值
crt_time timestamp default clock_timestamp() -- 上報時間
);
insert into sort_test(s_id,val,crt_time) values (?,?,?) on conflict (s_id) do update set val=excluded.val,crt_time=excluded.crt_time;
方法2，記錄曆史，同時記錄最新狀態。使用觸發器完成這項工作。
分解：
數據插入時，自動更新最後一條記錄。(寫入量和更新量同等)
例子
創建一個狀態表記錄最新狀態，創建一個觸發器，寫入曆史數據時，自動更新最新狀態表。
create unlogged table hist(
id serial8 primary key, -- 自增主鍵
s_id int, -- 傳感器ID
val numeric(10,2), -- 傳感器值
crt_time timestamp default clock_timestamp() -- 上報時間
);
create unlogged table hist_stat(
s_id int primary key, -- 傳感器ID
val numeric(10,2), -- 傳感器值
crt_time timestamp default clock_timestamp() -- 上報時間
);
create or replace function tg() returns trigger as $$
declare
begin
insert into hist_stat (s_id,val,crt_time) values (NEW.s_id,NEW.val,NEW.crt_time) on conflict (s_id) do update set val=excluded.val,crt_time=excluded.crt_time;
return null;
end;
$$ language plpgsql strict;
create trigger tg after insert on hist for each row execute procedure tg();
插入數據，自動更新到最新狀態
postgres=# insert into hist(s_id,val) values(1,1);
INSERT 0 1
postgres=# insert into hist(s_id,val) values(1,1);
INSERT 0 1
postgres=# insert into hist(s_id,val) values(1,1);
INSERT 0 1
postgres=# insert into hist(s_id,val) values(1,1);
INSERT 0 1
postgres=# insert into hist(s_id,val) values(1,1);
INSERT 0 1
postgres=# select * from hist;
id | s_id | val | crt_time
----+------+------+----------------------------
3 | 1 | 1.00 | 2017-04-13 22:23:25.165286
4 | 1 | 1.00 | 2017-04-13 22:23:26.23929
5 | 1 | 1.00 | 2017-04-13 22:23:26.646152
6 | 1 | 1.00 | 2017-04-13 22:23:26.991189
7 | 1 | 1.00 | 2017-04-13 22:23:27.376265
(5 rows)
postgres=# select * from hist_stat ;
s_id | val | crt_time
------+------+----------------------------
1 | 1.00 | 2017-04-13 22:23:27.376265
(1 row)
查詢時，直接查詢最新狀態表，連遞歸調用都省了。
postgres=# select * from hist_stat ;
s_id | val | crt_time
------+------+----------------------------
1 | 1.00 | 2017-04-13 22:23:27.376265
(1 row)
2. 對於例子3，由於分析的是曆史數據，而且分析維度是時間和空間兩個維度。
因此可以將其中一個維度作為分區，將數據打散，打散之後，對分區建立另一個維度的索引。
這樣的話，在查詢時，可以將數據盡量的收斂到更小的範圍。
空間和時間都支持分區。（空間分區建議使用網格化的表述，便於查找和定位分區）。
參考
《流計算風雲再起 - PostgreSQL攜PipelineDB力挺IoT》
《時序數據庫有哪些特點? TimescaleDB時序數據庫介紹》
《ApsaraDB的左右互搏(PgSQL+HybridDB+OSS) - 解決OLTP+OLAP混合需求》
《時序數據合並場景加速分析和實現 - 複合索引，窗口分組查詢加速，變態遞歸加速》
《從難纏的模煳查詢聊開 - PostgreSQL獨門絕招之一 GIN , GiST , SP-GiST , RUM 索引原理與技術背景》
《蜂巢的藝術與技術價值 - PostgreSQL PostGIS's hex-grid》
最後更新：2017-04-17 15:31:11