obsidian-sync

7.3 KiB

Raw Blame History Unescape Escape

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

一、环境准备

PG版本 >= 9.4

1.修改wal 开启逻辑复制（需要重启数据库）

编辑 postgresql.conf 文件，配置如下

# 更改wal日志方式为logical（方式有：minimal、replica 、logical  ）
wal_level = logical  

# 更改solts最大数量（默认值为10），flink-cdc默认一张表占用一个slots
max_replication_slots = 20

# 更改wal发送最大进程数（默认值为10），这个值和上面的solts设置一样
max_wal_senders = 20     

# 中断那些停止活动超过指定毫秒数的复制连接，可以适当设置大一点（默认60s，0表示禁用）
wal_sender_timeout = 180s

执行语句检查是否修改成功

SHOW wal_level;

2.发布表

-- 设置发布为true
update pg_publication set puballtables=true where pubname is not null;

-- 把所有表进行发布
CREATE PUBLICATION dbz_publication FOR ALL TABLES;

-- 查询哪些表已经发布
select * from pg_publication_tables;

3.更改表的复制标识包含更新和删除的值

-- 更改复制标识包含更新和删除之前值（目的是为了确保表 t_user 在实时同步过程中能够正确地捕获并同步更新和删除的数据变化。如果不执行这两条语句，那么 t_user 表的复制标识可能默认为 NOTHING，这可能导致实时同步时丢失更新和删除的数据行信息，从而影响同步的准确性）
ALTER TABLE t_user REPLICA IDENTITY FULL;

-- 查看复制标识（为f标识说明设置成功，f（表示 full），否则为 n（表示 nothing），即复制标识未设置）
select relreplident from pg_class where relname='t_user';

二、source配置详解

source {
  Postgres-CDC {
    slot.name = "pgslot01"
    base-url = "jdbc:postgresql://10.2.0.5:5432/bjpt_hekou_v3?loggerLevel=OFF"
    username = "zr"
    password = "root@123"
	database-names = ["bjpt_hekou_v3"]
	schema-names = ["public"]
    table-names = ["bjpt_hekou_v3.public.ms_t_crjry"]
    startup.mode = "initial"
	table-names-config = [
      {
        table = "bjpt_hekou_v3.public.ms_t_crjry"
        primaryKeys = ["wybs"]
      }
    ]
  }
}

三、完整配置示例

env {
  parallelism = 1
  job.mode = "STREAMING"
}
source {
  Postgres-CDC {
    slot.name = "pgslot01"
    base-url = "jdbc:postgresql://10.2.0.5:5432/bjpt_hekou_v3?loggerLevel=OFF"
    username = "zr"
    password = "root@123"
	database-names = ["bjpt_hekou_v3"]
	schema-names = ["public"]
    table-names = ["bjpt_hekou_v3.public.ms_t_crjry"]
    startup.mode = "initial"
	table-names-config = [
      {
        table = "bjpt_hekou_v3.public.ms_t_crjry"
        primaryKeys = ["wybs"]
      }
    ]
  }
}
sink {
	Doris {
        fenodes = "172.31.51.142:8030"
        query-port = 9030
        username = admin
        password = "6G_FahdUxAh@K"
        schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
        database = "bjbj"
        table = "cdc_test_ms_t_crjry_pg"
        sink.enable-2pc = "true"
        sink.enable-delete = "true"
        sink.label-prefix = "cdc_test_ms_t_crjry_pg"
        doris.config = {
            format="json"
            read_json_by_line="true"
        }
    }
}

四、注意点

必须指定slot.name名称,并且重启任务时需要注意slot名称是否存在 !
重启任务时如果数据库提示 ‘replication slot xxxx is active’ 查询表查看 active 字段，需要等待一段时间，pg会将active改为false状态

select * from pg_replication_slots

! 还可以选择删除复制槽，但是需要先将目标表数据清除(相当于重新消费) 执行一下语句删除slot

SELECT * FROM pg_drop_replication_slot('slot_name');

五、原理

参考文章：剖析postgresql逻辑复制原理 - 知乎 POSTGRESQL 逻辑复制槽，what when how ,check and monitor 这一篇-腾讯云开发者社区-腾讯云 PostgreSQL复制槽与WAL文件保护：深入解析与实战测试_flzhang_132的技术博客_51CTO博客

PG 有2种数据复制方式：流复制（物理复制）、逻辑复制。流复制和逻辑复制是对wal日志的不同使用。PG-CDC 是基于逻辑复制实现。

一、流复制

流复制承载了pg主备之间数据同步的功能，它的实现方式是将wal日志中记录的内容按照确切的块地址逐字节的拷贝到备库，因此主备之间数据分布是一致的，这意味着在主备机器上，同一条记录的ctid是相同的。

二、逻辑复制

逻辑复制同步数据的原理是，在wal日志产生的数据库上，由逻辑解析模块对wal日志进行初步的解析,它的解析结果为ReorderBufferChange（可以简单理解为HeapTupleData），再由pgoutput plugin对中间结果进行过滤(过滤出insert、delete、update操作)和消息化拼接后，然后将其发送到订阅端，订阅端根据接收到的HeapTupleData重新对其执行insert、delete、update的操作。(这里要注意，流复制是将数据从walrecord拷贝到数据页，逻辑复制是将数据重新执行一次insert、update或delete)。从原理可以看出逻辑复制的效率比流复制低。 !

2.1 逻辑复制槽 slot

复制槽的作用：

保护WAL文件：确保主库在所有的备库收到WAL之前不会移除。
持久化状态信息：复制槽的状态信息是持久化的，即使从库断开或主库重启这些信息也不会丢失
协调主备数据一致性：复制槽通过restart_lsn 来标记备库还需要的WAL位置在逻辑复制中，复制槽代表着数据变更流。跟物理复制槽一样，逻辑复制槽也可以保证复制异常中断后，相关的wal日志不被删除，以保证复制重连后仍然可以继续解析wal日志（官方解释：复制槽提供了一种办法确保主库不会“删除”还未发送到备库的WAL日志，也不会删除备库需要的多版本，即使备库掉线）。一个数据库可以有多个复制槽，一个复制槽只有一个outputplugin，一个复制槽代表一条复制链路。复制槽本质上是用来管理复制链路的。不同于流复制可以没有复制槽，逻辑复制是必须有复制槽的。

2.2 outputplugin

test_decoding：这是一个outputplugin样例，相当于output plugin原始形态。官方文档说这是一个template，但是它仍然可以解析。这个output plugin是pg自带的，但需要在contrib中编译。 pgoutput：发布订阅模式的默认outputplugin。在发布订阅中，walsender进程使用该outputplugin逻辑解码wal日志。 decoder_raw:解析成sql文本格式。这个不是pg自带的。 wal2json：这个outputplugin会将wal日志信息转化为json格式

7.3 KiB Raw Blame History Unescape Escape