Demo 演示 | 基于 Apache SeaTunnel 实现从 MySQL CDC 到 PostgreSQL 全量数据同步

本文具体演示了怎样通过 Apache SeaTunnel 2.3.9 实现 MySQL  CDC ​ 到 PostgreSQL 的全量数据同步。
话不多说，我们开始学习 MySQL 同步到 PostgreSQL 场景：
版本要求：

• MySQL --> MySQL-8.3
  
• PostgreSQL --> PostgreSQL-13.2
  
• Apache SeaTunnel --> Apache-SeaTunnel-2.3.9
  前置预备
  
  确认版本信息
  
  1. -- 查看版本信息
  2. select version();

  复制代码
  
  
  
  
  
  开启主从复制
  
  1. -- 查看配置信息
  2. show variables where variable_name in ('log_bin', 'binlog_format', 'binlog_row_image', 'gtid_mode', 'enforce_gtid_consistency');

  复制代码
  
  
  MySQL CDC 数据同步，必要读取 MySQL 的 binlog 日记。而且把 SeaTunnel 的集群节点作为 MySQL 集群的一个从节点。
  以是这里必要先确认 MySQL 开启了 binlog，以及开启了主从复制模式。
  注：MySQL 的 8.0 版本以上，binlog 是默认开启的，但主从复制模式必要手动开启。
  1. -- 开启主从复制,需要按顺序执行
  2. -- SET GLOBAL gtid_mode=OFF;
  3. -- SET GLOBAL enforce_gtid_consistency=OFF;
  4. SET GLOBAL gtid_mode=OFF_PERMISSIVE;
  5. SET GLOBAL gtid_mode=ON_PERMISSIVE;
  6. SET GLOBAL enforce_gtid_consistency=ON;
  7. SET GLOBAL gtid_mode=ON;

  复制代码
  
  
  开启用户权限
  
  用户权限必要具有复制权限，以是这里有两个核心的权限，REPLICATION SLAVE 和 REPLICATION CLIENT，授权完成以后完成一下权限的革新。
  1. -- 用户授权
  2. CREATE USER 'test'@'%' IDENTIFIED BY 'password';
  3. GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'test' ;
  4. flush privileges;

  复制代码
  SeaTunnel 集群
  
  集群日记
  
  每个作业独立日记文件：log4j2.properties
  起首是我们集群内里默认的设置是把日记输出到一个同一的文件内里去，但是在生产上通常环境下对任务的管理都是基于每个独立的任务，这里就必要把日记的输出模式改成每个作业输出独立的日记文件，如许会比力方便生产过程中去对日记做一个监控以及标题的排查，方便我们按照作业做任务的管理。
  1. ############################ log output to file    #############################
  2. # rootLogger.appenderRef.file.ref = fileAppender
  3. # 把日志的输出模式改成每个作业输出独立的日志文件
  4. rootLogger.appenderRef.file.ref = routingAppender
  5. ############################ log output to file    #############################

  复制代码
  客户端设置
  
  生产上的集群一样平常环境下都会把安装到根目次的 opt 文件夹下，客户端也发起把 SEATUNNEL_HOME 指向到 /opt/seatunnel 目次下。
  留意：如果现实安装有多个版本，大概并不在这个目次下，发起去创建一个软链接，只管与服务器摆设目次同等，制止出现有一些类找不到。
  当前版本利用 SeaTunnel 的脚本去提交下令，会引用到客户端的一些环境变量，包罗比如说 class 类的路径，这里会利用绝对路径。
  如果服务端和客户端不同等，那么在提交到集群的时间大概会报错。
  1. # 创建软链接
  2. ln -s /opt/apache-seatunnel-2.3.9 /opt/seatunnel
  3. # 设置环境变量
  4. export SEATUNNEL_HOME=/opt/seatunnel

  复制代码
  环境变量设置
  
  如果是 Linus 服务器，环境变量客户端保举按照官方的示例，把环境变量文件的设置项写到 /etc/profile.d 下。
  1. echo 'export SEATUNNEL_HOME=/opt/seatunnel' >> /etc/profile.d/seatunnel.sh
  2. echo 'export PATH=$SEATUNNEL_HOME/bin:$PATH' >> /etc/profile.d/seatunnel.sh
  3. source /etc/profile.d/seatunnel.sh

  复制代码
  任务设置
  
  注：这里设置并不是包罗全部的设置项，只是把生产上常用到的一些设置选项做了一个阐明。
  1. env {
  2.   job.mode = "STREAMING"
  3.   job.name = "DEMO"
  4.   parallelism = 3
  5.   checkpoint.interval = 30000 # 30s
  6.   checkpoint.timeout = 30000 # 30s
  8.   job.retry.times = 3
  9.   job.retry.interval.seconds = 3 # 3s
  10. }

  复制代码
  起首是 env 模块，由于它是一个流式的，以是必要指定它的设置模式是 STREAMING。
  然后是设置任务名，现实生产环境上可以按照库大概是按照表名字做一个匿名，如许看到任务名就可以知道它所对应的作业，便于对任务做管理。
  别的一个是并行度，并行度这里利用的是 3，这个并行度的设置，可以依据现实的集群的巨细，以及数据库的这个设置环境做一些调解。
  其次是查抄点的频率，这里设置的是 30 秒。如果说有更高的要求，可以调解为 10 秒大概更短一点；
  设置查抄点的超时时间。如果 Checkpoint 的凌驾了某一个时间，要判定为作业失败，这里也设置为 30 秒；设置作业的主动重试次数，这里一样平常接纳的都是设置为 3 次；每次从事的时间隔断设置为 3 秒，固然这里也可以时间更长一点。
  1. hocon
  2. source {
  3.   MySQL-CDC {
  4.     base-url = "jdbc:mysql://192.168.8.101:3306/test?serverTimezone=Asia/Shanghai"
  5.     username = "test"
  6.     password = "123456"
  7.    
  8.     database-names = ["test"]
  9.     # table-names = ["test.test_001","test.test_002"]
  10.     table-pattern = "test\\.test_.*"# 第一个点为普通字符串，需做转义； 第二个点表示匹配1个任意字符
  11.     table-names-config = [
  12.         {"table":"test.test_002","primaryKeys":["id"]}
  13.         ]
  14.    
  15.     startup.mode = "initial"# 先全量同步历史，再增量
  16.     snapshot.split.size = "8096"
  17.     snapshot.fetch.size = "1024"
  18.     server-id = "6500-8500"
  19.     connect.timeout.ms = 30000
  20.     connect.max-retries = 3
  21.     connection.pool.size = 20
  22.    
  23.     exactly_once = false   # 数据分析场景，关闭精确一致性，允许一定重复和缺失 提升性能
  24.     schema-changes.enabled = true# 模式演变 开启 避免反复修改表字段,但可能影响下游任务。add rename drop
  25.   }
  26. }

  复制代码
  别的一个核心重点就是 MySQL CDC 自己的设置项。这里必要阐明一下，最好指定一下 MySQL 毗连所利用的时区，制止出现 datetime 大概 timestamp 数据抽取的时间出现一些时区不对等的环境。
  其次是 username 和 password，这里必要设置具有主从复制从节点权限，复制读取 binlog 日记权限的一个账号，可以大概复制全部库下的全部表的查询权限，可以大概读取到全部表的日记。
  通常环境下做这个数据库设置的时间，都是每个 database 会放到一个任务内里，以是在这里指定一下，只去读取 database 为 test 对应的表。
  根据自己的必要做两个设置：一个是我们的表名，一个是表的正则表达式匹配区，如果是初次同步，要同步的表比力多，大概要同步整个库，大概整个库下面的部门表，它有一些同一的规则，比如说某个前缀大概某个后缀，那么保举利用正则表达式去做这个数据同步。
  这里阐明一下，设置正则表达式匹配的时间，我们要包罗 Database 名字和表名字。而 database 名和表名的中心要有一个点，而这个点在正则表达式内里它要体现一个恣意字符，以是在这里我们要对它举行转义。
  ​转义的方式是挑拨用两个反斜杠​，比如说这里要匹配以 test 为开头的表，那么在 test 的下划线的反面有一个点，代表的寄义是我要匹配恣意数目的恣意字符。
  第二个点代表匹配一个字符，然后新代表匹配恣意数目的这个字符是零到 N 如许的话我们就可以实现 test 库下的 test_为前缀表名。
  其次是我们对这个表还可以指定一些额外的设置。比如说我们如果说我们这里有张 test_002，test_002 它是没有主键列的。但是我们做数据同步的时间，可以指定逻辑上的主键方便我们做数据同步。
  启动模式
  
  ​其次是我们的启动模式​。启动模式这里也是一个默认值，就是一个初始化的利用，可以实现先全量，再增量。这也是我们生产常见的利用。其次是分片巨细以及每批次获取的数目，这里利用默认值即可。如果服务器性能比力好，大概集群数目比力大，也可以做一些调解。
  比力关键的是这里有一个 server-id ，MySQL 在举行同步的时间，SeaTunnel 的集群是要作为他要把自己伪装成 Mysql 的一个重复节点。那么在 MySQL 做主从复制的时间，要求每个重节点，就集群内的这些节点都要有自己唯一的 server-id。
  在这里如果你不设置，它会利用一个默认值，而官方是保举我们去指定的，各人可以在官方文档内里找到。这里有一个要求，server-id 的范围，肯定要大于并行的数目，否则的话会报错的。
  超时时间
  
  ​接下来是毗连的超时时间​。如果说我们的数据量比力大，那么这个超时时间可以设置的久一点。尚有主动重试之间毗连词的巨细，如果表数据量比力多，毗连词也可以得当调大一点。
  正确同等性
  
  像我们大概更多的时间是做这个 CDC 的数据同步，它是基于一个分析场景的，不会涉及到一些非常关键的业务，就是不会有特别大的这种正确同等性的高要求。以是发起是关闭正确同等性。如果确实是有这种强同等性的要求的话，是可以开启的。由于如果说你开启的话，这个同步的性能就会有些降落的。
  Schema evolution
  
  别的一个是模式演变（Schema evolution）。模式演变我是发起各人开启，如许可以制止我们去比如说源端做一些新增字段大概说删除字段的时间，我们可以不消去修改我们这个任务，它可以主动跟随我们源端的修改做一些改变。但这里也会带来一个影响，就是我们卑鄙任务，如果说我依靠的某张表名字发主动发生了变动，那么我们卑鄙的这个任务的装备语句大概就会报错了。以是各人根据自己的一个环境做一个均衡。
  如今这个模式演变（Schema evolution），根据官方的阐明支持 add column、drop column、rename column 和 modify column. 不是全部的 ddl 语句全部支持像创建表、删除表这些是无法辨认到的。
  这确实是一个非常好的功能，发起各人开启。
  1. sink {
  2.   jdbc {
  3.         url = "jdbc:postgresql://192.168.8.101:5432/test"
  4.         driver = "org.postgresql.Driver"
  5.         user = "postgres"
  6.         password = "123456"
  8.         generate_sink_sql = true
  9.         database = "test"
  10.         table = "${database_name}.${table_name}"
  11.         schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
  12.         data_save_mode = "APPEND_DATA"
  13.         # enable_upsert = false
  14.     }
  15. }

  复制代码
  别的是 Sink 端，Sink 端我们是要把这个数据插入到同步到 PostgreSQL 数据库，在这里要设置 PostgreSQL 的点击信息，这里除了设置驱动用户名和暗码以外，我们这里会保举各人利用天生式 SQL 的功能。
  开启以后它可以去主动基于源端的表布局去天生,我们同步到 PostgreSQL 内里的这些间表语句，插入一句删除语句，尚有这个基于主键的 update 语句，都可以自己天生，制止我们一些很复杂的逻辑。
  概念区分
  
  别的我们再把这个数据同步到 PostgreSQL 的时间要留意一下。由于 MySQL 只有 Database 和 Table 这两层的，而没有 Schema 的概念。
  而在 PostgreSQL 内里它是有三层，它分别有 Database、Schema 和 Table。那这就要求如果说我想把这个表和数据都同步到 PostgreSQL 的 test database 下，那么这里就要指定。
  别的由于我们在做数据同步的时间，会想要借用它的数据主动建表功能。以是说你去毗连这个 PostgreSQL 的时间，毗连用户最好有建表的权限。
  占位符
  
  别的建表的时间，这里可以去利用 Sink 的一个占位符功能。这个功能非常好用，可以基于由于我们源端可以有很多表，那么很多表它在举行同步的时间，它每张表都会有差异的名。是这里我们就可以利用占位符让它主动去天生，我们要创建什么表就不消挨个去指定了。这里也是支持一些拼接符号的，具体的话可以参考我们文官方的文档。
  生存模式
  
  schema_save_mode 也就是我们模式的生存模式，我们在做整库同步的时间黑白常好用的,可以帮我们节流线表步调，可以完成主动建表。
  别的一个是 APPEND_DATA。如果目的端已经有数据已经同步过一批任务了，这个时间可以制止删除一些数据，相对来讲比力安全。如果有特别需求的话，也可以去指定其他的模式，按照官方文档做修改检测。
  别的必要阐明的是 enable_upsert，如果说我们可以大概包管数据源端它的数据是不会重复的,可以把 upsert 关闭。如许的话可以极大的提拔数据同步的性能。但是如果说你不能包管你的数据源是没有重复的，那么我们发起还是把这个选项设置为 false。它可以基于主键做一些 upsert。具体的参数还是可以参考我们官方的文档
  任务提交与监控
  在设置文件编写完成后，我们可以通过 SeaTunnel 的下令行工具提交任务。
  以下是任务提交的示例下令：
  1. ./bin/start-seatunnel.sh --config /path/to/config.yaml --async

  复制代码
  关键参数阐明：
  
  
  
  • config：指定设置文件的路径。
    
  • async：体现以异步方式提交任务，任务提交后下令行可以立刻退出，任务会在背景继承实行。
    
  任务提交后，我们可以通过 SeaTunnel 的集群 UI 界面监控任务的实行状态。在 2.3.9 版本中，SeaTunnel 提供了一个直观的 UI 界面，可以检察任务的日记、实行状态以及数据量等信息。
  数据同步演示
  在本次演示中（可以回到文章开头看具体的演示大概 B 站搜刮干系视频），我们创建了两张表 test_001 和 test_002，并在 MySQL 中插入了一些数据。通过 SeaTunnel 的同步任务，这些数据被乐成同步到了 PostgreSQL 中。我们还演示了数据的插入、删除、更新以及表布局的变动利用，SeaTunnel 都可以大概及时地将这些变动同步到 PostgreSQL。
  关键点
  
  
  
  
  • 表布局同步：SeaTunnel 支持表布局的主动同步。当源端 MySQL 表布局发生厘革时，目的端 PostgreSQL 的表布局也会主动更新。
    
  • 数据同等性：SeaTunnel 包管了数据的同等性，全部的插入、删除和更新利用都可以大概正确地同步到目的数据库。