标签
PostgreSQL , PostGIS , 快递 , 包裹侠 , 地理位置 , 距离排序 , KNN
背景
物流行业对地理位置信息数据的处理有非常强烈的需求，例如
1. 实时跟踪快递员、货车的位置信息。对数据库的写入性能要求较高。
2. 对于当日件，需要按发货位置，实时召回附近的快递员。
3. 实时的位置信息非常庞大，为了数据分析的需求，需要保留数据，所以需要廉价的存储。例如对象存储。同时还需要和数据库或分析型的数据库产品实现联动。
阿里云的 PostgreSQL + HybridDB for PostgreSQL + OSS 对象存储可以很好的满足这个需求，详细的方案如下。
业务描述
以物流配送场景为例，介绍阿里云的解决方案。
数据量
快递员：百万级。
快递员的轨迹定位数据间隔：5秒。
一个快递员每天工作时间 7 ~ 19点 (12个小时)。
一个快递员一天产生8640条记录。
所有的快递员，全网一天产生86.4亿条记录。
业务需求
1. 绘制快递员轨迹（实时）
2. 召回快递员（实时）
当天件的需求。
表结构设计
一、轨迹表设计
主表
按快递员ID哈希，128张表。
(如果不分区，单表存储86.4亿记录，问题也不大，只是导出到OSS对象存储的过程可能比较长，如果OSS出现故障，再次导出又要很久)
另一方面的好处是便于扩容。
create table cainiao (
uid int, -- 快递员ID
pos point, -- 快递员位置
crt_date date, -- 日期
crt_time time(0) -- 时间
);
do language plpgsql $$
declare
sql text;
begin
for i in 0..127
loop
sql := format( 'create table %I (like cainiao)' , 'cainiao_'||i );
execute sql;
end loop;
end;
$$;
子表
每天1张子表，轮询使用，覆盖到周(便于维护, 导出到OSS后直接truncate)。一共7张子表。
do language plpgsql $$
declare
sql text;
begin
for i in 0..127
loop
for x in 0..6
loop
sql := format( 'create table %I (like cainiao)' , 'cainiao_'||i||'_'||x );
execute sql;
end loop;
end loop;
end;
$$;
历史轨迹存储
OSS对象存储。
阿里云PostgreSQL有oss_ext插件，可以将数据写入oss对象存储。同时也支持从oss对象存储读取数据（外部表的方式），对用户透明。
详见
https://help.aliyun.com/document_detail/44461.html
10.0分区表的例子(可选)
PostgreSQL 10.0 内置了分区表，所以以上分区，可以直接读写主表。
《PostgreSQL 10.0 preview 功能增强 - 内置分区表》
9.5以及以上版本，建议使用pg_pathman插件，一样可以达到分区表的目的。
《PostgreSQL 9.5+ 高效分区表实现 - pg_pathman》
分区表例子
create table cainiao (
uid int,
pos point,
crt_date date,
crt_time time(0)
)
PARTITION BY RANGE(crt_time);
do language plpgsql $$
declare
sql text;
begin
for i in 0..23
loop
if i<>23 then
sql := format( 'create table %I PARTITION OF cainiao FOR VALUES FROM (%L) TO (%L)' , 'cainiao_'||lpad(i::text, 2, '0') , (lpad(i::text, 2, '0')||':00:00') , (lpad((i+1)::text, 2, '0')||':00:00') );
else
sql := format( 'create table %I PARTITION OF cainiao FOR VALUES FROM (%L) TO (unbounded)' , 'cainiao_'||lpad(i::text, 2, '0') , (lpad(i::text, 2, '0')||':00:00') );
end if;
execute sql;
end loop;
end;
$$;
postgres=# \d+ cainiao
Table "public.cainiao"
Column | Type | Collation | Nullable | Default | Storage | Stats target | Description
----------+---------------------------+-----------+----------+---------+---------+--------------+-------------
uid | integer | | | | plain | |
pos | point | | | | plain | |
crt_date | date | | | | plain | |
crt_time | time(0) without time zone | | not null | | plain | |
Partition key: RANGE (crt_time)
Partitions: cainiao_00 FOR VALUES FROM ('00:00:00') TO ('01:00:00'),
cainiao_01 FOR VALUES FROM ('01:00:00') TO ('02:00:00'),
cainiao_02 FOR VALUES FROM ('02:00:00') TO ('03:00:00'),
cainiao_03 FOR VALUES FROM ('03:00:00') TO ('04:00:00'),
cainiao_04 FOR VALUES FROM ('04:00:00') TO ('05:00:00'),
cainiao_05 FOR VALUES FROM ('05:00:00') TO ('06:00:00'),
cainiao_06 FOR VALUES FROM ('06:00:00') TO ('07:00:00'),
cainiao_07 FOR VALUES FROM ('07:00:00') TO ('08:00:00'),
cainiao_08 FOR VALUES FROM ('08:00:00') TO ('09:00:00'),
cainiao_09 FOR VALUES FROM ('09:00:00') TO ('10:00:00'),
cainiao_10 FOR VALUES FROM ('10:00:00') TO ('11:00:00'),
cainiao_11 FOR VALUES FROM ('11:00:00') TO ('12:00:00'),
cainiao_12 FOR VALUES FROM ('12:00:00') TO ('13:00:00'),
cainiao_13 FOR VALUES FROM ('13:00:00') TO ('14:00:00'),
cainiao_14 FOR VALUES FROM ('14:00:00') TO ('15:00:00'),
cainiao_15 FOR VALUES FROM ('15:00:00') TO ('16:00:00'),
cainiao_16 FOR VALUES FROM ('16:00:00') TO ('17:00:00'),
cainiao_17 FOR VALUES FROM ('17:00:00') TO ('18:00:00'),
cainiao_18 FOR VALUES FROM ('18:00:00') TO ('19:00:00'),
cainiao_19 FOR VALUES FROM ('19:00:00') TO ('20:00:00'),
cainiao_20 FOR VALUES FROM ('20:00:00') TO ('21:00:00'),
cainiao_21 FOR VALUES FROM ('21:00:00') TO ('22:00:00'),
cainiao_22 FOR VALUES FROM ('22:00:00') TO ('23:00:00'),
cainiao_23 FOR VALUES FROM ('23:00:00') TO (UNBOUNDED)
二、实时位置表
实时位置表，记录快递员的实时位置（最后一条记录的位置）。
由于快递员的位置数据会不停的汇报，因此实时位置表的数据不需要持久化，可以使用unlogged table。
注意
（假如快递员的位置不能实时上报，那么请使用非unlogged table。）
create unlogged table cainiao_trace_realtime (
uid int primary key, -- 快递员ID
pos point, -- 快递员位置
crt_date date, -- 日期
crt_time time(0) -- 时间
);
位置字段，创建GIST空间索引。
create index idx_cainiao_trace_realtime_pos on cainiao_trace_realtime using gist (pos);
实时位置更新逻辑设计
为了实时更新快递员的位置，可以设置一个触发器，在快递员上传实时位置时，自动更新最后的位置。
注意
（如果实时位置表cainiao_trace_realtime使用了非unlogged table，那么考虑到(写入+update)的RT会升高一些，建议不要使用触发器来更新位置。建议程序将 插入和update 作为异步调用进行处理。例如在收到快递员上报的批量位置轨迹后，拆分为batch insert以及update 一次。）
（batch insert: insert into cainiao values (),(),(),....; update 最终状态: update cainiao_trace_realtime set xx=xx where uid=xx;）(好处：1. 插入和更新异步, 2. 插入批量执行, 3. 整体rt更低)
jdbc batch参考: 《PostgreSQL jdbc batch insert》
create or replace function ins_cainiao() returns trigger as $$
declare
begin
insert into cainiao_trace_realtime(uid,pos,crt_date,crt_time)
values (NEW.uid, NEW.pos, NEW.crt_date, NEW.crt_time)
on conflict (uid) do update set pos=excluded.pos,crt_date=excluded.crt_date,crt_time=excluded.crt_time;
return null;
end;
$$ language plpgsql strict;
对基表添加触发器
do language plpgsql $$
declare
sql text;
begin
for i in 0..127
loop
for x in 0..6
loop
sql := format( 'create trigger tg after insert on %I for each row execute procedure ins_cainiao()', 'cainiao_'||i||'_'||x );
execute sql;
end loop;
end loop;
end;
$$;
触发器示例如下
postgres=# \d+ cainiao_0_0
Table "public.cainiao_0_0"
Column | Type | Collation | Nullable | Default | Storage | Stats target | Description
----------+---------------------------+-----------+----------+---------+---------+--------------+-------------
uid | integer | | | | plain | |
pos | point | | | | plain | |
crt_date | date | | | | plain | |
crt_time | time(0) without time zone | | | | plain | |
Triggers:
tg AFTER INSERT ON cainiao_0_0 FOR EACH ROW EXECUTE PROCEDURE ins_cainiao()
性能测试
说明
1. 本文假设应用程序会根据 快递员UID ，时间字段 拼接出基表的表名。
否则就需要使用PostgreSQL的分区表功能（分区表的性能比直接操作基表差一些）。
2. 本文使用point代替经纬度，因为point比较好造数据，方便测试。
实际上point和经纬度都是地理位置类型，可以实现的场景类似。性能指标也可以用于参考。
1 实时轨迹测试
模拟快递员实时的上传轨迹，实时的更新快递员的最新位置。
pgbench的测试脚本如下
vi test.sql
\set uid random(1,1000000)
\set x random(-500000,500000)
\set y random(-500000,500000)
insert into cainiao_0_2 values (:uid, point(:x,:y), now()::date, now()::time);
开始测试，持续300秒。
numactl --physcpubind=0-31 pgbench -M prepared -n -r -P 1 -f ./test.sql -c 32 -j 32 -T 300
测试结果
每秒写入17.4万，单次请求延迟0.18毫秒。
transaction type: ./test.sql
scaling factor: 1
query mode: prepared
number of clients: 32
number of threads: 32
duration: 300 s
number of transactions actually processed: 52270642
latency average = 0.184 ms
latency stddev = 2.732 ms
tps = 174234.709260 (including connections establishing)
tps = 174236.529998 (excluding connections establishing)
script statistics:
- statement latencies in milliseconds:
0.001 \set uid random(1,1000000)
0.000 \set x random(-500000,500000)
0.000 \set y random(-500000,500000)
0.182 insert into cainiao_0_2 values (:uid, point(:x,:y), now()::date, now()::time);
2 召回快递员测试
比如当日件达到一定数量、或者到达一定时间点时，需要召回附近的快递员取件。
或者当用户寄当日件时，需要召回附近的快递员取件。
压测用例
随机选择一个点，召回半径为20000范围内，距离最近的100名快递员。
SQL样例
postgres=# explain (analyze,verbose,timing,costs,buffers) select * from cainiao_trace_realtime where circle '((0,0),20000)' @> pos order by pos <-> point '(0,0)' limit 100;
QUERY PLAN
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Limit (cost=0.41..112.45 rows=100 width=40) (actual time=0.096..0.342 rows=100 loops=1)
Output: uid, pos, crt_date, crt_time, ((pos <-> '(0,0)'::point))
Buffers: shared hit=126
-> Index Scan using idx_cainiao_trace_realtime_pos on public.cainiao_trace_realtime (cost=0.41..1167.86 rows=1042 width=40) (actual time=0.094..0.330 rows=100 loops=1)
Output: uid, pos, crt_date, crt_time, (pos <-> '(0,0)'::point)
Index Cond: ('<(0,0),20000>'::circle @> cainiao_trace_realtime.pos)
Order By: (cainiao_trace_realtime.pos <-> '(0,0)'::point)
Buffers: shared hit=126
Planning time: 0.098 ms
Execution time: 0.377 ms
(10 rows)
pgbench的测试脚本如下
vi test1.sql
\set x random(-500000,500000)
\set y random(-500000,500000)
select * from cainiao_trace_realtime where circle(point(:x,:y),20000) @> pos order by pos <-> point(:x,:y) limit 100;
开始测试，持续300秒。
numactl --physcpubind=32-63 pgbench -M prepared -n -r -P 1 -f ./test1.sql -c 32 -j 32 -T 300
测试结果
每秒处理召回请求 6万，单次请求延迟0.53毫秒。
transaction type: ./test1.sql
scaling factor: 1
query mode: prepared
number of clients: 32
number of threads: 32
duration: 300 s
number of transactions actually processed: 18087765
latency average = 0.531 ms
latency stddev = 0.103 ms
tps = 60292.169523 (including connections establishing)
tps = 60292.786291 (excluding connections establishing)
script statistics:
- statement latencies in milliseconds:
0.001 \set x random(-500000,500000)
0.000 \set y random(-500000,500000)
0.529 select * from cainiao_trace_realtime where circle(point(:x,:y),20000) @> pos order by pos <-> point(:x,:y) limit 100;
备注，如果只召回一名快递员，可以达到28万 tps.
transaction type: ./test1.sql
scaling factor: 1
query mode: prepared
number of clients: 32
number of threads: 32
duration: 300 s
number of transactions actually processed: 84257925
latency average = 0.114 ms
latency stddev = 0.033 ms
tps = 280858.872643 (including connections establishing)
tps = 280862.101773 (excluding connections establishing)
script statistics:
- statement latencies in milliseconds:
0.001 \set x random(-500000,500000)
0.000 \set y random(-500000,500000)
0.112 select * from cainiao_trace_realtime where circle(point(:x,:y),20000) @> pos order by pos <-> point(:x,:y) limit 1;
3 混合测试
同时压测快递员轨迹插入、随机召回快递员。
压测结果
插入TPS: 12.5万，响应时间0.25毫秒
查询TPS: 2.17万，响应时间1.47毫秒
transaction type: ./test.sql
scaling factor: 1
query mode: prepared
number of clients: 32
number of threads: 32
duration: 100 s
number of transactions actually processed: 12508112
latency average = 0.256 ms
latency stddev = 1.266 ms
tps = 125072.868788 (including connections establishing)
tps = 125080.518685 (excluding connections establishing)
script statistics:
- statement latencies in milliseconds:
0.002 \set uid random(1,1000000)
0.001 \set x random(-500000,500000)
0.000 \set y random(-500000,500000)
0.253 insert into cainiao_16 values (:uid, point(:x,:y), now()::date, now()::time);
transaction type: ./test1.sql
scaling factor: 1
query mode: prepared
number of clients: 32
number of threads: 32
duration: 100 s
number of transactions actually processed: 2174422
latency average = 1.472 ms
latency stddev = 0.455 ms
tps = 21743.641754 (including connections establishing)
tps = 21744.366018 (excluding connections establishing)
script statistics:
- statement latencies in milliseconds:
0.002 \set x random(-500000,500000)
0.000 \set y random(-500000,500000)
1.469 select * from cainiao_trace_realtime where circle(point(:x,:y),20000) @> pos order by pos <-> point(:x,:y) limit 100;
快递员实时位置表剥离
如果要尽量的降低RT，快递员实时位置表可以与轨迹明细表剥离，由应用程序来更新快递员的实时位置。
至于这个实时位置表，你要把它放在明细表的数据库，还是另外一个数据库？
我的建议是放在另外一个数据库，因为这种表的应用非常的独立（更新，查询），都是小事务。
而明细轨迹，可能涉及到比较大的查询，以插入，范围分析，数据合并，日轨迹查询为主。
将明细和实时轨迹独立开来，也是有原因的。
剥离后，明细位置你可以继续使用UNLOGGED TABLE，也可以使用普通表。
下面测试一下剥离后的性能。
pgbench脚本，更新快递员位置，查询某个随机点的最近100个快递员。
postgres=# \d cainiao_trace_realtime
Table "public.cainiao_trace_realtime"
Column | Type | Collation | Nullable | Default
----------+---------------------------+-----------+----------+---------
uid | integer | | not null |
pos | point | | |
crt_date | date | | |
crt_time | time(0) without time zone | | |
Indexes:
"cainiao_trace_realtime_pkey" PRIMARY KEY, btree (uid)
"idx_cainiao_trace_realtime_pos" gist (pos)
postgres=# select count(*),min(uid),max(uid) from cainiao_trace_realtime ;
count | min | max
---------+-----+---------
1000000 | 1 | 1000000
(1 row)
vi test1.sql
\set uid 1 1000000
\set x random(-500000,500000)
\set y random(-500000,500000)
insert into cainiao_trace_realtime (uid,pos) values (:uid, point(:x,:y)) on conflict (uid) do update set pos=excluded.pos;
vi test2.sql
\set x random(-500000,500000)
\set y random(-500000,500000)
select * from cainiao_trace_realtime where circle(point(:x,:y),20000) @> pos order by pos <-> point(:x,:y) limit 100;
压测结果1(更新 18万/s, 响应时间0.17毫秒)
numactl --physcpubind=0-31 pgbench -M prepared -n -r -P 1 -f ./test1.sql -c 32 -j 32 -T 300
transaction type: ./test1.sql
scaling factor: 1
query mode: prepared
number of clients: 32
number of threads: 32
duration: 300 s
number of transactions actually processed: 54283976
latency average = 0.177 ms
latency stddev = 2.241 ms
tps = 180943.029385 (including connections establishing)
tps = 180945.072949 (excluding connections establishing)
script statistics:
- statement latencies in milliseconds:
0.001 \set uid random(1,1000000)
0.001 \set x random(-500000,500000)
0.000 \set y random(-500000,500000)
0.175 insert into cainiao_trace_realtime (uid,pos) values (:uid, point(:x,:y)) on conflict (uid) do update set pos=excluded.pos;
压测结果2(查询 5.2万/s, 响应时间0.61毫秒)
numactl --physcpubind=0-31 pgbench -M prepared -n -r -P 1 -f ./test2.sql -c 32 -j 32 -T 100
transaction type: ./test2.sql
scaling factor: 1
query mode: prepared
number of clients: 32
number of threads: 32
duration: 100 s
number of transactions actually processed: 5195710
latency average = 0.616 ms
latency stddev = 0.154 ms
tps = 51956.132043 (including connections establishing)
tps = 51957.754525 (excluding connections establishing)
script statistics:
- statement latencies in milliseconds:
0.001 \set x random(-500000,500000)
0.000 \set y random(-500000,500000)
0.614 select * from cainiao_trace_realtime where circle(point(:x,:y),20000) @> pos order by pos <-> point(:x,:y) limit 100;
压测结果3(更新13万/s, 响应时间0.24毫秒 + 查询1.8万/s, 响应时间1.78毫秒)
numactl --physcpubind=0-31 pgbench -M prepared -n -r -P 1 -f ./test1.sql -c 32 -j 32 -T 100
transaction type: ./test1.sql
scaling factor: 1
query mode: prepared
number of clients: 32
number of threads: 32
duration: 100 s
number of transactions actually processed: 13057629
latency average = 0.245 ms
latency stddev = 0.805 ms
tps = 130575.023251 (including connections establishing)
tps = 130582.436319 (excluding connections establishing)
script statistics:
- statement latencies in milliseconds:
0.002 \set uid random(1,1000000)
0.001 \set x random(-500000,500000)
0.000 \set y random(-500000,500000)
0.242 insert into cainiao_trace_realtime (uid,pos) values (:uid, point(:x,:y)) on conflict (uid) do update set pos=excluded.pos;
numactl --physcpubind=32-63 pgbench -M prepared -n -r -P 1 -f ./test2.sql -c 32 -j 32 -T 100
transaction type: ./test2.sql
scaling factor: 1
query mode: prepared
number of clients: 32
number of threads: 32
duration: 100 s
number of transactions actually processed: 1791580
latency average = 1.786 ms
latency stddev = 2.128 ms
tps = 17915.362549 (including connections establishing)
tps = 17916.037454 (excluding connections establishing)
script statistics:
- statement latencies in milliseconds:
0.002 \set x random(-500000,500000)
0.000 \set y random(-500000,500000)
1.784 select * from cainiao_trace_realtime where circle(point(:x,:y),20000) @> pos order by pos <-> point(:x,:y) limit 100;
历史轨迹进入OSS对象存储
前面对实时轨迹数据使用一周的分表，目的就是有时间可以将其写入到OSS，方便维护。
每天可以将6天前的数据，写入OSS对象存储。
OSS对象存储。
阿里云PostgreSQL有oss_ext插件，可以将数据写入oss对象存储。同时也支持从oss对象存储读取数据（外部表的方式），对用户透明。
详见
https://help.aliyun.com/document_detail/44461.html
其他需求，轨迹合并设计
单个快递员，一天产生的轨迹是8640条。
PostgreSQL支持JSON、HSTORE(kv)、数组、复合数组 类型。每天将单个快递员的轨迹聚合为一条记录，可以大幅度提升按快递员查询轨迹的速度。
同样的场景可以参考：
《performance tuning about multi-rows query aggregated to single-row query》
聚合例子
create type trace as (pos point, crt_time time);
create table cainiao_trace_agg (crt_date date, uid int, trace_arr trace[], primary key(crt_date,uid));
insert into cainiao_trace_agg (crt_date , uid , trace_arr )
select crt_date, uid, array_agg( (pos,crt_time)::trace ) from cainiao_0_2 group by crt_date, uid;
查询某个快递员1天的轨迹，性能提升对比
聚合前(b-tree索引)，耗时8毫秒
postgres=# explain (analyze,verbose,timing,costs,buffers) select * from cainiao_0_2 where uid=340054;
QUERY PLAN
--------------------------------------------------------------------------------------------------------------------------------
Index Scan using idx on public.cainiao_0_2 (cost=0.57..193.61 rows=194 width=32) (actual time=0.033..7.711 rows=7904 loops=1)
Output: uid, pos, crt_date, crt_time
Index Cond: (cainiao_0_2.uid = 340054)
Buffers: shared hit=7720
Planning time: 0.090 ms
Execution time: 8.198 ms
(6 rows)
聚合后，耗时0.033毫秒
postgres=# explain (analyze,verbose,timing,costs,buffers) select * from cainiao_trace_agg where crt_date='2017-04-18' and uid=340054;
QUERY PLAN
---------------------------------------------------------------------------------------------------------------------------------------------------
Index Scan using cainiao_trace_agg_pkey on public.cainiao_trace_agg (cost=0.42..2.44 rows=1 width=978) (actual time=0.016..0.017 rows=1 loops=1)
Output: crt_date, uid, trace_arr
Index Cond: ((cainiao_trace_agg.crt_date = '2017-04-18'::date) AND (cainiao_trace_agg.uid = 340054))
Buffers: shared hit=4
Planning time: 0.098 ms
Execution time: 0.033 ms
(6 rows)
小结
1. 本文以物流轨迹系统为背景，对两个常见需求进行数据库的设计以及模型的压测：实时跟踪快递员轨迹，实时召回附近的快递员。
2. PostgreSQL 结合 OSS，实现了数据的冷热分离，历史轨迹写入OSS保存，再通过OSS可以共享给HybridDB for PostgreSQL，进行实时的数据挖掘分析。
3. 单机满足了每秒18万的轨迹写入，按最近距离召回快递员(100名快递员)可以达到6万/s的速度，按最近距离召回快递员(1名快递员)可以达到28万/s的速度。
4. 使用PostgreSQL的继承特性，可以更好的管理分区表，例如要查询礼拜一的所有分区，查询这些分区的主表。 如果要查某个模值的所有时间段数据，查询对应的主表即可。
参考
《PostGIS long lat geometry distance search tuning using gist knn function》
《ApsaraDB的左右互搏(PgSQL+HybridDB+OSS) - 解决OLTP+OLAP混合需求》
