北京高端网站建设,做网站好还是做商城好,海南在线人才在线,企业介绍ppt案例欣赏PostgreSQL 变化数据捕捉#xff08;CDC#xff09;基于CDC#xff08;变更数据捕捉#xff09;的增量数据集成总体步骤#xff1a;1.捕获源数据库中的更改数据2.将变更的数据转换为您的消费者可以接受的格式3.将数据发布到消费者或目标数据库PostgreSQL支持触发器#x…PostgreSQL 变化数据捕捉CDC基于CDC变更数据捕捉的增量数据集成总体步骤1.捕获源数据库中的更改数据2.将变更的数据转换为您的消费者可以接受的格式3.将数据发布到消费者或目标数据库PostgreSQL支持触发器trigger和预写日志WAL两种CDC机制。如果您希望 Postgres 数据更改发生时逐行流式传输则需要逻辑解码或 Postgres 逻辑复制功能。使用 Postgres 逻辑解码逻辑解码是 PostgreSQL 的基于日志的 CDC逻辑复制的正式名称。逻辑解码使用 PostgreSQL 预写日志的内容来存储数据库中发生的所有活动。step1修改PostgreSQL数据库配置postgresql.conf文件wal_level logicalmax_replication_slots 10max_wal_senders 20wal_level 设置为 logical允许 WAL 日志记录逻辑解码所需的信息。max_replication_slots 确保 max_replication_slots 使用 WAL 的 PostgreSQL 连接器的数量加上您的数据库使用的其他复制槽的数量。max_wal_senders 指定 WAL 的最大并发连接数的参数确保 max_wal_senders 至少是逻辑复制槽数的两倍。例如如果您的数据库总共使用 10 个复制槽则该 max_wal_senders 值必须为 20 或更大。配置修改以后需要重复 PostgreSQL 服务生效配置。step2为需要同步的数据库db创建逻辑复制插槽replication slot关于复制 SQL 函数更多详情可以参考官方文档9.26.系统管理函数介绍 http://www.postgres.cn/docs/12/functions-admin.html函数pg_create_logical_replication_slot(slot_name name, plugin name)返回类型(slot_name name, lsn pg_lsn)说明使用输出插件plugin创建一个名为 slot_name的新逻辑解码复制槽。创建时需要指定逻辑复制插槽名称和输出插件SELECT pg_create_logical_replication_slot(replication_slot01, test_decoding); -- 使用 test_decoding 输出插件SELECT pg_create_logical_replication_slot(replication_slot01, pgoutput); -- 使用 pgoutput 输出插件注意权限如果用户没有权限可能会报错ERROR: must be superuser or replication role to use replication slots授权\c - postgresALTER ROLE test REPLICATION; --流复制权限\du验证插槽是否创建成功SELECT * FROM pg_replication_slots;testdb01 SELECT * FROM pg_replication_slots; slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn ----------------------------------------------------------------------------------------------------------------------------------------------------- replication_slot01 | test_decoding | logical | 16437 | testdb01 | f | f | | | 585 | 0/1807EA8 | 0/1807EE0 replication_slot02 | pgoutput | logical | 16437 | testdb01 | f | f | | | 585 | 0/1809CD8 | 0/1809D10(2 rows)testdb01 注意1逻辑复制插槽名称每个逻辑复制插槽都有一个名称创建时需要指定名称可以包含小写字母、数字和下划线2输出插件创建逻辑复制插槽时需要指定输出插件输出插件有test_decoding、pgoutput、wal2json等 test_decoding 输出插件 PostgreSQL 9.4原生附带了 test_decoding 输出插件$PG_HOME/lib 目录下对应有 test_decoding.so如果您的消费者支持 test_decoding 则可以使用 test_decoding 输出插件。 pgoutput 输出插件 PostgreSQL 10原生附带了 pgoutput 输出插件$PG_HOME/lib 目录下对应有 pgoutput.so如果您的消费者支持 pgoutput 则可以使用 pgoutput 输出插件。pgoutput插件输出的是二进制的数据 wal2json 输出插件 wal2json 是另一个流行的逻辑解码输出插件PostgreSQL原生不携带、需要数据库服务单独安装插件才能使用。wal2json输出的是json格式的数据step3为数据库中的所有表或指定表创建一个发布。如果以指定表方式创建发布则后面可以在发布中管理添加、删除表帮助说明\h create publicationDROP PUBLICATION IF EXISTS pub01;CREATE PUBLICATION pub01 FOR TABLE test01;或者DROP PUBLICATION IF EXISTS pub01;CREATE PUBLICATION pub01 FOR TABLE test01, test02, test03;或者DROP PUBLICATION IF EXISTS pub01;CREATE PUBLICATION pub01 FOR ALL TABLES;查看创建发布的结果testdb01 select * from pg_publication; oid | pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate -------------------------------------------------------------------------------------- 24629 | pub01 | 16436 | f | t | t | t | t(1 row)也可以执行查看testdb01 \dRp List of publications Name | Owner | All tables | Inserts | Updates | Deletes | Truncates ---------------------------------------------------------------- pub01 | test | f | t | t | t | t(1 row)创建发布时还可以选择在发布中包含哪些操作。例如下面仅创建table01表 仅包含INSERT和UPDATE的发布CREATE PUBLICATION pub01 FOR TABLE table01 WITH (publish INSERT, UPDATE);注意非超级用户创建发布只能指定表创建、不能 FOR ALL TABLES 创建否则报错ERROR: must be superuser to create FOR ALL TABLES publicationFOR ALL TABLES 只允许超级用户创建发布或者创建后由超级用户修改已创建的发布的配置开启foralltables。testdb01 \c - postgrestestdb01# update pg_publication set puballtablestrue where pubname is not null; -- 设置puballtables开关testdb01# select * from pg_publication; oid | pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate -------------------------------------------------------------------------------------- 24629 | pub01 | 16436 | t | t | t | t | t(1 row)step4验证指定的表是否在发布中testdb01 SELECT * FROM pg_publication_tables WHERE pubnamepub01;注意创建发布不会开始复制。它只为未来的订阅者定义一个分组和过滤逻辑。step5查看逻辑复制效果获取结果的相关命令pg_logical_slot_get_changes : 查询并删除数据。仅在第一次返回结果多次调用可能会返回空结果集这意味着当get命令执行时结果会被提供和删除这增强了我们编写使用这些事件创建表副本的逻辑的能力。函数pg_logical_slot_get_changes(slot_name name, upto_lsn pg_lsn, upto_nchanges int, VARIADIC options text[])返回类型(lsn pg_lsn, xid xid, data text)说明返回槽slot_name中的改变从上一次已经被消费的点开始返回。 如果upto_lsn和upto_nchanges为 NULL逻辑解码将一 直继续到 WAL 的末尾。如果upto_lsn为非 NULL解码将只包括那些在指 定 LSN 之前提交的事务。如果upto_nchanges为非 NULL 解码将在其产生的行数超过指定值后停止。不过要注意 被返回的实际行数可能更大因为对这个限制的检查只会在增加了解码每个新的提交事务产生 的行之后进行。pg_logical_slot_get_binary_changes变化数据以bytea返回。pg_logical_slot_peek_changes : 只查询不删数据。多次调用每次都会返回相同的结果。是另一个 PostgreSQL 命令用于在不使用 WAL 条目的情况下查看更改。函数pg_logical_slot_peek_changes(slot_name name, upto_lsn pg_lsn, upto_nchanges int, VARIADIC options text[])返回类型(lsn text, xid xid, data text)说明行为就像pg_logical_slot_get_changes()函数 不过改变不会被消费 即在未来的调用中还会返回这些改变。pg_logical_slot_peek_binary_changes变化数据以bytea返回。关于复制 SQL 函数更多详情可以参考官方文档9.26.系统管理函数介绍 http://www.postgres.cn/docs/12/functions-admin.html测试表中I/D/U变更数据操作然后查看逻辑复制效果testdb01 SELECT * FROM pg_logical_slot_peek_changes(replication_slot01, NULL, NULL); -- 只查询不删数据......testdb01 SELECT * FROM pg_logical_slot_get_changes(replication_slot01, NULL, NULL); -- 查询并删除数据 lsn | xid | data ---------------------------------------------------------------------------------------------------------- 0/180A050 | 586 | BEGIN 586 0/180A050 | 586 | table public.test01: INSERT: id[bigint]:1 info[character varying]:aa cnt[integer]:123 0/180A160 | 586 | table public.test01: INSERT: id[bigint]:2 info[character varying]:bb cnt[integer]:456 0/180A1E8 | 586 | table public.test01: INSERT: id[bigint]:3 info[character varying]:cc cnt[integer]:55 0/180A2A0 | 586 | COMMIT 586 0/180A2A0 | 587 | BEGIN 587 0/180A2A0 | 587 | table public.test01: INSERT: id[bigint]:4 info[character varying]:uuu cnt[integer]:66 0/180A358 | 587 | COMMIT 587 0/180A358 | 588 | BEGIN 588 0/180A358 | 588 | table public.test01: DELETE: id[bigint]:2 0/180A3D0 | 588 | COMMIT 588 0/180A3D0 | 589 | BEGIN 589 0/180A3D0 | 589 | table public.test01: UPDATE: id[bigint]:3 info[character varying]:xxxx cnt[integer]:55 0/180A458 | 589 | COMMIT 589(14 rows)testdb01 SELECT * FROM pg_logical_slot_get_changes(replication_slot01, NULL, NULL); -- 查询并删除数据 lsn | xid | data ----------------(0 rows)testdb01 pgoutput输出插件的结果查看则执行testdb01 SELECT * FROM pg_logical_slot_peek_binary_changes(replication_slot02, null, null, proto_version, 1, publication_names, pub01); -- 查询但不删除数据......testdb01 SELECT * FROM pg_logical_slot_get_binary_changes(replication_slot02, null, null, proto_version, 1, publication_names, pub01); -- 查询并删除数据 lsn | xid | data ---------------------------------------------------------------------------------------------------------------------------------------------- 0/1809DF8 | 585 | \x420000000001809fd0000296c5f556e57d00000249 0/1809DF8 | 585 | \x52000060417075626c696300746573743031006400030169640000000014ffffffff00696e666f00000004130000006800636e740000000017ffffffff 0/1809DF8 | 585 | \x44000060414b00037400000001316e6e 0/1809F40 | 585 | \x44000060414b00037400000001346e6e 0/1809F88 | 585 | \x44000060414b00037400000001336e6e 0/180A000 | 585 | \x43000000000001809fd0000000000180a000000296c5f556e57d 0/180A050 | 586 | \x42000000000180a270000296c5f7bb75300000024a 0/180A050 | 586 | \x49000060414e0003740000000131740000000261617400000003313233 0/180A160 | 586 | \x49000060414e0003740000000132740000000262627400000003343536 0/180A1E8 | 586 | \x49000060414e00037400000001337400000002636374000000023535 0/180A2A0 | 586 | \x4300000000000180a270000000000180a2a0000296c5f7bb7530 0/180A2A0 | 587 | \x42000000000180a328000296c5f7d8abe90000024b 0/180A2A0 | 587 | \x49000060414e0003740000000134740000000375757574000000023636 0/180A358 | 587 | \x4300000000000180a328000000000180a358000296c5f7d8abe9 0/180A358 | 588 | \x42000000000180a3a0000296c5f80ae6ca0000024c 0/180A358 | 588 | \x44000060414b00037400000001326e6e 0/180A3D0 | 588 | \x4300000000000180a3a0000000000180a3d0000296c5f80ae6ca 0/180A3D0 | 589 | \x42000000000180a428000296c5f826073e0000024d 0/180A3D0 | 589 | \x55000060414e000374000000013374000000047878787874000000023535 0/180A458 | 589 | \x4300000000000180a428000000000180a458000296c5f826073e(20 rows)testdb01 SELECT * FROM pg_replication_slots; slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn ----------------------------------------------------------------------------------------------------------------------------------------------------- replication_slot01 | test_decoding | logical | 16437 | testdb01 | f | f | | | 590 | 0/180A540 | 0/180A578 replication_slot02 | pgoutput | logical | 16437 | testdb01 | f | f | | | 590 | 0/180A540 | 0/180A578(2 rows)PG logical replication plug-in (pgoutput) 输出的消息格式详情可以参考PG官方说明https://www.postgresql.org/docs/10/protocol-logicalrep-message-formats.htmlstep6销毁逻辑复制插槽SELECT pg_drop_replication_slot(replication_slot01);使用复制插槽的注意事项1每个插槽只有一个输出插件创建插槽时由您来选择使用哪个。2每个插槽仅提供来自一个数据库的更改。3一个数据库可以有多个插槽。4每个数据更改通常在每个插槽中发出一次。5但是当 Postgres 实例重新启动时插槽可能会重新发出更改。消费者必须处理这种情况。6未使用的插槽对 Postgres 实例的可用性构成威胁。Postgres 将为这些未使用的更改保存所有 WAL 文件。这可能导致存储溢出。关于 PostgreSQL WAL 消费者可以获取 Postgres 逻辑解码流的任何应用程序都是 PostgreSQL WAL 消费者。pg_recvlogicalpg_recvlogical 是一个 PostgreSQL 应用程序他是 PostgreSQL 原生的逻辑解码工具它可以管理槽并使用槽中的流。它包含在 Postgres 发行版中因此它可能已经随 PostgreSQL 一起安装在 $PG_HOME/bin 目录下。pg_recvlogical 使用默认的 test_decoding 逻辑解码插件pg安装完后会在pg安装目录的lib目录下创建test_decoding链接库文件。pg_recvlogical 使用介绍创建逻辑复制插槽$ pg_recvlogical --create-slot -S replication_slot01 -d testdb01启动复制槽解码启动后会实时的将日志解码到制定的文件中也可以不启动、等需要解码时再启动解码$ pg_recvlogical --start -S replication_slot01 -d testdb01 -f replication_slot01_decoding.log 查看解码结果$ cat replication_slot01_decoding.log查看数据库lsnpostgres# select pg_current_wal_lsn();使用pg_recvlogical进行日志区间解码$ pg_recvlogical --start -S replication_slot01 -d testdb01 -I 4C/180B020 -E 4C/180B1EE -f ret.log
