MySQL,Oracle,SQL Server等准实时同步到PostgreSQL的方案之一 - FDW外部访问接口

本文涉及的产品
云数据库 RDS SQL Server,基础系列 2核4GB
云原生数据库 PolarDB 分布式版,标准版 2核8GB
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
简介:

标签

PostgreSQL , MySQL , Oracle , SQL Server , DRDS , 数据同步 , FDW , 调度 , 逻辑删除 , 空洞 , 事务时间


背景

PostgreSQL历经几十年的发展,厚积薄发,每年一个大版本,每年都有惊喜,2017年10月推出的10版本,包含重大重磅特性。PostgreSQL在逐渐成为企业的核心库。

pic

企业数据源可能众多,为了实现数据的同步,方法有很多,较为常见的是这样的方法,将数据变更汇总到消息队列,通过消息订阅的方式,传播数据。消息队列可以承担持久化存储、幂等消费、数据清洗、流式计算等工作。当然,为了提高消息队列的吞吐率,越简单越好。就像路由器和交换机一样,各司其职。

《debezium - 数据实时捕获和传输管道(CDC)》

pic

在阿里云的PostgreSQL生态中,还可以有这样的方法(数据实时写入PG,经过PG的清洗,通过OSS流入HDB PG。实现OLTP和OLAP的数据流动。):

pic

除了以上方法,实际上PostgreSQL还有一个杀手级功能,FDW,通过FDW,可以连接各种数据源,目前已有的接口如下(看样子包含了地球上所有的数据源):

https://wiki.postgresql.org/wiki/Fdw

下面我们来看一下,如何基于FDW构建一个简单的准实时同步方案。

FDW是什么

pic

FDW是PostgreSQL的一个外部数据访问接口规范,用户可以自定义任何数据源的FDW handler,这样就能访问你要访问的数据接口了。

https://www.postgresql.org/docs/10/static/fdwhandler.html

上游数据源设计规范

为了适应使用FDW来实现准实时同步的需求,上游数据源必须要符合一定的规约。

1、被同步的表必须有主键。

2、被同步的表必须有修改时间(非空),修改时间字段的初始值为记录创建时间,每次更新记录,更新为当前时间。

3、被同步的表必须有逻辑删除标记(不要使用delete直接删除数据)

或者DELETE的操作与DDL操作一样,采用调度(业务系统与PG系统同时操作的模式)。

4、被同步的表,修改时间字段,必须有索引。

逻辑删除

简单介绍一下逻辑删除。

create table tbl(  
  id int primary key,            -- 主键  
  info text,   
  state char(1),                 -- 逻辑删除标记  
  crt_time timestamp not null,   -- 写入时间  
  mod_time timestamp not null    -- 记录被修改的时间,记录创建时,设置mod_time为创建时间。  
);  

逻辑删除时,并非调用delete,而是更新state这个标记。

update tbl set state='d' , mod_time=now() where id=?;  

PostgreSQL支持使用RULE,将DELETE操作转换为UPDATE操作,例子

postgres=# create table log(id serial8 primary key, info text, state char(1), crt_time timestamp, mod_time timestamp);
CREATE TABLE

postgres=# create rule r1 as on delete to log do instead update log set state='d',mod_time=now() where id=OLD.id;
CREATE RULE
postgres=# insert into log (info,crt_time,mod_time) values ('abc',now(),now());
INSERT 0 1
postgres=# select * from log;
 id | info | state |          crt_time          |          mod_time          
----+------+-------+----------------------------+----------------------------
  1 | abc  |       | 2017-10-27 15:15:44.577036 | 2017-10-27 15:15:44.577036
(1 row)

postgres=# delete from log where id=1;
DELETE 0
postgres=# select * from log;
 id | info | state |          crt_time          |          mod_time          
----+------+-------+----------------------------+----------------------------
  1 | abc  | d     | 2017-10-27 15:15:44.577036 | 2017-10-27 15:15:54.367215
(1 row)

postgres=# update log set info='test';
UPDATE 1
postgres=# select * from log;
 id | info | state |          crt_time          |          mod_time          
----+------+-------+----------------------------+----------------------------
  1 | test | d     | 2017-10-27 15:15:44.577036 | 2017-10-27 15:15:54.367215
(1 row)

下游同步时,就可以同步到这样的操作。

如果业务上一定要DELETE,那么不适合本方案。请使用binlog的同步方式。例如rds_dbsync这个工具。

https://github.com/aliyun/rds_dbsync/

《MySQL准实时同步到PostgreSQL, Greenplum的方案之一 - rds_dbsync》

调度设计与例子

设计

1、创建外部表

2、创建本地表,结构与外部表一致

3、创建同步函数

4、同步函数逻辑

设置栅栏,对于一张外部表,同一时刻只允许开启一个任务,防止重复拖数据(虽然insert on conflict会幂等处理,但是也浪费资源)。

设置上一次同步的截止时间戳

从上一次截止的时间戳开始同步,直到N分钟前的数据(可调),目的是防止空洞(比如一些老事务未提交,时间戳还是老的,会成为空洞)。

5、使用linux crontab,创建定时任务

例子

环境部署,MySQL fdw的使用,详见:

《PostgreSQL 10 + PostGIS + Sharding(pg_pathman) + MySQL(fdw外部表) on ECS 部署指南(适合新用户)》

假设mysql_fdw foreign data wrapper, foreign server, user mapping, 权限 都已经配置好了。

1、外部表

CREATE FOREIGN TABLE remote_table1(  
     id int,  
     info text,  
     state char(1),  
     crt_time timestamp,  
     mod_time timestamp  
     )  
SERVER mysql_server1  
     OPTIONS (dbname 'db1', table_name 'remote_table1');  

2、本地表(与远程表定义一致)

create table local_table1(  
  id int primary key,            -- 主键  
  info text,   
  state char(1),                 -- 逻辑删除标记  
  crt_time timestamp not null,   -- 写入时间  
  mod_time timestamp not null    -- 记录被修改的时间,记录创建时,设置mod_time为创建时间。  
);  

3、同步函数1

create or replace function sync_tbl1(rmt_table name) returns void as $$   -- 每个表配置一个独立的advisory ID,防止单个表有多个任务同时跑。  
declare  
  v_mod_time timestamp;        -- 起始时间  
  v_upper_mod_time timestamp;  -- 截止时间  
begin  
  if not pg_try_advisory_lock(hashtext(rmt_table)) then return; end if;   -- 保护,确保同一个表,只有单个任务在同步。不同的表不会干扰。  
  
  select max(mod_time) into v_mod_time from local_table1 ;  
  if v_mod_time is null then   
    v_mod_time='0001-01-01'::timestamp;   
  end if;  
  
  v_upper_mod_time := now()-interval '5 min';     -- 最多同步到5分钟前,防止空洞(比如一些老事务未提交,时间戳还是老的,会成为空洞)。  
  
  insert into local_table1    
  (  
    select * from remote_table1 where mod_time > v_mod_time and mod_time <= v_upper_mod_time;  
  )   
  on conflict (id) do update set info=excluded.info,state=excluded.state,crt_time=excluded.crt_time,mod_time=excluded.mod_time ;   -- 每个表的主键,以及字段可能不一样,需要修改一下。  
    
  return;  
end;  
$$ language plpgsql strict;  

4、调度

假设每分钟调度一次。

crontab -e  
* * * * * psql -c "select sync_tbl1('remote_table1');"  

对于很大数据量的数据,PostgreSQL可以使用分区表。

《PostgreSQL 10.0 preview 功能增强 - 内置分区表》

注意

空洞问题。

最多同步到5分钟前,防止空洞(比如一些老事务未提交,时间戳还是老的,会成为空洞)。

参考

《debezium - 数据实时捕获和传输管道(CDC)》

《PostgreSQL 10 + PostGIS + Sharding(pg_pathman) + MySQL(fdw外部表) on ECS 部署指南(适合新用户)》

《PostgreSQL 10.0 preview 功能增强 - 内置分区表》

相关实践学习
如何在云端创建MySQL数据库
开始实验后,系统会自动创建一台自建MySQL的 源数据库 ECS 实例和一台 目标数据库 RDS。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助 &nbsp; &nbsp; 相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
目录
相关文章
|
4天前
|
Oracle NoSQL 关系型数据库
主流数据库对比:MySQL、PostgreSQL、Oracle和Redis的优缺点分析
主流数据库对比:MySQL、PostgreSQL、Oracle和Redis的优缺点分析
14 2
|
7天前
|
关系型数据库 MySQL 网络安全
5-10Can't connect to MySQL server on 'sh-cynosl-grp-fcs50xoa.sql.tencentcdb.com' (110)")
5-10Can't connect to MySQL server on 'sh-cynosl-grp-fcs50xoa.sql.tencentcdb.com' (110)")
|
8天前
|
关系型数据库 MySQL 数据库
docker启动mysql多实例连接报错Can’t connect to local MySQL server through socket ‘/var/run/mysqld/mysqld.sock’
docker启动mysql多实例连接报错Can’t connect to local MySQL server through socket ‘/var/run/mysqld/mysqld.sock’
30 0
|
2月前
|
数据采集 关系型数据库 MySQL
大数据-业务数据采集-FlinkCDC The MySQL server is not configured to use a ROW binlog_format
大数据-业务数据采集-FlinkCDC The MySQL server is not configured to use a ROW binlog_format
34 1
|
2月前
|
网络协议 NoSQL 网络安全
【Azure 应用服务】由Web App“无法连接数据库”而逐步分析到解析内网地址的办法(SQL和Redis开启private endpoint,只能通过内网访问,无法从公网访问的情况下)
【Azure 应用服务】由Web App“无法连接数据库”而逐步分析到解析内网地址的办法(SQL和Redis开启private endpoint,只能通过内网访问,无法从公网访问的情况下)
|
2月前
|
SQL 机器学习/深度学习 开发工具
【机器学习 Azure Machine Learning】Azure Machine Learning 访问SQL Server 无法写入问题 (使用微软Python AML Core SDK)
【机器学习 Azure Machine Learning】Azure Machine Learning 访问SQL Server 无法写入问题 (使用微软Python AML Core SDK)
|
2月前
|
SQL 数据库 Windows
【应用服务 App Service】当使用EntityFrameWorkCore访问Sql Server数据库时,在Azure App Service会出现Cannot create a DbSet for ** because this type is not included in the model for the context的错误
【应用服务 App Service】当使用EntityFrameWorkCore访问Sql Server数据库时,在Azure App Service会出现Cannot create a DbSet for ** because this type is not included in the model for the context的错误
|
3月前
|
SQL Oracle 关系型数据库
|
3月前
|
SQL Oracle 关系型数据库
MySQL、SQL Server和Oracle数据库安装部署教程
数据库的安装部署教程因不同的数据库管理系统(DBMS)而异,以下将以MySQL、SQL Server和Oracle为例,分别概述其安装部署的基本步骤。请注意,由于软件版本和操作系统的不同,具体步骤可能会有所变化。
165 3
|
3月前
|
安全 关系型数据库 MySQL
【Python】已解决:pymysql.err.OperationalError:(2003 “Can’t connect to MySQL server on ‘localhost’ ([WinEr
【Python】已解决:pymysql.err.OperationalError:(2003 “Can’t connect to MySQL server on ‘localhost’ ([WinEr
190 1

相关产品

  • 云数据库 RDS MySQL 版
  • 云原生数据库 PolarDB
  • 云数据库 RDS PostgreSQL 版
  • 推荐镜像

    更多
    下一篇
    无影云桌面