本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

# 從 SQL Server 擷取持續複寫的資料變更
<a name="CHAP_Source.SQLServer.CDC"></a>

本主題說明如何在 SQL Server 來源上設定 CDC 複寫。

**Topics**
+ [在內部部署或 Amazon EC2 上擷取自我管理 SQL Server 的資料變更](#CHAP_Source.SQLServer.CDC.Selfmanaged)
+ [設定雲端 SQL Server 資料庫執行個體的進行中複寫](#CHAP_Source.SQLServer.Configuration)

## 在內部部署或 Amazon EC2 上擷取自我管理 SQL Server 的資料變更
<a name="CHAP_Source.SQLServer.CDC.Selfmanaged"></a>

若要從來源 Microsoft SQL Server 資料庫擷取變更，請確定資料庫已設定為完整備份。在完整復原模式或大量記錄模式下設定資料庫。

對於自我管理的 SQL Server 來源， AWS DMS 會使用下列項目：

**MS-Replication**  
針對具有主索引鍵的資料表擷取變更。您可以為來源 SQL Server 執行個體上的 AWS DMS 端點使用者提供 sysadmin 權限，以自動設定此項目。或者，您可以遵循本節中的步驟來準備來源，並使用沒有 AWS DMS 端點 sysadmin 權限的使用者。

**MS-CDC**  
針對不具有主索引鍵的資料表擷取變更。必須在資料庫層級啟用 MS-CDC，而且是針對所有資料表個別啟用。

設定 SQL Server 資料庫進行中複寫 (CDC) 時，您可以執行下列其中一項：
+ 使用 sysadmin 角色設定進行中複寫。
+ 設定進行中複寫不使用 sysadmin 角色。

**注意**  
您可以使用下列指令碼來尋找沒有主索引鍵或唯一索引鍵的所有資料表：  

```
USE [DBname]
SELECT SCHEMA_NAME(schema_id) AS schema_name, name AS table_name
FROM sys.tables
WHERE OBJECTPROPERTY(object_id, 'TableHasPrimaryKey') = 0
        AND  OBJECTPROPERTY(object_id, 'TableHasUniqueCnst') = 0
ORDER BY schema_name, table_name;
```

### 在自我管理的 SQL 伺服器上設定進行中複寫
<a name="CHAP_Source.SQLServer.CDC.MSCDC"></a>

本節包含的相關資訊說明的是，在使用或不使用 sysadmin 角色的自我管理 SQL 伺服器上如何設定進行中複寫。

**Topics**
+ [在自我管理的 SQL Server 上設定進行中複寫：使用 sysadmin 角色](#CHAP_Source.SQLServer.CDC.MSCDC.Sysadmin)
+ [在獨立 SQL Server 上設定進行中複寫：沒有 sysadmin 角色](#CHAP_SupportScripts.SQLServer.standalone)
+ [在可用群組環境中的 SQL Server 上設定進行中複寫：而不使用 sysadmin 角色](#CHAP_SupportScripts.SQLServer.ag)

#### 在自我管理的 SQL Server 上設定進行中複寫：使用 sysadmin 角色
<a name="CHAP_Source.SQLServer.CDC.MSCDC.Sysadmin"></a>

AWS DMS SQL Server 的持續複寫對具有主索引鍵的資料表使用原生 SQL Server 複寫，並為沒有主索引鍵的資料表變更資料擷取 (CDC)。

在設定進行中複寫之前，請參閱[使用來自 SQL Server 來源進行中複寫 (CDC) 的先決條件](CHAP_Source.SQLServer.md#CHAP_Source.SQLServer.Prerequisites)。

對於具有主索引鍵的資料表， 通常 AWS DMS 可以在來源上設定所需的成品。不過，針對自我管理的 SQL Server 來源執行個體，必須先手動設定 SQL Server 分佈。執行此操作後，具有 sysadmin 許可的 AWS DMS 來源使用者可以自動為具有主索引鍵的資料表建立發佈。

若要檢查是否已設定分發，請執行下列命令。

```
sp_get_distributor
```

如果資料欄分佈的結果是 `NULL`，則表示尚未設定分佈。您可以使用下列程序設定分佈。<a name="CHAP_Source.SQLServer.CDC.MSCDC.Setup"></a>

**設定分佈**

1. 使用 SQL Server Management Studio (SSMS) 工具連線至 SQL Server 來源資料庫。

1. 開啟**複寫**資料夾的內容 (按一下滑鼠右鍵) 選單，然後選擇**設定分佈**。[設定分佈精靈] 隨即顯示。

1. 按照精靈的指示輸入預設值，然後建立分佈。<a name="CHAP_Source.SQLServer.CDC.MSCDC.Setup.CDC"></a>

**設定 CDC**

AWS DMS 如果您未使用唯讀複本， 3.4.7 版和更新版本可以自動為您的資料庫和所有資料表設定 MS CDC。若要使用此功能，請將 `SetUpMsCdcForTables` ECA 設定為 true。如需 ECA 的相關資訊，請參閱[端點設定](CHAP_Source.SQLServer.md#CHAP_Source.SQLServer.ConnectionAttrib)。

對於 3.4.7 AWS DMS 以前的 版本，或作為來源的唯讀複本，請執行下列步驟：

1. 對於沒有主索引鍵的資料表，請為資料庫設定 MS-CDC。若要這麼做，請使用已指派 sysadmin 角色的帳戶，並執行下列命令。

   ```
   use [DBname]
   EXEC sys.sp_cdc_enable_db
   ```

1. 接下來，為每個來源資料表設定 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
   ```

1. 針對沒有主索引鍵、也沒有唯一索引鍵的每個資料表，執行以下查詢以設定 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 文件](https://msdn.microsoft.com/en-us/library/cc627369.aspx)。

#### 在獨立 SQL Server 上設定進行中複寫：沒有 sysadmin 角色
<a name="CHAP_SupportScripts.SQLServer.standalone"></a>

此章節描述如何設定獨立 SQL Server 資料庫來源的進行中複寫，此資料庫不需要使用者帳戶擁有 sysadmin 權限。

**注意**  
執行本節中的步驟後，非系統管理員 DMS 使用者將會擁有執行下列動作的權限：  
從線上交易記錄檔讀取變更
從交易記錄備份檔讀取變更的磁碟存取權
新增或變更 DMS 使用的出版物
將文章新增至出版物

1. 如 [從 SQL Server 擷取持續複寫的資料變更](#CHAP_Source.SQLServer.CDC) 中所述，設定用於複寫的 Microsoft SQL Server。

1. 在來源資料庫上啟用 MS-REPLICATION。此操作可以手動完成，也可以透過以 sysadmin 使用者身分執行任務一次來完成。

1. 使用下列指令碼在來源資料庫上建立 `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
   ```

1. 使用下列指令碼在 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
   ```

1. 使用下列指令碼在 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';
   ```

1. 使用下列指令碼從憑證建立登入：

   ```
   Use [master]
   Go
   
   CREATE LOGIN awsdms_rtm_dump_dblog_login FROM CERTIFICATE [awsdms_rtm_dump_dblog_cert];
   ```

1. 使用下列指令碼將登入新增至 sysadmin 伺服器角色：

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

1. 透過以下指令碼，使用該憑證將簽章新增至 [master].[awsdms].[rtm\$1dump\$1dblog]：

   ```
   Use [master]
   GO
   ADD SIGNATURE
   TO [master].[awsdms].[rtm_dump_dblog] BY CERTIFICATE [awsdms_rtm_dump_dblog_cert] WITH PASSWORD = '@5trongpassword';
   ```
**注意**  
如果您重新建立預存程序，則需要再次新增簽章。

1. 使用下列指令碼，在 Master 資料庫上建立 [awsdms].[rtm\$1position\$11st\$1timestamp]：

   ```
   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
   ```

1. 使用下列指令碼在 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';
   ```

1. 使用下列指令碼從憑證建立登入：

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

1. 使用下列指令碼將登入新增至 sysadmin 角色：

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

1. 使用下列指令碼，使用該憑證將簽章新增至 [master].[awsdms].[rtm\$1position\$11st\$1timestamp]：

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

1. 使用下列指令碼，授與 DMS 使用者執行新預存程序的存取權：

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

1. 在下列每個資料庫中建立具有下列許可和角色的使用者：
**注意**  
您應該在每個複本上建立具有相同 SID 的 dmsnosysadmin 使用者帳戶。下列 SQL 查詢可協助驗證每個複本上的 dmsnosysadmin 帳戶 SID 值。如需建立使用者的相關資訊，請參閱 [Microsoft SQL 伺服器說明文件](https://learn.microsoft.com/en-us/sql/)中的 [CREATE USER (Transact-SQL)](https://learn.microsoft.com/en-us/sql/t-sql/statements/create-user-transact-sql)。如需針對 Azure SQL 資料庫建立 SQL 使用者帳戶的詳細資訊，請參閱[作用中異地複寫](https://learn.microsoft.com/en-us/azure/azure-sql/database/active-geo-replication-overview)。

   ```
   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 self_managed_user
   grant select on msdb.dbo.backupmediafamily to self_managed_user
   grant select on msdb.dbo.backupfile to self_managed_user
   ```

   在來源資料庫上執行下列指令碼：

   ```
   use Source_DB
       Go
       EXEC sp_addrolemember N'db_owner', N'DMS_user'
   ```

1. 最後，將額外連線屬性 (ECA) 新增至來源 SQL Server 端點：

   ```
   enableNonSysadminWrapper=true;
   ```

#### 在可用群組環境中的 SQL Server 上設定進行中複寫：而不使用 sysadmin 角色
<a name="CHAP_SupportScripts.SQLServer.ag"></a>

本節描述如何在可用性群組環境中設定 SQL Server 資料庫來源的持續複寫，此資料庫不要求使用者帳戶須擁有 sysadmin 權限。

**注意**  
執行本節中的步驟後，非系統管理員 DMS 使用者將會擁有執行下列動作的權限：  
從線上交易記錄檔讀取變更
從交易記錄備份檔讀取變更的磁碟存取權
新增或變更 DMS 使用的出版物
將文章新增至出版物

**在可用群組環境中不使用 sysadmin 使用者的情況下設定進行中複寫**

1. 如 [從 SQL Server 擷取持續複寫的資料變更](#CHAP_Source.SQLServer.CDC) 中所述，設定用於複寫的 Microsoft SQL Server。

1. 在來源資料庫上啟用 MS-REPLICATION。此操作可以手動完成，也可以透過使用 sysadmin 使用者執行任務一次來完成。
**注意**  
您應該將 MS-REPLICATION 分發者設定為本機，或以允許透過關聯的連結伺服器存取非 sysadmin 使用者的方式進行。

1. 如果已啟用**在單一任務內專用 sp\$1repldone** 端點選項，請停止 MS-REPLICATION 日誌讀取器作業。

1. 在每個複本上執行以下步驟：

   1. 在主資料庫中建立 `[awsdms]`[awsdms] 結構描述：

      ```
      CREATE SCHEMA [awsdms]
      ```

   1. 在 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
      ```

   1. 在 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
      ```

   1. 在 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'
      ```

   1. 從憑證建立登入：

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

   1. 將登入新增至 sysadmin 伺服器角色：

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

   1. 透過此憑證，將簽章新增至 [master].[awsdms].[rtm\$1dump\$1dblog] 程序：

      ```
      USE [master]
      GO
      
      ADD SIGNATURE
        TO [master].[awsdms].[rtm_dump_dblog]
        BY CERTIFICATE [awsdms_rtm_dump_dblog_cert]
        WITH PASSWORD = '@hardpassword1';
      ```
**注意**  
如果您重新建立預存程序，則需要再次新增簽章。

   1. 在 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
      ```

   1. 在 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';
      ```

   1. 從憑證建立登入：

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

   1. 將登入新增至 sysadmin 伺服器角色：

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

   1. 使用憑證將簽章新增至 `[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';
      ```
**注意**  
如果您重新建立預存程序，則需要再次新增簽章。

   1. 在下列每個資料庫中建立具有下列許可/角色的使用者：
**注意**  
您應該在每個複本上建立具有相同 SID 的 dmsnosysadmin 使用者帳戶。下列 SQL 查詢可協助驗證每個複本上的 dmsnosysadmin 帳戶 SID 值。如需建立使用者的相關資訊，請參閱 [Microsoft SQL 伺服器說明文件](https://learn.microsoft.com/en-us/sql/)中的 [CREATE USER (Transact-SQL)](https://learn.microsoft.com/en-us/sql/t-sql/statements/create-user-transact-sql)。如需針對 Azure SQL 資料庫建立 SQL 使用者帳戶的詳細資訊，請參閱[作用中異地複寫](https://learn.microsoft.com/en-us/azure/azure-sql/database/active-geo-replication-overview)。

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

   1. 在每個複本的主資料庫上授予許可：

      ```
      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;
      ```

   1. 在每個複本上 msdb 資料庫授予許可：

      ```
      USE msdb
      GO
      GRANT select on msdb.dbo.backupset TO self_managed_user
      GRANT select on msdb.dbo.backupmediafamily TO self_managed_user
      GRANT select on msdb.dbo.backupfile TO self_managed_user
      ```

   1. 將 `db_owner` 角色新增至 `dmsnosysadmin` 來源資料庫。由於資料庫已同步處理，因此您只能在主要複本上新增角色。

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

## 設定雲端 SQL Server 資料庫執行個體的進行中複寫
<a name="CHAP_Source.SQLServer.Configuration"></a>

本節說明如何在雲端託管的 SQL Server 資料庫執行個體上設定 CDC。雲端託管的 SQL 伺服器執行個體是在 Amazon RDS for SQL Server、Azure SQL 受管執行個體或任何其他受管雲端 SQL Server 執行個體上執行的執行個體。如需每種資料庫類型之進行中複寫限制的相關資訊，請參閱[使用 SQL Server 做為 來源的限制 AWS DMS](CHAP_Source.SQLServer.md#CHAP_Source.SQLServer.Limitations)。

在設定進行中複寫之前，請參閱[使用來自 SQL Server 來源進行中複寫 (CDC) 的先決條件](CHAP_Source.SQLServer.md#CHAP_Source.SQLServer.Prerequisites)。

與自我管理的 Microsoft SQL Server 來源不同，Amazon RDS for SQL Server 不支援 MS-Replication。因此， AWS DMS 需要對具有或沒有主索引鍵的資料表使用 MS-CDC。

Amazon RDS 不會授予 sysadmin 權限，以設定 AWS 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
   ```

1. 針對有主索引鍵的每個資料表，執行以下查詢以啟用 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
   ```

1. 設定保留期間：
   + 對於使用 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'
     ```
   + 參數 `@pollinginterval` 的測量單位為秒，建議值設定為 86399。這表示當 `@pollinginterval = 86399` 時，交易日誌會保留變更 86,399 秒 (一天)。此程序 `exec sp_cdc_start_job 'capture'` 會啟動設定。
**注意**  
對於某些版本的 SQL Server，如果將 `pollinginterval` 的值設定為超過 3599 秒，就會將值重設為預設值 (五秒)。發生這種情況時，T-Log 項目會被清除，然後 AWS DMS 才能讀取它們。若要判斷哪些 SQL Server 版本會受到這個已知問題的影響，請參閱[這篇 Microsoft 知識庫文章](https://support.microsoft.com/en-us/topic/kb4459220-fix-incorrect-results-occur-when-you-convert-pollinginterval-parameter-from-seconds-to-hours-in-sys-sp-cdc-scan-in-sql-server-dac8aefe-b60b-7745-f987-582dda2cfa78)。

     如果您將 Amazon RDS 與異地同步備份搭配使用，請確保您也將次要設定為具有正確的值，以防容錯移轉。

     ```
     exec rdsadmin..rds_set_configuration 'cdc_capture_pollinginterval' , <5 or preferred value>
     ```

**維護複 AWS DMS 寫任務停止超過一小時的保留期**
**注意**  
使用 DMS 3.5.3 及更高版本複寫 RDS for SQL Server 來源時，不需要下列步驟。

1. 使用以下命令，停止截斷交易日誌的任務。

   ```
   exec sp_cdc_stop_job 'capture'
   ```

1. 在 AWS DMS 主控台上尋找您的任務，並繼續任務。

1. 選擇**監控**索引標籤，接著勾選 `CDCLatencySource` 指標。

1. 一旦 `CDCLatencySource` 指標等於 0 (零) 並停在該階段，請使用下列命令重新開始截斷交易日誌的作業。

   ```
   exec sp_cdc_start_job 'capture'
   ```

請記得開始截斷 SQL Server 交易日誌的任務。否則，SQL Server 執行個體上的儲存空間可能會填滿。

### 使用 RDS for SQL Server 做為 來源時的建議設定 AWS DMS
<a name="CHAP_Source.SQLServer.Configuration.Settings"></a>

#### For AWS DMS 3.5.3 及更高版本
<a name="CHAP_Source.SQLServer.Configuration.Settings.353"></a>

**注意**  
對於您在發行 DMS 3.5.3 版之後建立或修改的端點，預設會啟用 RDS for SQL Server 日誌備份功能的初始版本。若要對現有端點使用此功能，請修改端點而不進行任何變更。

AWS DMS 3.5.3 版推出支援從日誌備份讀取。DMS 主要依賴從作用中交易日誌讀取來複寫事件。如果交易在 DMS 可以從作用中日誌中讀取之前進行備份，任務會隨需存取 RDS 備份，並從後續備份日誌中讀取，直到趕上作用中交易日誌為止。為了確保 DMS 可以存取日誌備份，請將 RDS 自動化備份保留期設定為至少一天。如需有關設定自動備份保留期的資訊，請參閱《*Amazon RDS 使用者指南*》中的[備份保留期](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_ManagingAutomatedBackups.html#USER_WorkingWithAutomatedBackups.BackupRetention)。

存取日誌備份的 DMS 任務會使用 RDS 執行個體上的儲存體。請注意，任務只會存取複寫所需的日誌備份。Amazon RDS 會在幾個小時內移除這些下載的備份。此移除不會影響保留在 Amazon S3 或 Amazon RDS `RESTORE DATABASE`功能的 Amazon RDS 備份。如果您想要使用 DMS 複寫，建議您在 RDS for SQL Server 來源上配置額外的儲存空間。估算所需儲存量的一種方法是識別 DMS 將從中開始或繼續複寫的備份，並使用 RDS `tlog backup`中繼資料函數新增所有後續備份的檔案大小。如需 `tlog backup`函數的詳細資訊，請參閱《*Amazon RDS 使用者指南*》中的[列出可用的交易日誌備份](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER.SQLServer.AddlFeat.TransactionLogAccess.html#USER.SQLServer.AddlFeat.TransactionLogAccess.Listing)。

或者，您可以選擇根據 Amazon RDS 執行個體的 CloudWatch `FreeStorageSpace` 指標啟用儲存體自動擴展和/或觸發儲存體擴展。

強烈建議您不要在交易日誌備份中太早開始或繼續，因為這可能會導致 SQL Server 執行個體上的儲存體填滿。在這種情況下，建議啟動完全載入。從交易日誌備份複寫比從作用中交易日誌讀取還慢。如需詳細資訊，請參閱[RDS for SQL Server 的交易日誌備份處理](CHAP_Troubleshooting_Latency_Source_SQLServer.md#CHAP_Troubleshooting_Latency_Source_SQLServer_backup)。

請注意，存取日誌備份需要額外權限。如需詳細資訊，請參閱 中的詳細說明[從雲端 SQL Server 資料庫設定持續複寫的許可](CHAP_Source.SQLServer.md#CHAP_Source.SQLServer.Permissions.Cloud)。在任務開始複寫之前，請確定您授予這些權限。

#### For AWS DMS 3.5.2 及更新版本
<a name="CHAP_Source.SQLServer.Configuration.Settings.352"></a>

當您使用 Amazon RDS for SQL Server 做為來源時，MS-CDC 擷取任務依賴參數 `maxscans`和 `maxtrans`。這些參數會管理 MS-CDC 擷取對交易日誌執行的掃描數量上限，以及每次掃描處理的交易數量。

對於交易數目大於 `maxtrans*maxscans` 的資料庫，提高 `polling_interval` 值可能會導致作用中交易日誌記錄不斷累積。反過來，此累積會導致交易日誌變大。

請注意， AWS DMS 不依賴 MS-CDC 擷取任務。MS-CDC 擷取作業會將交易日誌項目標記為已處理。這可讓交易日誌備份作業從交易日誌中移除項目。

我們建議您監控交易日誌的大小，以及 MS-CDC 作業是否成功。如果 MS-CDC 任務失敗，交易日誌可能會過度增長並導致 AWS DMS 複寫失敗。您可以使用來源資料庫中的 `sys.dm_cdc_errors` 動態管理檢視，來監控 MS-CDC 擷取作業錯誤。您可以使用 `DBCC SQLPERF(LOGSPACE)` 管理命令來監控交易日誌大小。

**若要解決 MS-CDC 造成的交易日誌增加**

1. 檢查資料庫`Log Space Used %`的 AWS DMS 是否正在複寫，並驗證它是否持續增加。

   ```
   DBCC SQLPERF(LOGSPACE)
   ```

1. 識別封鎖交易日誌備份程序的項目。

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

   如果 `log_reuse_wait_desc` 值等於 `REPLICATION`，則日誌備份保留是由 MS-CDC 中的延時所造成。

1. 增加 `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` 等於每天從來源資料庫 AWS DMS 複寫之資料表所產生的平均事件數。

如果您將這些參數設定為高於建議值，擷取作業會處理交易日誌中的所有事件。如果您將這些參數設定為低於建議值，MS-CDC 延時會增加，而且交易日誌也會增加。

由於工作負載中的變更會產生不同數量的事件，因此找出 `maxscans` 和 `maxtrans` 適當的值可能很困難。在此情況下，建議您設定 MS-CDC 延時的監控。如需詳細資訊，請參閱 SQL Server 文件中的[監控程序](https://docs.microsoft.com/en-us/sql/relational-databases/track-changes/administer-and-monitor-change-data-capture-sql-server?view=sql-server-ver15#Monitor)。然後根據監控結果動態設定 `maxtrans` 和 `maxscans`。

如果 AWS DMS 任務找不到繼續或繼續任務所需的日誌序號 (LSNs)，則任務可能會失敗，並需要完全重新載入。

**注意**  
使用 從 RDS for SQL Server 來源 AWS DMS 複寫資料時，您可能會在 Amazon RDS 執行個體的停止啟動事件後嘗試繼續複寫時發生錯誤。這是因為 SQL Server 代理程式程序會在停止啟動事件之後重新啟動時，重新啟動擷取作業程序。這會略過 MS-CDC 輪詢間隔。  
因此，在交易磁碟區低於 MS-CDC 擷取任務處理的資料庫上，這可能會導致資料處理或標記為複寫和備份，然後 AWS 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` 值。