Configure an Amazon RDS SQL Server database for CDC

This page describes how to configure change data capture (CDC) to stream data from an Amazon RDS SQL Server database to a supported destination, such as BigQuery or Cloud Storage.

  1. Enable change data capture (CDC) for your source database. To enable CDC for your source database, connect to the database and run the following command at a SQL prompt, in a terminal, or using the Amazon RDS dashboard:

    EXECmsdb.dbo.rds_cdc_enable_db'DATABASE_NAME'

    Replace DATABASE_NAME with the name of your source database.

  2. Enable CDC on each table for which you need to capture changes:

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

    Replace the following:

    • DATABASE_NAME: the name of your source database
    • SCHEMA_NAME: the name of the schema to which the tables belong
    • TABLE_NAME: the name of the table for which you want to enable CDC

  3. Start the SQL Server Agent and make sure it's running at all times. If the SQL Server Agent remains down for an extended period, the logs might get truncated, leading to a permanent loss of the change data that wasn't read by Datastream.

    For information about running the SQL Server Agent, see Start, stop, or restart an instance of the SQL Server Agent.

  4. Enable snapshot isolation.

    When you backfill data from your SQL Server database, it's important to ensure consistent snapshots. If you don't apply the settings described in this section, changes made to the database during the backfill process might lead to duplicates or incorrect results. Applying snapshot isolation settings is required if your stream includes tables without primary keys.

    Enabling snapshot isolation creates a temporary view of your database at the start of the backfill process. This ensures that the data being copied remains consistent, even if other users are making changes to the live tables at the same time. Enabling snapshot isolation might have a slight performance impact, but it's essential for reliable data extraction.

    To enable snapshot isolation:

    1. Connect to your database using a SQL Server client.
    2. Run the following command:
    ALTERDATABASEDATABASE_NAMESETALLOW_SNAPSHOT_ISOLATIONON;

    Replace DATABASE_NAME with the name of you database.

  5. Create a Datastream user:

    1. Connect to the source database and enter the following command:

      USEDATABASE_NAME;

    2. Create a login to use while setting up the connection profile in Datastream.

      CREATELOGINYOUR_LOGINWITHPASSWORD='PASSWORD';

    3. Create a user and assign the db_owner and db_denydatawriter roles to them:

      CREATEUSERUSER_NAMEFORLOGINYOUR_LOGIN;

      EXECsp_addrolemember'db_owner','USER_NAME';
EXECsp_addrolemember'db_denydatawriter','USER_NAME';

    4. Add this user to the master database:

      USEmaster;
CREATEUSERUSER_NAMEFORLOGINYOUR_LOGIN;

Additional steps required for the transaction logs CDC method

The steps described in this section are only required when you configure your source SQL Server database for use with the transaction logs CDC method.

  1. Grant SELECT permissions for the sys.fn_dblog function.

    USEmaster;
GRANTSELECTONsys.fn_dblogTOUSER_NAME;

  2. Add your user to the msdb database and assign the following permissions to them:

    USEmsdb;
CREATEUSERUSER_NAMEFORLOGINYOUR_LOGIN;
GRANTSELECTONdbo.sysjobsTOUSER_NAME;

  3. Assign the following permissions to your user in the master database: 

    USEmaster;
GRANTVIEWSERVERSTATETOYOUR_LOGIN;

  4. Set the polling interval for which you want the changes to be available on your source.

    USE[DATABASE_NAME]
EXECsys.sp_cdc_change_job@job_type='capture',@pollinginterval=86399
EXECsp_cdc_stop_job'capture'
EXECsp_cdc_start_job'capture'

    The @pollinginterval parameter is measured in seconds with a recommended value set to 86399. This means that the transaction log retains changes for 86,399 seconds (one day). Executing the sp_cdc_start_job 'capture procedure initiates the settings.

  5. Set up a log truncation safeguard.

    To make sure that the CDC reader has enough time to read the logs while allowing log truncation to prevent using up the storage space, you can set up a log truncation safeguard:

    1. Connect to the database using a SQL Server client.

    2. Create a dummy table in the database:

      USE[DATABASE_NAME];
CREATETABLEdbo.gcp_datastream_truncation_safeguard(
[id]INTIDENTITY(1,1)PRIMARYKEY,
CreatedDateDATETIMEDEFAULTGETDATE(),
[char_column]CHAR(8)
);

    3. Create a stored procedure that runs an active transaction for a period that you specify to prevent log truncation:

      CREATEPROCEDURE[dbo].[DatastreamLogTruncationSafeguard]@transaction_logs_retention_timeINT
AS
BEGIN
-- Start a new transaction
BEGINTRANSACTION;
INSERTINTOdbo.gcp_datastream_truncation_safeguard(char_column)VALUES('a')
DECLARE@formatted_timeVARCHAR(5)
SET@formatted_time=CONVERT(VARCHAR(5),DATEADD(MINUTE,@transaction_logs_retention_time,0),108);
-- Wait for X minutes before ending the transaction
WAITFORDELAY@formatted_time;
-- Commit the transaction
COMMITTRANSACTION;
END;

    4. Create another stored procedure. This time, you create a job that runs the stored procedure that you created in the previous step according to a specified cadence:

      CREATEPROCEDURE[dbo].[SetUpDatastreamJob]@transaction_logs_retention_timeINT
AS
BEGIN
DECLARE@database_nameVARCHAR(MAX)
SET@database_name=(SELECTDB_NAME());;
DECLARE@command_strVARCHAR(MAX);
SET@command_str=CONCAT('Use ',@database_name,'; exec dbo.DatastreamLogTruncationSafeguard @transaction_logs_retention_time = '+CAST(@transaction_logs_retention_timeASVARCHAR(10)));
DECLARE@job_nameVARCHAR(MAX);
SET@job_name=
CONCAT(@database_name,'_','DatastreamLogTruncationSafeguardJob1')
-- Add 3 schedules to the job to run again after specified time.
IF
NOTEXISTS(
SELECT*
FROMmsdb.dbo.sysjobs
WHEREname=@job_name
)
BEGIN
EXEC
msdb.dbo.sp_add_job
@job_name
=@job_name,
@enabled=1,
@description=N'Execute the procedure to run an active transaction for x minutes.';
EXECmsdb.dbo.sp_add_jobstep@job_name=@job_name,
@step_name=N'Execute_DatastreamLogTruncationSafeguard',
@subsystem=N'TSQL',
@command=@command_str;
-- Add a schedule that runs the stored procedure every given minutes starting now.
DECLARE@schedule_name_1VARCHAR(MAX);
SET@schedule_name_1=CONCAT(@database_name,'_','DatastreamEveryGivenMinutesFromNow')
DECLARE@start_time_1time;
SET@start_time_1=DATEADD(SECOND,1,GETDATE());
DECLARE@formatted_start_time_1INT;
SET@formatted_start_time_1=CONVERT(INT,REPLACE(CONVERT(VARCHAR(8),@start_time_1,114),':',''));
EXECmsdb.dbo.sp_add_schedule
@schedule_name=@schedule_name_1,
@freq_type=4,-- daily start
@freq_subday_type=4,-- every X minutes daily
@freq_interval=1,
@freq_subday_interval=@transaction_logs_retention_time,
@active_start_time=@formatted_start_time_1;
EXECmsdb.dbo.sp_attach_schedule
@job_name=@job_name,
@schedule_name=@schedule_name_1;
-- Add a schedule that runs the stored procedure after every given minutes starting after some delay.
DECLARE@schedule_name_2VARCHAR(MAX);
Set@schedule_name_2=CONCAT(@database_name,'_','DatastreamEveryGivenMinutesAfterDelay');
DECLARE@start_time_2time;
SET@start_time_2=DATEADD(MINUTE,@transaction_logs_retention_time/2,GETDATE());
DECLARE@formatted_start_time_2INT;
SET@formatted_start_time_2=CONVERT(INT,REPLACE(CONVERT(VARCHAR(8),@start_time_2,114),':',''));
EXECmsdb.dbo.sp_add_schedule
@schedule_name=@schedule_name_2,
@freq_type=4,-- daily start
@freq_subday_type=4,-- every x minutes daily
@freq_interval=1,
@freq_subday_interval=@transaction_logs_retention_time,
@active_start_time=@formatted_start_time_2;
EXECmsdb.dbo.sp_attach_schedule
@job_name=@job_name,
@schedule_name=@schedule_name_2;
-- Add a schedule that runs the stored procedure on the SQL Server Agent startup.
DECLARE@schedule_name_agent_startupVARCHAR(MAX);
Set@schedule_name_agent_startup=CONCAT(@database_name,'_','DatastreamSqlServerAgentStartupSchedule')
EXECmsdb.dbo.sp_add_schedule
@schedule_name=@schedule_name_agent_startup,
@freq_type=64,-- start on SQL Server Agent startup
@active_start_time=@formatted_start_time_1;
EXECmsdb.dbo.sp_attach_schedule
@job_name=@job_name,
@schedule_name=@schedule_name_agent_startup;
EXECmsdb.dbo.sp_add_jobserver
@job_name=@job_name,
@server_name=@@servername;
END
END;

    5. Execute the stored procedure that creates the Datastream job.

      DECLARE@transaction_logs_retention_timeINT=(INT)
EXEC[dbo].[SetUpDatastreamJob]@transaction_logs_retention_time

      Replace INT with the number of minutes for which you want to retain the logs. For example:

      • The value of 60 sets the retention time to 1 hour
      • The value of 24 * 60 sets the retention time to 1 day
      • The value of 3 * 24 * 60 sets the retention time to 3 days

What's next

Except as otherwise noted, the content of this page is licensed under the Creative Commons Attribution 4.0 License, and code samples are licensed under the Apache 2.0 License. For details, see the Google Developers Site Policies. Java is a registered trademark of Oracle and/or its affiliates.

Last updated 2026年08月26日 UTC.