捕获数据更改，以便从 SQL Server 进行持续复制 - Amazon 数据库迁移服务
捕获本地或 EC2 上的自管理 SQL Server 的更改在云 SQL Server 数据库实例上设置持续复制
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

捕获数据更改，以便从 SQL Server 进行持续复制

本主题介绍如何在 SQL Server 源上设置 CDC 复制。

捕获本地或 Amazon EC2 上的自管理 SQL Server 的数据更改

要从源 Microsoft SQL Server 数据库中捕获更改，请确保该数据库已配置为进行完整备份。将数据库配置为完全恢复模式或批量日志记录模式。

对于自管理 SQL Server 源，Amazon DMS 使用以下内容：

MS-Replication

用于捕获带主键的表的更改。您可以通过在源 SQL Server 实例上向 Amazon DMS 端点用户提供 sysadmin 权限来自动配置此项。或者，也可以按照本部分中提供的步骤操作，准备源并使用不具备 Amazon DMS 端点的 sysadmin 权限的用户。

MS-CDC

用于捕获不带主键的表的更改。MS-CDC 必须在数据库级别上启用，并且为各个表分别启用。

设置 SQL Server 数据库进行持续复制（CDC）时，可以执行下列操作之一：

  • 使用 sysadmin 角色设置持续复制。

  • 将持续复制设置为不使用 sysadmin 角色。

在自管理 SQL Server 上设置持续复制

本部分包含有关使用或不使用 sysadmin 角色在自管理 SQL Server 上设置持续复制的信息。

在自管理 SQL Server 上设置持续复制：使用 sysadmin 角色

适用于 SQL Server 的 Amazon DMS 持续复制对带主键的表使用本机 SQL Server 复制，对无主键的表使用更改数据捕获（CDC）。

在设置持续复制之前，请参阅在 SQL Server 源中使用持续复制（CDC）的先决条件。

对于带主键的表，Amazon DMS 通常可以在源上配置所需的构件。但对于自管理 SQL Server 源数据库实例，必须先手动配置 SQL Server 分发。然后，具有 sysadmin 权限的 Amazon DMS 源用户可以自动为带主键的表创建发布。

要检查是否已配置分发，请运行以下命令。

sp_get_distributor

如果列分发的结果是 NULL，则未配置分发。请使用以下过程设置分发。

设置分发

  1. 使用 SQL Server Management Studio（SSMS）工具连接到 SQL Server 源数据库。

  2. 打开复制文件夹的上下文（右键单击）菜单，然后选择配置分发。此时将显示“配置分发向导”。

  3. 按照向导输入默认值并创建分发。

设置 CDC

如果您没有使用只读副本，Amazon DMS 3.4.7 及更高版本可以自动为您的数据库和所有表设置 MS CDC。要使用此功能，请将 SetUpMsCdcForTables ECA 设置为 true。有关 ECA 的信息，请参阅端点设置。

对于 Amazon DMS 3.4.7 之前的版本或将只读副本作为源的情况，请执行以下步骤：

  1. 对于无主键的表，请为数据库设置 MS-CDC。要执行此操作，请使用分配了 sysadmin 角色的账户，然后运行以下命令。

    use [DBname]
EXEC sys.sp_cdc_enable_db

  2. 接下来，为每个源表设置 MS-CDC。对于具有唯一键但没有主键的每个表，运行以下查询来设置 MS-CDC。

    exec sys.sp_cdc_enable_table
@source_schema = N'schema_name',
@source_name = N'table_name',
@index_name = N'unique_index_name',
@role_name = NULL,
@supports_net_changes = 1
GO

  3. 对于没有主键或唯一键的每个表，运行以下查询来设置 MS-CDC。

    exec sys.sp_cdc_enable_table
@source_schema = N'schema_name',
@source_name = N'table_name',
@role_name = NULL
GO

有关为特定表设置 MS-CDC 的更多信息，请参阅 SQL Server 文档。

在独立 SQL Server 上设置持续复制：不使用 sysadmin 角色

本节介绍如何针对 SQL Server 数据库源设置不需要用户账户具有 sysadmin 权限的持续复制。

注意

运行本节中的步骤后，非 sysadmin DMS 用户将有权执行以下操作：

  • 从联机事务日志文件中读取更改

  • 通过访问磁盘从事务日志备份文件中读取更改

  • 添加或更改 DMS 使用的发布内容

  • 向发布内容中添加文章

  1. 按照捕获数据更改，以便从 SQL Server 进行持续复制中所述设置用于复制的 Microsoft SQL Server。

  2. 在源数据库上启用 MS-REPLICATION。这可以手动完成，也可以通过 sysadmin 用户身份运行一次任务来完成。

  3. 使用以下脚本，在源数据库上创建 awsdms 架构：

    use master
go
create schema awsdms
go


-- Create the table valued function [awsdms].[split_partition_list] on the Master database, as follows:
USE [master]
GO

set ansi_nulls on
go

set quoted_identifier on
go

if (object_id('[awsdms].[split_partition_list]','TF')) is not null

drop function [awsdms].[split_partition_list];

go

create function [awsdms].[split_partition_list]

(

@plist varchar(8000), —A delimited list of partitions

@dlm nvarchar(1) —Delimiting character

)

returns @partitionsTable table —Table holding the BIGINT values of the string fragments

(

pid bigint primary key

)   

as

begin

declare @partition_id bigint;

declare @dlm_pos integer;

declare @dlm_len integer;

set @dlm_len = len(@dlm);

while (charindex(@dlm,@plist)>0)

begin

set @dlm_pos = charindex(@dlm,@plist);

set @partition_id = cast( ltrim(rtrim(substring(@plist,1,@dlm_pos-1))) as bigint);

insert into @partitionsTable (pid) values (@partition_id)

set @plist = substring(@plist,@dlm_pos+@dlm_len,len(@plist));

end

set @partition_id = cast (ltrim(rtrim(@plist)) as bigint);

insert into @partitionsTable (pid) values ( @partition_id );

return

end

GO

  4. 使用以下脚本，在 Master 数据库上创建 [awsdms].[rtm_dump_dblog] 过程：

    use [MASTER]

go

if (object_id('[awsdms].[rtm_dump_dblog]','P')) is not null drop procedure [awsdms].[rtm_dump_dblog];
go


set ansi_nulls on
go

set quoted_identifier on
GO



CREATE procedure [awsdms].[rtm_dump_dblog]

(

@start_lsn varchar(32),

@seqno integer,

@filename varchar(260),

@partition_list varchar(8000), — A comma delimited list: P1,P2,... Pn

@programmed_filtering integer,

@minPartition bigint,

@maxPartition bigint

)

as begin

declare @start_lsn_cmp varchar(32); — Stands against the GT comparator

SET NOCOUNT ON — – Disable "rows affected display"

set @start_lsn_cmp = @start_lsn;

if (@start_lsn_cmp) is null

set @start_lsn_cmp = '00000000:00000000:0000';

if (@partition_list is null)

begin

RAISERROR ('Null partition list waspassed',16,1);

return

end

if (@start_lsn) is not null

set @start_lsn = '0x'+@start_lsn;

if (@programmed_filtering=0)



SELECT

[Current LSN],

[operation],

[Context],

[Transaction ID],

[Transaction Name],

[Begin Time],

[End Time],

[Flag Bits],

[PartitionID],

[Page ID],

[Slot ID],

[RowLog Contents 0],

[Log Record],

[RowLog Contents 1]

FROM

fn_dump_dblog (

@start_lsn, NULL, N'DISK', @seqno, @filename,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default)

where [Current LSN] collate SQL_Latin1_General_CP1_CI_AS > @start_lsn_cmp collate SQL_Latin1_General_CP1_CI_AS

and

(

( [operation] in ('LOP_BEGIN_XACT','LOP_COMMIT_XACT','LOP_ABORT_XACT') )

or

( [operation] in ('LOP_INSERT_ROWS','LOP_DELETE_ROWS','LOP_MODIFY_ROW')

and

( ( [context] in ('LCX_HEAP','LCX_CLUSTERED','LCX_MARK_AS_GHOST') ) or ([context] = 'LCX_TEXT_MIX' and (datalength([RowLog Contents 0]) in (0,1))))

and [PartitionID] in ( select * from master.awsdms.split_partition_list (@partition_list,','))

)

or

([operation] = 'LOP_HOBT_DDL')

)


else


SELECT

[Current LSN],

[operation],

[Context],

[Transaction ID],

[Transaction Name],

[Begin Time],

[End Time],

[Flag Bits],

[PartitionID],

[Page ID],

[Slot ID],

[RowLog Contents 0],

[Log Record],

[RowLog Contents 1] — After Image

FROM

fn_dump_dblog (

@start_lsn, NULL, N'DISK', @seqno, @filename,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default,

default, default, default, default, default, default, default)

where [Current LSN] collate SQL_Latin1_General_CP1_CI_AS > @start_lsn_cmp collate SQL_Latin1_General_CP1_CI_AS

and

(

( [operation] in ('LOP_BEGIN_XACT','LOP_COMMIT_XACT','LOP_ABORT_XACT') )

or

( [operation] in ('LOP_INSERT_ROWS','LOP_DELETE_ROWS','LOP_MODIFY_ROW')

and

( ( [context] in ('LCX_HEAP','LCX_CLUSTERED','LCX_MARK_AS_GHOST') ) or ([context] = 'LCX_TEXT_MIX' and (datalength([RowLog Contents 0]) in (0,1))))

and ([PartitionID] is not null) and ([PartitionID] >= @minPartition and [PartitionID]<=@maxPartition)

)

or

([operation] = 'LOP_HOBT_DDL')

)



SET NOCOUNT OFF — Re-enable "rows affected display"

end

GO

  5. 使用以下脚本，在 Master 数据库上创建证书：

    Use [master]
Go

CREATE CERTIFICATE [awsdms_rtm_dump_dblog_cert] ENCRYPTION BY PASSWORD = N'@5trongpassword'

WITH SUBJECT = N'Certificate for FN_DUMP_DBLOG Permissions';

  6. 使用以下脚本，从证书创建登录名：

    Use [master]
Go

CREATE LOGIN awsdms_rtm_dump_dblog_login FROM CERTIFICATE [awsdms_rtm_dump_dblog_cert];

  7. 使用以下脚本，将登录名添加到 sysadmin 服务器角色：

    ALTER SERVER ROLE [sysadmin] ADD MEMBER [awsdms_rtm_dump_dblog_login];

  8. 通过以下脚本，使用证书将签名添加到 [master].[awsdms].[rtm_dump_dblog]。

    Use [master]
GO
ADD SIGNATURE
TO [master].[awsdms].[rtm_dump_dblog] BY CERTIFICATE [awsdms_rtm_dump_dblog_cert] WITH PASSWORD = '@5trongpassword';
    注意

    如果重新创建存储过程，您需要再次添加签名。

  9. 使用以下脚本在 Master 数据库上创建 [awsdms].[rtm_position_1st_timestamp]：

    use [master]
    if object_id('[awsdms].[rtm_position_1st_timestamp]','P') is not null
    DROP PROCEDURE [awsdms].[rtm_position_1st_timestamp];
    go
    create procedure [awsdms].[rtm_position_1st_timestamp]
    (
    @dbname                sysname,      -- Database name
    @seqno                 integer,      -- Backup set sequence/position number within file
    @filename              varchar(260), -- The backup filename
    @1stTimeStamp          varchar(40)   -- The timestamp to position by
    ) 
    as begin

    SET NOCOUNT ON       -- Disable "rows affected display"

    declare @firstMatching table
    (
    cLsn varchar(32),
    bTim datetime
    )

    declare @sql nvarchar(4000)
    declare @nl                       char(2)
    declare @tb                       char(2)
    declare @fnameVar                 nvarchar(254) = 'NULL'

    set @nl  = char(10); -- New line
    set @tb  = char(9)   -- Tab separator

    if (@filename is not null)
    set @fnameVar = ''''+@filename +''''

    set @sql='use ['+@dbname+'];'+@nl+
    'select top 1 [Current LSN],[Begin Time]'+@nl+
    'FROM fn_dump_dblog (NULL, NULL, NULL, '+ cast(@seqno as varchar(10))+','+ @fnameVar+','+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default)'+@nl+
    'where operation=''LOP_BEGIN_XACT''' +@nl+
    'and [Begin Time]>= cast('+''''+@1stTimeStamp+''''+' as datetime)'+@nl

    --print @sql
    delete from  @firstMatching 
    insert into @firstMatching  exec sp_executesql @sql    -- Get them all

    select top 1 cLsn as [matching LSN],convert(varchar,bTim,121) as [matching Timestamp] from @firstMatching;

    SET NOCOUNT OFF      -- Re-enable "rows affected display"

    end
    GO

  10. 使用以下脚本，在 Master 数据库上创建证书：

    Use [master]
Go
CREATE CERTIFICATE [awsdms_rtm_position_1st_timestamp_cert]
ENCRYPTION BY PASSWORD = '@5trongpassword'
WITH SUBJECT = N'Certificate for FN_POSITION_1st_TIMESTAMP Permissions';

  11. 使用以下脚本，从证书创建登录名：

    Use [master]
Go
CREATE LOGIN awsdms_rtm_position_1st_timestamp_login FROM CERTIFICATE [awsdms_rtm_position_1st_timestamp_cert];

  12. 使用以下脚本，将登录名添加到 sysadmin 角色：

    ALTER SERVER ROLE [sysadmin] ADD MEMBER [awsdms_rtm_position_1st_timestamp_login];

  13. 通过以下脚本，使用证书将签名添加到 [master].[awsdms].[rtm_position_1st_timestamp]：

    Use [master]
    GO
    ADD SIGNATURE
    TO [master].[awsdms].[rtm_position_1st_timestamp]
    BY CERTIFICATE [awsdms_rtm_position_1st_timestamp_cert]
    WITH PASSWORD = '@5trongpassword';

  14. 使用以下脚本向 DMS 用户授予对新存储过程的执行访问权限：

    use master
go
GRANT execute on [awsdms].[rtm_position_1st_timestamp] to dms_user;

  15. 在以下每个数据库中，创建具有以下权限和角色的用户：

    注意

    您应该在每个副本上使用相同的 SID 创建 dmsnosysadmin 用户账户。以下 SQL 查询可以帮助验证每个副本上的 dmsnosysadmin 账户 SID 值。有关创建用户的更多信息，请参阅 Microsoft SQL Server 文档中的 CREATE USER（Transact-SQL）。有关为 Azure SQL 数据库创建 SQL 用户账户的更多信息，请参阅活动异地复制。

    use master
go
grant select on sys.fn_dblog to [DMS_user]
grant view any definition to [DMS_user]
grant view server state to [DMS_user]—(should be granted to the login).
grant execute on sp_repldone to [DMS_user]
grant execute on sp_replincrementlsn to [DMS_user]
grant execute on sp_addpublication to [DMS_user]
grant execute on sp_addarticle to [DMS_user]
grant execute on sp_articlefilter to [DMS_user]
grant select on [awsdms].[split_partition_list] to [DMS_user]
grant execute on [awsdms].[rtm_dump_dblog] to [DMS_user]
    use MSDB
go
grant select on msdb.dbo.backupset to [DMS_user]
grant select on msdb.dbo.backupmediafamily to [DMS_user]
grant select on msdb.dbo.backupfile to [DMS_user]

    在源数据库上运行以下脚本：

    EXEC sp_addrolemember N'db_owner', N'DMS_user'
use Source_DB
go

  16. 最后，向源 SQL Server 端点添加额外连接属性（ECA）：

    enableNonSysadminWrapper=true;

在可用性组环境中的 SQL Server 上设置持续复制：无 sysadmin 角色

本节介绍如何在不需要用户账户具有 sysadmin 权限的可用性组环境中为 SQL Server 数据库源设置持续复制。

注意

运行本节中的步骤后，非 sysadmin DMS 用户将有权执行以下操作：

  • 从联机事务日志文件中读取更改

  • 通过访问磁盘从事务日志备份文件中读取更改

  • 添加或更改 DMS 使用的发布内容

  • 向发布内容中添加文章

在可用性组环境中设置持续复制而不使用 sysadmin 用户

  1. 按照捕获数据更改，以便从 SQL Server 进行持续复制中所述设置用于复制的 Microsoft SQL Server。

  2. 在源数据库上启用 MS-REPLICATION。这可以手动完成，也可以使用 sysadmin 用户身份运行一次任务来完成。

    注意

    您应该将 MS-REPLICATION 分发服务器配置为本地分发服务器，或者配置为允许非 sysadmin 用户通过关联的链接服务器进行访问。

  3. 如果启用在单个任务中独占使用 sp_repldone 端点选项，请停止 MS-REPLICATION 日志读取器作业。

  4. 在每个副本上，执行以下步骤：

    1. 在 Master 数据库中创建 [awsdms][awsdms] 架构：

      CREATE SCHEMA [awsdms]

    2. 在 Master 数据库中创建 [awsdms].[split_partition_list] 表值函数：

      USE [master]
GO

SET ansi_nulls on
GO
  
SET quoted_identifier on
GO

IF (object_id('[awsdms].[split_partition_list]','TF')) is not null
  DROP FUNCTION [awsdms].[split_partition_list];
GO

CREATE FUNCTION [awsdms].[split_partition_list] 
( 
  @plist varchar(8000),    --A delimited list of partitions    
  @dlm nvarchar(1)    --Delimiting character
) 
RETURNS @partitionsTable table --Table holding the BIGINT values of the string fragments
(
  pid bigint primary key
) 
AS 
BEGIN
  DECLARE @partition_id bigint;
  DECLARE @dlm_pos integer;
  DECLARE @dlm_len integer;  
  SET @dlm_len = len(@dlm);
  WHILE (charindex(@dlm,@plist)>0)
  BEGIN 
    SET @dlm_pos = charindex(@dlm,@plist);
    SET @partition_id = cast( ltrim(rtrim(substring(@plist,1,@dlm_pos-1))) as bigint);
    INSERT into @partitionsTable (pid) values (@partition_id)
    SET @plist = substring(@plist,@dlm_pos+@dlm_len,len(@plist));
  END 
  SET @partition_id = cast (ltrim(rtrim(@plist)) as bigint);
  INSERT into @partitionsTable (pid) values (  @partition_id  );
  RETURN
END
GO

    3. 在 Master 数据库中创建 [awsdms].[rtm_dump_dblog] 过程：

      USE [MASTER] 
GO

IF (object_id('[awsdms].[rtm_dump_dblog]','P')) is not null
  DROP PROCEDURE [awsdms].[rtm_dump_dblog]; 
GO

SET ansi_nulls on
GO 

SET quoted_identifier on 
GO
                                    
CREATE PROCEDURE [awsdms].[rtm_dump_dblog]
(
  @start_lsn            varchar(32),
  @seqno                integer,
  @filename             varchar(260),
  @partition_list       varchar(8000), -- A comma delimited list: P1,P2,... Pn
  @programmed_filtering integer,
  @minPartition         bigint,
  @maxPartition         bigint
) 
AS 
BEGIN

  DECLARE @start_lsn_cmp varchar(32); -- Stands against the GT comparator

  SET NOCOUNT ON  -- Disable "rows affected display"

  SET @start_lsn_cmp = @start_lsn;
  IF (@start_lsn_cmp) is null
    SET @start_lsn_cmp = '00000000:00000000:0000';

  IF (@partition_list is null)
    BEGIN
      RAISERROR ('Null partition list was passed',16,1);
      return
      --set @partition_list = '0,';    -- A dummy which is never matched
    END

  IF (@start_lsn) is not null
    SET @start_lsn = '0x'+@start_lsn;

  IF (@programmed_filtering=0)
    SELECT
      [Current LSN],
      [operation],
      [Context],
      [Transaction ID],
      [Transaction Name],
      [Begin Time],
      [End Time],
      [Flag Bits],
      [PartitionID],
      [Page ID],
      [Slot ID],
      [RowLog Contents 0],
      [Log Record],
      [RowLog Contents 1] -- After Image
    FROM
      fn_dump_dblog (
        @start_lsn, NULL, N'DISK', @seqno, @filename,
        default, default, default, default, default, default, default,
        default, default, default, default, default, default, default,
        default, default, default, default, default, default, default,
        default, default, default, default, default, default, default,
        default, default, default, default, default, default, default,
        default, default, default, default, default, default, default,
        default, default, default, default, default, default, default,
        default, default, default, default, default, default, default,
        default, default, default, default, default, default, default)
    WHERE 
      [Current LSN] collate SQL_Latin1_General_CP1_CI_AS > @start_lsn_cmp collate SQL_Latin1_General_CP1_CI_AS -- This aims for implementing FN_DBLOG based on GT comparator.
      AND
      (
        (  [operation] in ('LOP_BEGIN_XACT','LOP_COMMIT_XACT','LOP_ABORT_XACT') )
        OR
        (  [operation] in ('LOP_INSERT_ROWS','LOP_DELETE_ROWS','LOP_MODIFY_ROW')
          AND
          ( ( [context]   in ('LCX_HEAP','LCX_CLUSTERED','LCX_MARK_AS_GHOST') ) or ([context] = 'LCX_TEXT_MIX') )
          AND       
          [PartitionID] in ( select * from master.awsdms.split_partition_list (@partition_list,','))
        )
      OR
      ([operation] = 'LOP_HOBT_DDL')
    )
    ELSE
      SELECT
        [Current LSN],
        [operation],
        [Context],
        [Transaction ID],
        [Transaction Name],
        [Begin Time],
        [End Time],
        [Flag Bits],
        [PartitionID],
        [Page ID],
        [Slot ID],
        [RowLog Contents 0],
        [Log Record],
        [RowLog Contents 1] -- After Image
      FROM
        fn_dump_dblog (
          @start_lsn, NULL, N'DISK', @seqno, @filename,
          default, default, default, default, default, default, default,
          default, default, default, default, default, default, default,
          default, default, default, default, default, default, default,
          default, default, default, default, default, default, default,
          default, default, default, default, default, default, default,
          default, default, default, default, default, default, default,
          default, default, default, default, default, default, default,
          default, default, default, default, default, default, default,
          default, default, default, default, default, default, default)
      WHERE [Current LSN] collate SQL_Latin1_General_CP1_CI_AS > @start_lsn_cmp collate SQL_Latin1_General_CP1_CI_AS -- This aims for implementing FN_DBLOG based on GT comparator.
      AND
      (
        (  [operation] in ('LOP_BEGIN_XACT','LOP_COMMIT_XACT','LOP_ABORT_XACT') )
        OR
        (  [operation] in ('LOP_INSERT_ROWS','LOP_DELETE_ROWS','LOP_MODIFY_ROW')
          AND
          ( ( [context]   in ('LCX_HEAP','LCX_CLUSTERED','LCX_MARK_AS_GHOST') ) or ([context] = 'LCX_TEXT_MIX') )
          AND ([PartitionID] is not null) and ([PartitionID] >= @minPartition and [PartitionID]<=@maxPartition)
        )
        OR
        ([operation] = 'LOP_HOBT_DDL')
      )
      SET NOCOUNT OFF -- Re-enable "rows affected display"
END
GO

    4. 在 Master 数据库上创建证书：

      USE [master]
GO
CREATE CERTIFICATE [awsdms_rtm_dump_dblog_cert]
  ENCRYPTION BY PASSWORD = N'@hardpassword1'
  WITH SUBJECT = N'Certificate for FN_DUMP_DBLOG Permissions'

    5. 从证书创建登录名：

      USE [master]
GO
CREATE LOGIN awsdms_rtm_dump_dblog_login FROM CERTIFICATE
  [awsdms_rtm_dump_dblog_cert];

    6. 将登录名添加到 sysadmin 服务器角色：

      ALTER SERVER ROLE [sysadmin] ADD MEMBER [awsdms_rtm_dump_dblog_login];

    7. 使用证书将签名添加到 [master].[awsdms].[rtm_dump_dblog] 过程：

      USE [master]
GO

ADD SIGNATURE
  TO [master].[awsdms].[rtm_dump_dblog]
  BY CERTIFICATE [awsdms_rtm_dump_dblog_cert]
  WITH PASSWORD = '@hardpassword1';
      注意

      如果重新创建存储过程，您需要再次添加签名。

    8. 在 Master 数据库中创建 [awsdms].[rtm_position_1st_timestamp] 过程：

      USE [master]
IF object_id('[awsdms].[rtm_position_1st_timestamp]','P') is not null
  DROP PROCEDURE [awsdms].[rtm_position_1st_timestamp];
GO
CREATE PROCEDURE [awsdms].[rtm_position_1st_timestamp]
(
  @dbname                sysname,      -- Database name
  @seqno                 integer,      -- Backup set sequence/position number within file
  @filename              varchar(260), -- The backup filename
  @1stTimeStamp          varchar(40)   -- The timestamp to position by
) 
AS 
BEGIN
  SET NOCOUNT ON       -- Disable "rows affected display"

  DECLARE @firstMatching table
  (
    cLsn varchar(32),
    bTim datetime
  )
  DECLARE @sql nvarchar(4000)
  DECLARE @nl                       char(2)
  DECLARE @tb                       char(2)
  DECLARE @fnameVar                 sysname = 'NULL'

  SET @nl  = char(10); -- New line
  SET @tb  = char(9)   -- Tab separator

  IF (@filename is not null)
    SET @fnameVar = ''''+@filename +''''
  SET @filename = ''''+@filename +''''
  SET @sql='use ['+@dbname+'];'+@nl+
    'SELECT TOP 1 [Current LSN],[Begin Time]'+@nl+
    'FROM fn_dump_dblog (NULL, NULL, NULL, '+ cast(@seqno as varchar(10))+','+ @filename +','+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default,'+@nl+
    @tb+'default, default, default, default, default, default, default)'+@nl+
    'WHERE operation=''LOP_BEGIN_XACT''' +@nl+
    'AND [Begin Time]>= cast('+''''+@1stTimeStamp+''''+' as datetime)'+@nl

    --print @sql
    DELETE FROM @firstMatching 
    INSERT INTO @firstMatching  exec sp_executesql @sql    -- Get them all
    SELECT TOP 1 cLsn as [matching LSN],convert(varchar,bTim,121) AS[matching Timestamp] FROM @firstMatching;

    SET NOCOUNT OFF      -- Re-enable "rows affected display"

END
GO

    9. 在 Master 数据库上创建证书：

      USE [master]
GO
CREATE CERTIFICATE [awsdms_rtm_position_1st_timestamp_cert]
  ENCRYPTION BY PASSWORD = N'@hardpassword1'
  WITH SUBJECT = N'Certificate for FN_POSITION_1st_TIMESTAMP Permissions';

    10. 从证书创建登录名：

      USE [master]
GO
CREATE LOGIN awsdms_rtm_position_1st_timestamp_login FROM CERTIFICATE
  [awsdms_rtm_position_1st_timestamp_cert];

    11. 将登录名添加到 sysadmin 服务器角色：

      ALTER SERVER ROLE [sysadmin] ADD MEMBER [awsdms_rtm_position_1st_timestamp_login];

    12. 使用证书将签名添加到 [master].[awsdms].[rtm_position_1st_timestamp] 过程：

      USE [master]
GO
ADD SIGNATURE
  TO [master].[awsdms].[rtm_position_1st_timestamp]
  BY CERTIFICATE [awsdms_rtm_position_1st_timestamp_cert]
  WITH PASSWORD = '@hardpassword1';
      注意

      如果重新创建存储过程，您需要再次添加签名。

    13. 在以下每个数据库中，创建具有以下权限/角色的用户：

      注意

      您应该在每个副本上使用相同的 SID 创建 dmsnosysadmin 用户账户。以下 SQL 查询可以帮助验证每个副本上的 dmsnosysadmin 账户 SID 值。有关创建用户的更多信息，请参阅 Microsoft SQL Server 文档中的 CREATE USER（Transact-SQL）。有关为 Azure SQL 数据库创建 SQL 用户账户的更多信息，请参阅活动异地复制。

      SELECT @@servername servername, name, sid, create_date, modify_date
  FROM sys.server_principals
  WHERE name = 'dmsnosysadmin';

    14. 在每个副本上授予对 Master 数据库的权限：

      USE master
GO 

GRANT select on sys.fn_dblog to dmsnosysadmin;
GRANT view any definition to dmsnosysadmin;
GRANT view server state to dmsnosysadmin -- (should be granted to the login).
GRANT execute on sp_repldone to dmsnosysadmin;
GRANT execute on sp_replincrementlsn to dmsnosysadmin;
GRANT execute on sp_addpublication to dmsnosysadmin;
GRANT execute on sp_addarticle to dmsnosysadmin;
GRANT execute on sp_articlefilter to dmsnosysadmin;
GRANT select on [awsdms].[split_partition_list] to dmsnosysadmin;
GRANT execute on [awsdms].[rtm_dump_dblog] to dmsnosysadmin;
GRANT execute on [awsdms].[rtm_position_1st_timestamp] to dmsnosysadmin;

    15. 在每个副本上授予对 MSDB 数据库的权限：

      USE msdb
GO
GRANT select on msdb.dbo.backupset to dmsnosysadmin
GRANT select on msdb.dbo.backupmediafamily to dmsnosysadmin
GRANT select on msdb.dbo.backupfile to dmsnosysadmin

    16. 在源数据库上将 db_owner 角色添加到 dmsnosysadmin。由于数据库已同步，您只能在主副本上添加角色。

      use <source DB>
GO 
EXEC sp_addrolemember N'db_owner', N'dmsnosysadmin'

在云 SQL Server 数据库实例上设置持续复制

此部分介绍如何在云托管 SQL Server 数据库实例上设置 CDC。云托管的 SQL 服务器实例是在 Amazon RDS for SQL Server、Azure SQL 托管实例或任何其他托管云 SQL Server 实例上运行的实例。有关每种数据库类型的持续复制限制的信息，请参阅将 SQL Server 作为 Amazon DMS 的源的限制。

在设置持续复制之前，请参阅在 SQL Server 源中使用持续复制（CDC）的先决条件。

不同于自管理 Microsoft SQL Server 源，Amazon RDS for SQL Server 不支持 MS-Replication。因此，Amazon DMS 需要对带或不带主键的表使用 MS-CDC。

Amazon RDS 未授予设置复制构件的 sysadmin 权限，Amazon DMS 使用该权限在源 SQL Server 实例中进行持续更改。请确保在以下过程中使用主用户权限为 Amazon RDS 实例启用 MS-CDC。

在云 SQL Server 数据库实例上启用 MS-CDC

  1. 在数据库级别运行以下查询之一。

    对于 RDS for SQL Server 数据库实例，请使用此查询。

    exec msdb.dbo.rds_cdc_enable_db 'DB_name'

    对于 Azure SQL 托管数据库实例，请使用此查询。

    USE DB_name 
GO 
EXEC sys.sp_cdc_enable_db 
GO

  2. 对于带主键的每个表，运行以下查询来启用 MS-CDC。

    exec sys.sp_cdc_enable_table
@source_schema = N'schema_name',
@source_name = N'table_name',
@role_name = NULL,
@supports_net_changes = 1
GO

    对于有唯一键但没有主键的每个表，运行以下查询来启用 MS-CDC。

    exec sys.sp_cdc_enable_table
@source_schema = N'schema_name',
@source_name = N'table_name',
@index_name = N'unique_index_name',
@role_name = NULL,
@supports_net_changes = 1
GO

    对于既没有主键又没有唯一键的每个表，运行以下查询来启用 MS-CDC。

    exec sys.sp_cdc_enable_table
@source_schema = N'schema_name',
@source_name = N'table_name',
@role_name = NULL
GO

  3. 设置保留期：

    • 对于使用 DMS 版本 3.5.3 及更高版本进行复制的 RDS for SQL Server 实例，请确保将保留期设置为 5 秒（默认值）。如果要从 DMS 3.5.2 及更低版本升级或迁移到 DMS 3.5.3 及更高版本，当任务在新实例或升级后的实例上运行后，请更改轮询间隔值。以下脚本将保留期设置为 5 秒：

      use dbname
EXEC sys.sp_cdc_change_job @job_type = 'capture' ,@pollinginterval = 5
exec sp_cdc_stop_job 'capture'
exec sp_cdc_start_job 'capture'

    • 对于使用 DMS 版本 3.5.2 及更低版本进行复制的 Azure SQL MI 和 RDS for SQL Server 实例，请使用以下命令：

      use dbname
EXEC sys.sp_cdc_change_job @job_type = 'capture' ,@pollinginterval = 86399
exec sp_cdc_stop_job 'capture'
exec sp_cdc_start_job 'capture'

      参数 @pollinginterval 按秒计，建议值设置为 86399。当该参数为 @pollinginterval = 86399 时，这意味着事务日志将更改保留 86399 秒（一天）。过程 exec sp_cdc_start_job 'capture' 启动这些设置。

      注意

      对于某些 SQL Server 版本，如果 pollinginterval 的值设置为超过 3599 秒，该值将重置为默认的五秒。发生这种情况时，T-Log 条目会在 Amazon DMS 可以读取之前被清除。要确定哪些 SQL Server 版本受到此已知问题的影响，请参阅这篇 Microsoft 知识库文章。

      如果您使用带有多可用区的 Amazon RDS，请确保也设置辅助项，以便在失效转移时具有正确的值。

      exec rdsadmin..rds_set_configuration 'cdc_capture_pollinginterval' , <5 or 86399>
在 Amazon DMS 复制任务停止超过一小时时保持保留期
注意

使用 DMS 3.5.3 及更高版本复制 RDS for SQL Server 源时无需执行以下步骤。

  1. 使用以下命令停止截断事务日志的作业。

    exec sp_cdc_stop_job 'capture'

  2. 在 Amazon DMS 控制台上找到您的任务并继续执行此任务。

  3. 选择监控选项卡，然后勾选 CDCLatencySource 指标。

  4. 如果 CDCLatencySource 指标等于 0（零）并保持不变，请使用以下命令重新启动截断事务日志的作业。

    exec sp_cdc_start_job 'capture'

记得启动截断 SQL Server 事务日志的作业。否则，您的 SQL Server 实例上的存储空间可能会被填满。

使用 RDS for SQL Server 作为 Amazon DMS 源时的推荐设置

对于 Amazon DMS 3.5.3 及更高版本

注意

对于在 DMS 版本 3.5.3 发布之后创建或修改的端点，默认启用 RDS for SQL Server 日志备份特征的初始版本。要将此特征用于现有端点，请在不进行任何更改的情况下修改端点。

Amazon DMS 版本 3.5.3 支持从日志备份中读取数据。DMS 主要依靠从活动事务日志中读取数据来复制事件。如果在 DMS 可以从活动日志中读取事务之前对事务进行了备份，则该任务将按需访问 RDS 备份并从后续备份日志中读取数据，直到赶上活动事务日志的步伐。为确保 DMS 可以访问日志备份，请将 RDS 自动备份保留期设置为至少一天。有关设置自动备份保留期的信息，请参阅《Amazon RDS 用户指南》中的备份保留期。

访问日志备份的 DMS 任务会使用 RDS 实例上的存储空间。请注意，该任务仅访问复制所需的日志备份。Amazon RDS 会在几小时内删除这些下载的备份。此删除不会影响保留在 Amazon S3 中的 Amazon RDS 备份，也不会影响 Amazon RDS RESTORE DATABASE 功能。如果您打算使用 DMS 进行复制，建议在 RDS for SQL Server 源上分配额外的存储空间。估算所需存储量的一种方法是确定 DMS 将从哪个备份开始或恢复复制，然后使用 RDS tlog backup 元数据函数将所有后续备份的文件大小相加。有关 tlog backup 函数的更多信息，请参阅《Amazon RDS 用户指南》中的列出可用的事务日志备份。

或者，您可以根据 Amazon RDS 实例的 CloudWatch FreeStorageSpace 指标，选择启用存储自动扩展和/或触发存储扩展。

强烈建议您不要从事务日志备份中太远的位置开始或恢复，因为这可能会导致 SQL Server 实例上的存储空间被填满。在这种情况下，建议启动完全加载。从事务日志备份中复制比从活动事务日志中读取要慢。有关更多信息，请参阅 针对 RDS for SQL Server 的事务日志备份处理。

请注意，访问日志备份需要额外的权限。有关更多信息，请参阅从云 SQL Server 数据库进行持续复制所需的设置权限 中的详细说明。确保在任务开始复制之前授予这些权限。

对于 Amazon DMS 3.5.2 及更低版本

当您使用 Amazon RDS for SQL Server 作为源时，MS-CDC 捕获作业依赖于参数 maxscans 和 maxtrans。这些参数控制 MS-CDC 捕获对事务日志执行的最大扫描次数以及每次扫描处理的事务数。

对于事务数大于 maxtrans*maxscans 的数据库，增加其 polling_interval 值可能会导致活动事务日志记录的累积。这种累积反过来又会导致事务日志的大小增加。

请注意，Amazon DMS 并不依赖于 MS-CDC 捕获作业。MS-CDC 捕获作业将事务日志条目标记为已处理。这样可允许事务日志备份作业从事务日志中删除条目。

我们建议您监控事务日志的大小和 MS-CDC 作业是否成功。如果 MS-CDC 作业失败，事务日志可能会过度增长，并导致 Amazon DMS 复制失败。您可以使用源数据库中的 sys.dm_cdc_errors 动态管理视图来监控 MS-CDC 捕获作业错误。您可以使用 DBCC SQLPERF(LOGSPACE) 管理命令来监控事务日志的大小。

解决由 MS-CDC 引起的事务日志增加问题

  1. 检查 Amazon DMS 正在从中复制的数据库的 Log Space Used %，验证它是否在持续增加。

    DBCC SQLPERF(LOGSPACE)

  2. 确定阻碍事务日志备份过程的原因。

    Select log_reuse_wait, log_reuse_wait_desc, name from sys.databases where name = db_name();

    如果 log_reuse_wait_desc 值等于 REPLICATION，则说明日志备份保留是 MS-CDC 中的延迟所导致。

  3. 通过增加 maxtrans 和 maxscans 参数值来增加捕获作业处理的事件数。

    EXEC sys.sp_cdc_change_job @job_type = 'capture' ,@maxtrans = 5000, @maxscans = 20 
exec sp_cdc_stop_job 'capture'
exec sp_cdc_start_job 'capture'

要解决此问题，请设置 maxscans 和 maxtrans 的值，以使 maxtrans*maxscans 等于 Amazon DMS 每天为从源数据库复制的表生成的事件的平均数。

如果将这些参数设置为高于建议值，则捕获作业会处理事务日志中的所有事件。如果将这些参数设置为低于建议值，则 MS-CDC 延迟会增加，事务日志也会增加。

为 maxscans 和 maxtrans 确定合适的值可能很困难，因为工作负载的变化会产生不同数量的事件。在这种情况下，建议您对 MS-CDC 延迟设置监控。有关更多信息，请参阅 SQL Server 文档中的监控流程。然后根据监控结果动态配置 maxtrans 和 maxscans。

如果 Amazon DMS 任务找不到恢复或继续任务所需的日志序列号（LSN），该任务可能会失败，需要完全重新加载。

注意

使用 Amazon DMS 从 RDS for SQL Server 源复制数据时，在 Amazon RDS 实例停止启动事件后尝试恢复复制时，您可能会遇到错误。这是由于 SQL Server Agent 进程在停止启动事件后重新启动时，会重新启动捕获作业进程。这绕过了 MS-CDC 轮询间隔。

因此，在事务量低于 MS-CDC 捕获作业处理量的数据库上，这可能会导致数据在 Amazon DMS 可从停止位置恢复之前被处理或标记为已复制和备份，从而导致以下错误：

[SOURCE_CAPTURE ]E: Failed to access LSN '0000dbd9:0006f9ad:0003' in the backup log sets since BACKUP/LOG-s are not available. [1020465] (sqlserver_endpoint_capture.c:764)

要缓解此问题，请按照之前的建议设置 maxtrans 和 maxscans 值。