PostgreSQL 最佳实践 - 逻辑增量复制(MySQL <-> PgSQL <-> PgSQL)

背景

alidecode是RDS PG提供的一个逻辑复制插件，使用它，可以将RDS PG通过逻辑复制的方式，把数据同步到线下的PostgreSQL。

还可以将数据从PG同步到其他的数据处理平台，例如Kafka。

同时还支持将MySQL的数据同步到PostgreSQL。(使用SQL管理函数创建逻辑复制的SLOT，调用alidecode 函数，函数会从WAL日志解析出REDO，从而实现实时的数据变更拽取。)

你还可用dbsync来实现平滑升级PostgreSQL的大版本，是不是很酷呢。

mysql2pgsql

https://help.aliyun.com/document_detail/35458.html

pgsql2pgsql

https://help.aliyun.com/document_detail/35459.html

源码

https://github.com/aliyun/rds_dbsync

下面是简单的使用方法。

准备工作

提交工单，开放用户的replication角色。(目前RDS用户根账号已经默认带了replication权限，这个步骤可以忽略)

postgres=# alter role digoal replication;  
ALTER ROLE  

阿里云的RDS PG需要将主备的pg_hba.conf进行修正，增加replication的条目。

例子

$ vi $PGDATA/pg_hba.conf  
host replication digoal 0.0.0.0/0 md5  

阿里云的RDS PG需要调整主备的postgresql.conf，将wal_level改成logical。

并重启主备数据库，所以用户要开通此功能，需要重启实例哦

wal_level = logical  

用户需要在RDS管控平台，配置白名单，允许alidecode客户端所在的主机连接RDS数据库。

下载alidecode客户端。

安装postgresql, mysql。（需要用到头文件）

如果你不需要将mysql的数据同步到PG，则不需要编译mysql的部分。

在Makefile和dbsync.h中注释掉mysql的部分即可。

编译前，你需要配置一下pgsync.cpp，这里需要配置三个连接串。

src对应RDS的连接串。

local对应的是一个中间库，它用来记录任务信息，记录全量同步时的增量数据(全量同步数据时，并行的接收xlog，接收的XLOG转义成SQL存在中间库)。

desc对应目标库，即数据要同步到这个库。

$ vi dbsync.cpp   
        src =   (char *)"host=digoal_111.pg.rds.aliyuncs.com port=3433 dbname=db1 user=digoal password=digoal";  
        local = (char *)"host=127.0.0.1 port=1925 dbname=db2 user=postgres password=postgres";  
        desc = (char *)"host=127.0.0.1 port=1925 dbname=db1 user=postgres password=postgres";  

编译

$ make  

alidecode不负责DDL的同步，所以DDL需要用户自己操作

例子

/home/dege.zzz/pgsql9.5/bin/pg_dump -F p -s --no-privileges --no-tablespaces --no-owner -h digoal_111.pg.rds.aliyuncs.com -p 3433 -U digoal db1 | psql db1 -f -  

同步数据，执行dbsync就可以了

./dbsync   
full sync start 2016-05-26 15:35:42.336903, end 2016-05-26 15:35:42.699032 restart decoder sync  
decoder sync start 2016-05-26 15:35:42.337482  
decoder slot rds_logical_sync_slot exist  
starting logical decoding sync thread  
starting decoder apply thread  
pg_recvlogical: starting log streaming at 0/0 (slot rds_logical_sync_slot)  
pg_recvlogical: confirming recv up to 0/0, flush to 0/0 (slot rds_logical_sync_slot)  

元数据记录在db2，如果失败要重新来过的话，建议清除它，同时清除目标库的已同步数据，然后重新调用dbsync。

db2=# \dt  
             List of relations  
 Schema |      Name      | Type  |  Owner     
--------+----------------+-------+----------  
 public | db_sync_status | table | postgres  
 public | sync_sqls      | table | postgres  
(2 rows)  
  
db2=# drop table db_sync_status ;  
DROP TABLE  
db2=# drop table sync_sqls ;  
DROP TABLE  

压测rds

pgbench -M prepared -n -r -P 1 -c 80 -j 80 -T 100 -h digoal_111o.pg.rds.aliyuncs.com -p 3433 -U digoal db1  

可以看到同步的过程

pg_recvlogical: confirming recv up to 1/4EE3F08, flush to 1/4EE3F08 (slot rds_logical_sync_slot)  
pg_recvlogical: confirming recv up to 1/4F8BA20, flush to 1/4F8BA20 (slot rds_logical_sync_slot)  
pg_recvlogical: confirming recv up to 1/5025228, flush to 1/5025228 (slot rds_logical_sync_slot)  
pg_recvlogical: confirming recv up to 1/50C6E68, flush to 1/50C6E68 (slot rds_logical_sync_slot)  
pg_recvlogical: confirming recv up to 1/51578A0, flush to 1/51578A0 (slot rds_logical_sync_slot)  
pg_recvlogical: confirming recv up to 1/51E7CF8, flush to 1/51E7CF8 (slot rds_logical_sync_slot)  

压测完后，查看数据是否一致

psql -h 127.0.0.1 db1  
db1=# select sum(hashtext(t.*::text)) from pgbench_accounts t;  
      sum        
---------------  
 -582104340143  
(1 row)  
  
psql -h digoal_111o.pg.rds.aliyuncs.com -p 3433 -U digoal db1  
psql (9.6beta1, server 9.4.1)  
Type "help" for help.  
  
db1=> select sum(hashtext(t.*::text)) from pgbench_accounts t;  
      sum        
---------------  
 -582104340143  
(1 row)  

db_sync工具下载地址

https://github.com/aliyun/rds_dbsync

参考

https://github.com/aliyun/rds_dbsync

digoal’s 大量PostgreSQL文章入口