PostgreSQL 重复 数据清洗 优化

背景

重复数据清洗是一个比较常见的业务需求，比如有些数据库不支持唯一约束，或者程序设计之初可能没有考虑到需要在某些列上面加唯一约束，导致应用在上线一段时间后，产生了一些重复的数据。

那么重复数据的清洗需求就来了。

有哪些清洗手段，如何做到高效的清洗呢？

一个小小的应用场景，带出了10项数据库技术点，听我道来。

重复数据清洗手段

比如一个表，有几个字段本来应该是唯一的，产生了重复值，现在给你一个规则，保留重复值中的一条，其他删掉。

例子

postgres=# create table tbl_dup(   

  id serial8,   

  sid int,   

  crt_time timestamp,   

  mdf_time timestamp,   

  c1 text default md5(random()::text),   

  c2 text default md5(random()::text),   

  c3 text default md5(random()::text),   

  c4 text default md5(random()::text),   

  c5 text default md5(random()::text),   

  c6 text default md5(random()::text),   

  c7 text default md5(random()::text),   

  c8 text default md5(random()::text)   

);   

删除重复的 (sid + crt_time) 组合，并保留重复值中，mdf_time最大的一条。

生成测试数据100万条，1/10 的重复概率，同时为了避免重复数据在一个数据块中，每跳跃500条生成一条重复值。

就生成测试数据 ，是不是觉得已经很炫酷了呢？一条SQL就造了一批这样的数据。

insert into tbl_dup (sid, crt_time, mdf_time)   

select   

  case when mod(id,11)=0 then id+500 else id end,   

  case when mod(id,11)=0 then now()+(''||id+500||' s')::interval else now()+(''||id||' s')::interval end,   

  clock_timestamp()   

from generate_series(1,1000000) t(id);  

验证, 重复记录的ctid不在同一个数据块中。

验证方法是不是很酷呢？用了窗口查询。

postgres=# select * from (select ctid,sid,crt_time,mdf_time, count(*) over(partition by sid,crt_time) as cnt from tbl_dup) t where t.cnt>=2;  

    ctid    |  sid   |          crt_time          |          mdf_time          | cnt   

------------+--------+----------------------------+----------------------------+-----  

 (0,11)     |    511 | 2016-12-29 17:42:13.935348 | 2016-12-29 17:33:43.092625 |   2  

 (20,11)    |    511 | 2016-12-29 17:42:13.935348 | 2016-12-29 17:33:43.102726 |   2  

 (20,22)    |    522 | 2016-12-29 17:42:24.935348 | 2016-12-29 17:33:43.102927 |   2  

 (0,22)     |    522 | 2016-12-29 17:42:24.935348 | 2016-12-29 17:33:43.09283  |   2  

 (21,8)     |    533 | 2016-12-29 17:42:35.935348 | 2016-12-29 17:33:43.103155 |   2  

 (1,8)      |    533 | 2016-12-29 17:42:35.935348 | 2016-12-29 17:33:43.093191 |   2  

 (21,19)    |    544 | 2016-12-29 17:42:46.935348 | 2016-12-29 17:33:43.103375 |   2  

 (1,19)     |    544 | 2016-12-29 17:42:46.935348 | 2016-12-29 17:33:43.093413 |   2  

....  

包含重复的值大概这么多

postgres=# select count(*) from (select * from (select ctid,sid,crt_time,mdf_time, count(*) over(partition by sid,crt_time) as cnt from tbl_dup) t where t.cnt=2) t;  

 count    

--------  

 181726  

(1 row)  

Time: 1690.709 ms  

你如果觉得这个还挺快的，偷偷告诉你测试环境CPU型号。

Intel(R) Xeon(R) CPU E5-2630 0 @ 2.30GHz

接下来开始去重了

方法1, 插入法

将去重后的结果插入一张新的表中，耗时5.8秒

create table tbl_uniq(like tbl_dup including all);  

insert into tbl_uniq (id,sid,crt_time,mdf_time,c1,c2,c3,c4,c5,c6,c7,c8)  

select id,sid,crt_time,mdf_time,c1,c2,c3,c4,c5,c6,c7,c8 from   

(select row_number() over(partition by sid,crt_time order by mdf_time desc) as rn, * from tbl_dup) t  

where t.rn=1;  

INSERT 0 909137  

Time: 5854.349 ms  

分析优化空间，显示排序可以优化

postgres=# explain (analyze,verbose,timing,costs,buffers)  insert into tbl_uniq (id,sid,crt_time,mdf_time,c1,c2,c3,c4,c5,c6,c7,c8)  

select id,sid,crt_time,mdf_time,c1,c2,c3,c4,c5,c6,c7,c8 from   

(select row_number() over(partition by sid,crt_time order by mdf_time desc) as rn, * from tbl_dup) t  

where t.rn=1;  

                                                                                                QUERY PLAN                                                                                                  

----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------  

 Insert on public.tbl_uniq  (cost=423098.84..458098.84 rows=5000 width=292) (actual time=5994.723..5994.723 rows=0 loops=1)  

   Buffers: shared hit=1021856 read=36376 dirtied=36375, temp read=37391 written=37391  

   ->  Subquery Scan on t  (cost=423098.84..458098.84 rows=5000 width=292) (actual time=1715.278..3620.269 rows=909137 loops=1)  

         Output: t.id, t.sid, t.crt_time, t.mdf_time, t.c1, t.c2, t.c3, t.c4, t.c5, t.c6, t.c7, t.c8  

         Filter: (t.rn = 1)  

         Rows Removed by Filter: 90863  

         Buffers: shared hit=40000, temp read=37391 written=37391  

         ->  WindowAgg  (cost=423098.84..445598.84 rows=1000000 width=300) (actual time=1715.276..3345.392 rows=1000000 loops=1)  

               Output: row_number() OVER (?), tbl_dup.id, tbl_dup.sid, tbl_dup.crt_time, tbl_dup.mdf_time, tbl_dup.c1, tbl_dup.c2, tbl_dup.c3, tbl_dup.c4, tbl_dup.c5, tbl_dup.c6, tbl_dup.c7, tbl_dup.c8  

               Buffers: shared hit=40000, temp read=37391 written=37391  

               ->  Sort  (cost=423098.84..425598.84 rows=1000000 width=292) (actual time=1715.263..2174.426 rows=1000000 loops=1)  

                     Output: tbl_dup.sid, tbl_dup.crt_time, tbl_dup.mdf_time, tbl_dup.id, tbl_dup.c1, tbl_dup.c2, tbl_dup.c3, tbl_dup.c4, tbl_dup.c5, tbl_dup.c6, tbl_dup.c7, tbl_dup.c8  

                     Sort Key: tbl_dup.sid, tbl_dup.crt_time, tbl_dup.mdf_time DESC  

                     Sort Method: external sort  Disk: 299128kB  

                     Buffers: shared hit=40000, temp read=37391 written=37391  

                     ->  Seq Scan on public.tbl_dup  (cost=0.00..50000.00 rows=1000000 width=292) (actual time=0.012..398.007 rows=1000000 loops=1)  

                           Output: tbl_dup.sid, tbl_dup.crt_time, tbl_dup.mdf_time, tbl_dup.id, tbl_dup.c1, tbl_dup.c2, tbl_dup.c3, tbl_dup.c4, tbl_dup.c5, tbl_dup.c6, tbl_dup.c7, tbl_dup.c8  

                           Buffers: shared hit=40000  

 Planning time: 0.174 ms  

 Execution time: 6120.921 ms  

(20 rows)  

优化1

索引，消除排序，优化后只需要3.9秒

对于在线业务，PostgreSQL可以使用并行CONCURRENTLY创建索引，不会堵塞DML。

postgres=# create index CONCURRENTLY idx_tbl_dup on tbl_dup(sid,crt_time,mdf_time desc);  

CREATE INDEX  

Time: 765.426 ms  

postgres=# truncate tbl_uniq;  

TRUNCATE TABLE  

Time: 208.808 ms  

postgres=# insert into tbl_uniq (id,sid,crt_time,mdf_time,c1,c2,c3,c4,c5,c6,c7,c8)                                                  

select id,sid,crt_time,mdf_time,c1,c2,c3,c4,c5,c6,c7,c8 from   

(select row_number() over(partition by sid,crt_time order by mdf_time desc) as rn, * from tbl_dup) t  

where t.rn=1;  

INSERT 0 909137  

Time: 3978.425 ms  

postgres=# explain (analyze,verbose,timing,costs,buffers)  insert into tbl_uniq (id,sid,crt_time,mdf_time,c1,c2,c3,c4,c5,c6,c7,c8)  

select id,sid,crt_time,mdf_time,c1,c2,c3,c4,c5,c6,c7,c8 from   

(select row_number() over(partition by sid,crt_time order by mdf_time desc) as rn, * from tbl_dup) t  

where t.rn=1;  

                                                                                                QUERY PLAN                                                                                                  

----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------  

 Insert on public.tbl_uniq  (cost=0.42..159846.13 rows=5000 width=292) (actual time=4791.360..4791.360 rows=0 loops=1)  

   Buffers: shared hit=1199971 read=41303 dirtied=36374  

   ->  Subquery Scan on t  (cost=0.42..159846.13 rows=5000 width=292) (actual time=0.061..2177.768 rows=909137 loops=1)  

         Output: t.id, t.sid, t.crt_time, t.mdf_time, t.c1, t.c2, t.c3, t.c4, t.c5, t.c6, t.c7, t.c8  

         Filter: (t.rn = 1)  

         Rows Removed by Filter: 90863  

         Buffers: shared hit=218112 read=4929  

         ->  WindowAgg  (cost=0.42..147346.13 rows=1000000 width=300) (actual time=0.060..1901.174 rows=1000000 loops=1)  

               Output: row_number() OVER (?), tbl_dup.id, tbl_dup.sid, tbl_dup.crt_time, tbl_dup.mdf_time, tbl_dup.c1, tbl_dup.c2, tbl_dup.c3, tbl_dup.c4, tbl_dup.c5, tbl_dup.c6, tbl_dup.c7, tbl_dup.c8  

               Buffers: shared hit=218112 read=4929  

               ->  Index Scan using idx_tbl_dup on public.tbl_dup  (cost=0.42..127346.13 rows=1000000 width=292) (actual time=0.051..601.249 rows=1000000 loops=1)  

                     Output: tbl_dup.id, tbl_dup.sid, tbl_dup.crt_time, tbl_dup.mdf_time, tbl_dup.c1, tbl_dup.c2, tbl_dup.c3, tbl_dup.c4, tbl_dup.c5, tbl_dup.c6, tbl_dup.c7, tbl_dup.c8  

                     Buffers: shared hit=218112 read=4929  

 Planning time: 0.304 ms  

 Execution time: 4834.392 ms  

(15 rows)  

Time: 4835.484 ms