使用 PostgreSQL 数据库作为源 Amazon DMS

使用 Po Amazon stgreSQL 数据库实例的主用户账户作为 PostgreSQL 源终端节点的用户账户。主用户账户具有允许它设置 CDC 所需的角色。

如果您使用主用户账户以外的账户，请务必使用主账户为所使用的账户创建多个对象。有关更多信息，请参阅在不使用主用户账户的情况下迁移 Amazon RDS for PostgreSQL 数据库：

将数据库集群rds.logical_replication参数组中的参数设置为 1。此静态参数要求重启数据库实例才能生效。在应用该参数期间，Amazon DMS 设置 wal_level、max_wal_senders、max_replication_slots 和 max_connections 参数。这些参数更改可以增加预写日志 (WAL) 生成，因此，只有在您使用逻辑复制槽时才设置 rds.logical_replication。

该wal_sender_timeout参数结束非活动时间超过指定毫秒数的复制连接。Amazon由托管的 PostgreSQL 数据库的默认值为 30000 毫秒（30 秒）。将该值设置为 0（零）会禁用超时机制，这是 DMS 的有效设置。

设置wal_sender_timeout为非零值时，带有 CDC 的 DMS 任务至少需要 10000 毫秒（10 秒），如果该值介于 0 和 10000 之间，则会失败。将该值保持在 5 分钟以内，以免在 DMS 复制实例的多可用区故障转移期间造成延迟。

确保数据库集群max_worker_processes参数组中参数的值等于或大于max_logical_replication_workersautovacuum_max_workers、和的总组合值max_parallel_workers。大量的后台工作进程可能会影响小型实例上的应用程序工作负载。因此，如果设置的值max_worker_processes高于默认值，请监控数据库的性能。

选择要在其中创建对象的架构。默认架构为 public。确保架构存在并且可由 OtherThanMaster 账户访问。

使用除主账户之外的用户账户（此处为账户）登录 PostgreSQL 数据库实例。OtherThanMaster

awsdms_ddl_audit通过运行以下命令创建表，将以下代码objects_schema中的内容替换为要使用的架构名称。

CREATE TABLE objects_schema.awsdms_ddl_audit
(
  c_key    bigserial primary key,
  c_time   timestamp,    -- Informational
  c_user   varchar(64),  -- Informational: current_user
  c_txn    varchar(16),  -- Informational: current transaction
  c_tag    varchar(24),  -- Either 'CREATE TABLE' or 'ALTER TABLE' or 'DROP TABLE'
  c_oid    integer,      -- For future use - TG_OBJECTID
  c_name   varchar(64),  -- For future use - TG_OBJECTNAME
  c_schema varchar(64),  -- For future use - TG_SCHEMANAME. For now - holds current_schema
  c_ddlqry  text         -- The DDL query associated with the current DDL event
);
                           
                        

运行以下命令以创建函数 awsdms_intercept_ddl，并将以下代码中的 objects_schema 替换为要使用的架构名称。

CREATE OR REPLACE FUNCTION objects_schema.awsdms_intercept_ddl()
  RETURNS event_trigger
LANGUAGE plpgsql
SECURITY DEFINER
  AS $$
  declare _qry text;
BEGIN
  if (tg_tag='CREATE TABLE' or tg_tag='ALTER TABLE' or tg_tag='DROP TABLE' or tg_tag = 'CREATE TABLE AS') then
         SELECT current_query() into _qry;
         insert into objects_schema.awsdms_ddl_audit
         values
         (
         default,current_timestamp,current_user,cast(TXID_CURRENT()as varchar(16)),tg_tag,0,'',current_schema,_qry
         );
         delete from objects_schema.awsdms_ddl_audit;
end if;
END;
$$;
                        

注销 OtherThanMaster 账户并使用分配了 rds_superuser 角色的账户登录。

运行以下命令以创建事件触发器 awsdms_intercept_ddl。

CREATE EVENT TRIGGER awsdms_intercept_ddl ON ddl_command_end 
EXECUTE PROCEDURE objects_schema.awsdms_intercept_ddl();
                        

确保访问这些事件的所有用户和角色都具有必要的 DDL 权限。例如：

grant all on public.awsdms_ddl_audit to public;
grant all on public.awsdms_ddl_audit_c_key_seq to public;

在你的源 PostgreSQL 数据库上创建一个 pglogical 扩展：

1. 设置正确的参数：

   • 对于自建的 PostgreSQL 数据库，请设置数据库参数。shared_preload_libraries= 'pglogical'

   • 对于 Amazon RDS 上的 PostgreSQL 和兼容 Amazon Aurora PostgreSQL 的版本数据库，请在同一 RDS 参数组中将shared_preload_libraries参数pglogical设置为。

2. 重启你的 PostgreSQL 源数据库。

3. 在 PostgreSQL 数据库上，运行以下命令，create extension pglogical;

运行以下命令以验证 pglogical 是否已成功安装：

select * FROM pg_catalog.pg_extension

确定您要用作开始点的更早复制任务（父任务）所使用的逻辑复制槽。然后，查询源数据库上的 pg_replication_slots 视图，以确保此槽没有任何活动的连接。如果是，请先解决并关闭它们，然后再继续。

对于以下步骤，假定您的逻辑复制槽为 abc1d2efghijk_34567890_z0yx98w7_6v54_32ut_1srq_1a2b34c5d67ef。

创建一个新的源终端节点，其中包含以下额外连接属性设置。

slotName=abc1d2efghijk_34567890_z0yx98w7_6v54_32ut_1srq_1a2b34c5d67ef;

使用控制台Amazon CLI或 Amazon DMS API 创建仅限 CDC 的新任务。例如，使用 CLI 可以运行以下create-replication-task命令。

aws dms create-replication-task --replication-task-identifier postgresql-slot-name-test 
--source-endpoint-arn arn:aws:dms:us-west-2:012345678901:endpoint:ABCD1EFGHIJK2LMNOPQRST3UV4 
--target-endpoint-arn arn:aws:dms:us-west-2:012345678901:endpoint:ZYX9WVUTSRQONM8LKJIHGF7ED6 
--replication-instance-arn arn:aws:dms:us-west-2:012345678901:rep:AAAAAAAAAAA5BB4CCC3DDDD2EE 
--migration-type cdc --table-mappings "file://mappings.json" --cdc-start-position "4AF/B00000D0" 
--replication-task-settings "file://task-pg.json"

要在使用控制Amazon DMS台创建仅限 CDC 的新任务时启用自定义 CDC 启动模式，请执行以下操作：

当此 CDC 任务运行时，如果指定的逻辑复制槽不存在，则会Amazon DMS引发错误。如果未使用有效的设置创建任务，则还会引发错误cdc-start-position。

创建复制槽，如下所示：

SELECT * FROM pg_create_logical_replication_slot('replication_slot_name', 'pglogical');
                    

数据库创建复制槽后，获取并记下该插槽的 restart_lsn 和 conf irmed _flush_lsn 值：

select * from pg_replication_slots where slot_name like 'replication_slot_name';

请注意，在复制时段之后创建的 CDC 任务的本地 CDC 起始位置不能早于 conf irmed_flush_lsn 值。

有关 restart_lsn 和 conf ir med_flush_lsn 值的信息，请参阅 pg_replicati on_slots

创建一个 pglogical 节点。

SELECT pglogical.create_node(node_name := 'node_name', dsn := 'your_dsn_name');
                    

使用pglogical.create_replication_set函数创建两个复制集。第一个复制集跟踪具有主键的表的更新和删除。第二个复制集仅跟踪插入，其名称与第一个复制集相同，并添加了前缀 “i”。

SELECT pglogical.create_replication_set('replication_slot_name', false, true, true, false);
SELECT pglogical.create_replication_set('ireplication_slot_name', true, false, false, true);

向复制集添加表。

SELECT pglogical.replication_set_add_table('replication_slot_name', 'schemaname.tablename', true);
SELECT pglogical.replication_set_add_table('ireplication_slot_name', 'schemaname.tablename', true);
                    

创建源端点时，请设置以下额外连接属性 (ECA)。

PluginName=PGLOGICAL;slotName=slot_name;