为 CDC 配置 Cloud SQL for SQL Server 数据库

本页面介绍如何配置变更数据捕获 (CDC)，以 将数据从 Cloud SQL for SQL Server 数据库流式传输到受支持的目标， 例如 BigQuery 或 Cloud Storage。

1. 连接到 Cloud SQL 实例。您可以在 Cloud Shell 提示符中使用 gcloud sql connect 命令来执行此操作。

2. 通过运行以下命令在数据库上启用 CDC：

   EXEC msdb.dbo.gcloudsql_cdc_enable_db 'DATABASE_NAME'
   

   将 DATABASE_NAME 替换为源数据库的名称。

3. 在需要捕获更改的表上启用 CDC：

   USE [DATABASE_NAME]
   EXEC sys.sp_cdc_enable_table
   @source_schema = N'SCHEMA_NAME',
   @source_name = N'TABLE_NAME',
   @role_name = NULL
   GO
   

4. 启用快照隔离。

   从 SQL Server 数据库回填数据时，务必确保快照的一致性。如果您不应用本部分中所述的设置，则在回填过程中对数据库所做的更改可能会导致重复或不正确的结果。如果您的数据流包含没有主键的表，则必须应用快照隔离设置。

   启用快照隔离会在回填过程开始时创建数据库的临时视图。这样可确保复制的数据保持一致，即使其他用户同时对实时表进行更改也是如此。 启用快照隔离可能会对性能产生轻微影响，但对于可靠的数据提取至关重要。

   如需启用快照隔离，请执行以下操作：

   1. 使用 SQL Server 客户端连接到您的数据库。
   2. 运行以下命令：

   ALTER DATABASE DATABASE_NAME SET ALLOW_SNAPSHOT_ISOLATION ON;
   

   将 DATABASE_NAME 替换为数据库的名称。

5. 创建 Datastream 用户：

   1. 在 Google Cloud 控制台中，前往 Cloud SQL 实例 页面。

      转到“Cloud SQL 实例”

   2. 创建用户并为其分配 db_owner 和 db_denydatawriter 角色：

   CREATE USER USER_NAME FOR LOGIN YOUR_LOGIN;
   

   EXEC sp_addrolemember 'db_owner', 'USER_NAME';
   EXEC sp_addrolemember 'db_denydatawriter', 'USER_NAME';
   

事务日志 CDC 方法所需的额外步骤

只有在配置源 SQL Server 数据库以与事务日志 CDC 方法搭配使用时，才需要执行本部分中所述的步骤。

1. 设置轮询间隔，以便更改在源上可用。

   USE [DATABASE_NAME]
   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。这意味着事务日志会保留 86,399 秒（一天）的更改。执行 sp_cdc_start_job 'capture 过程会启动 这些设置。

2. 设置日志截断保护措施。

   如需确保 CDC 读取器有足够的时间读取日志，同时允许日志截断以防止耗尽存储空间，您可以设置日志截断保护措施：

   1. 使用 SQL Server 客户端连接到数据库。

   2. 在数据库中创建虚拟表：

      USE [DATABASE_NAME];
      CREATE TABLE dbo.gcp_datastream_truncation_safeguard (
        [id] INT IDENTITY(1,1) PRIMARY KEY,
        CreatedDate DATETIME DEFAULT GETDATE(),
        [char_column] CHAR(8)
        );
      

   3. 创建一个存储过程，该过程会在您指定的时间段内运行活跃事务，以防止日志截断：

      CREATE PROCEDURE [dbo].[DatastreamLogTruncationSafeguard] @transaction_logs_retention_time INT
      AS
      BEGIN
        -- Start a new transaction
        BEGIN TRANSACTION;
        INSERT INTO dbo.gcp_datastream_truncation_safeguard (char_column) VALUES ('a')
      
      DECLARE @formatted_time VARCHAR(5)
      SET @formatted_time = CONVERT(VARCHAR(5), DATEADD(MINUTE, @transaction_logs_retention_time, 0), 108);
        -- Wait for X minutes before ending the transaction
        WAITFOR DELAY @formatted_time;
        -- Commit the transaction
        COMMIT TRANSACTION;
      END;
      

   4. 创建另一个存储过程。这次，您将创建一个作业，该作业会按照指定的节奏运行在上一步中创建的存储过程：

      CREATE PROCEDURE [dbo].[SetUpDatastreamJob] @transaction_logs_retention_time INT
      AS
      BEGIN
        DECLARE @database_name VARCHAR(MAX)