Help us improve the AWS re:Post Knowledge Center by sharing your feedback in a brief survey. Your input can influence how we create and update our content to better support your AWS journey.
如何使用逻辑复制在 Amazon RDS for PostgreSQL 数据库实例之间复制表？
我想使用逻辑复制在我的 Amazon Relational Database Service (Amazon RDS) for PostgreSQL 数据库实例中的数据库之间复制表。我不想使用扩展。
解决方法
Amazon RDS for PostgreSQL 支持 PostgreSQL 10.4 及更高版本的逻辑复制。Amazon Aurora PostgreSQL 兼容版 2.2.0 及更高版本支持 PostgreSQL 10.6 及更高版本的逻辑复制。
有关详细信息，请参阅为 Amazon RDSD for PostgreSQL 执行逻辑复制。
以下解决方案将两个源表复制到两个目标表。
启用逻辑复制
完成以下步骤：
- 创建自定义参数组，并将 rds.logical_replication 参数设置为 1。
**注意：**由于 rds.logical_replication 参数是静态参数，因此必须重启数据库实例。重启数据库实例后，wal_level 参数更改为 logical。 - 将参数组与数据库实例相关联。
- 运行以下查询以验证 wal_level 是否为 logical 且 rds.logical_replication 是否已启用：
预期输出：SELECT name,setting FROM pg_settings WHERE name IN ('wal_level','rds.logical_replication');name | setting -------------------------+--------- rds.logical_replication | on wal_level | logical (2 rows)
创建源表并插入数据
连接到 PostgreSQL 数据库实例，然后在源数据库中完成以下步骤：
-
运行以下命令以创建源表：
CREATE TABLE reptab1 (slno int primary key); CREATE TABLE reptab2 (name varchar(20)); -
运行以下命令以将数据添加到源表中：
INSERT INTO reptab1 VALUES (generate_series(1,1000)); INSERT INTO reptab2 SELECT SUBSTR ('ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789',((random()*(36-1)+1)::integer),1) FROM generate_series(1,50);
为源表创建发布
在源数据库中，完成以下步骤：
-
运行以下命令以创建这两个表的发布：
CREATE PUBLICATION testpub FOR TABLE reptab1, reptab2; -
运行以下查询以验证发布的详细信息是否正确：
SELECT * FROM pg_publication;预期输出：
oid | pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate | pubviaroot --------+---------+----------+--------------+-----------+-----------+-----------+-------------+------------ 115069 | testpub | 16395 | f | t | t | t | t | f (1 row) -
运行以下查询以验证源表是否在发布中：
SELECT * FROM pg_publication;预期输出：
pubname | schemaname | tablename ---------+------------+----------- testpub | public | reptab1 testpub | public | reptab2 (2 rows)
连接到目标数据库并创建目标表
在目标数据库中，完成以下步骤：
-
运行以下命令以创建目标表：
CREATE TABLE reptab1 (slno int primary key); CREATE TABLE reptab2 (name varchar(20));**注意：**使用与源表相同的名称。
-
运行以下查询以验证数据是否不在目标表中。
表 1：SELECT count(*) FROM reptab1;预期输出：
count ------- 0 (1 row)表 2：
SELECT count(*) FROM reptab2;预期输出：
count ------- 0 (1 row)
在目标数据库中创建订阅
在创建订阅之前，运行以下命令以确认您没有在数据库日志中存储用户名和密码的纯文本版本：
SET log_min_messages to 'PANIC'; SET log_statement to NONE;
要在目标数据库中创建订阅，请完成以下步骤：
-
运行以下命令以创建订阅：
CREATE SUBSCRIPTION testsub CONNECTION 'host=source RDS/host endpoint port=5432 dbname=source_db_name user=user password=password' PUBLICATION testpub;**注意：**将 source RDS/host endpoint 替换为源数据库实例的端点，将 source_db_name 替换为数据库实例的名称。将 user 替换为您的用户名，将 password 替换为您的密码。
-
运行以下查询以验证订阅是否有效：
SELECT oid,subname,subenabled,subslotname,subpublications FROM pg_subscription;预期输出：
oid | subname | subenabled | subslotname | subpublications -------+---------+------------+-------------+----------------- 16434 | testsub | t | testsub | {testpub} (1 row) -
运行以下查询以验证数据是否在目标表中。
表 1：SELECT count(*) FROM reptab1;预期输出：
count ------- 1000 (1 row)表 2：
SELECT count(*) FROM reptab2;预期输出：
count ------- 50 (1 row)
在源数据库中验证复制槽的详细信息
当您在目标数据库中创建订阅时，目标数据库会在源数据库中创建一个复制槽。
要验证复制槽的详细信息，请在源数据库中运行以下查询：
SELECT * FROM pg_replication_slots;
预期输出：
slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | safe_wal_size ----------+----------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+--------------- testsub | pgoutput | logical | 115048 | source | f | t | 846 | | 6945 | 58/B4000568 | 58/B40005A0 | reserved | (1 row)
测试从源表的复制
在源数据库中，完成以下步骤：
-
运行以下命令将行添加到源表中：
INSERT INTO reptab1 VALUES(generate_series(1001,2000)); INSERT INTO reptab2 SELECT SUBSTR('ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789',((random()*(36-1)+1)::integer),1) FROM generate_series(1,50); -
运行以下查询以验证源表中的数据是否已更改。
表 1：SELECT count(*) FROM reptab1;预期输出：
count ------- 2000 (1 row)表 2：
SELECT count(*) FROM reptab2;预期输出：
count ------- 100 (1 row)
验证源表中的数据是否已复制到目标表
在目标数据库中，运行以下查询以确认源表中的数据已复制到目标表。
表 1：
SELECT count(*) FROM reptab1;
预期输出：
count ------- 2000 (1 row)
表 2：
SELECT count(*) FROM reptab2;
预期输出：
count ------- 100 (1 row)
清理复制槽并关闭逻辑复制
完成复制后，如果不再需要复制，请清空该槽并关闭逻辑复制。非活动复制槽会导致预写日志 (WAL) 文件在源数据库实例上不断堆积。WAL 文件可能会填满存储空间并导致停机。
完成以下步骤：
-
在目标数据库上，运行以下命令以移除订阅：
DROP SUBSCRIPTION testsub; -
在目标数据库上，运行以下查询以验证订阅是否已移除：
SELECT * FROM pg_subscription;预期输出：
oid | subdbid | subname | subowner | subenabled | subconninfo | subslotname | subsynccommit | subpublications ----+---------+---------+----------+------------+-------------+-------------+---------------+----------------- (0 rows)**注意：**当您移除目标数据库上的订阅时，目标数据库也会移除源数据库中的复制槽。
-
在源数据库上，运行以下查询以验证复制槽是否已从源数据库中移除：
SELECT * FROM pg_replication_slots;预期输出：
slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | safe_wal_size ----------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+-------------- (0 rows) -
在源数据库上，运行以下命令以移除发布：
DROP PUBLICATION testpub; -
在源数据库上，运行以下查询以验证发布是否已移除：
SELECT * FROM pg_publication; SELECT * FROM pg_publication_tables;预期输出：
oid | pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate | pubviaroot -----+---------+----------+--------------+-----------+-----------+-----------+-------------+------------ (0 rows) pubname | schemaname | tablename ---------+------------+----------- (0 rows) -
在数据库实例的自定义参数组中，将 rds.logical_replication 参数设置为 0。
**注意：**重启数据库实例以使更改生效。 -
根据您的使用情况查看 max_replication_slots、max_wal_senders、max_logical_replication_workers、max_worker_processes 和 max_sync_workers_per_subscription 参数。
-
运行以下查询以检查是否存在非活动复制槽以及槽的大小：
SELECT slot_name, pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(),restart_lsn)) AS replicationSlotLag, active FROM pg_replication_slots; -
（可选）运行以下命令以移除复制槽：
SELECT pg_drop_replication_slot('Your_slotname_name')
相关信息
PostgreSQL 网站上的 Replication（复制）
PostgreSQL 网站上的 Logical replication（逻辑复制）
