セルフマネージド SQL Server データベースを構成する

次の手順では、Datastream で使用するためにセルフマネージド SQL Server データベースを構成する方法について説明します。

  1. ソース データベースで CDC を有効にします。これを行うには、データベースに接続し、SQL プロンプトまたはターミナルで次のコマンドを実行します。

    USE [DATABASE_NAME]
    GO
    EXEC sys.sp_cdc_enable_db
    GO
    

    DATABASE_NAME は、ソース データベースの名前に置き換えます。

  2. 変更を取得する必要があるテーブルで CDC を有効にします。

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

    以下を置き換えます。

    • DATABASE_NAME: ソース データベースの名前
    • SCHEMA_NAME: テーブルが属するスキーマの名前
    • TABLE_NAME: CDC を有効にするテーブルの名前
  3. SQL Server エージェントを起動し、常に実行していることを確認します。SQL Server エージェントが長期間停止している場合、ログが切り捨てられ、Datastream で読み取られなかった変更データが完全に失われる可能性があります。

    SQL Server エージェントの実行については、SQL Server エージェントのインスタンスを起動、停止、再起動するをご覧ください。

  4. Datastream ユーザーを作成する:

    1. ソース データベースに接続し、次のコマンドを入力します。

      USE DATABASE_NAME;
      
    2. Datastream で接続プロファイルを設定するときに使用するログインを作成します。

      CREATE LOGIN YOUR_LOGIN WITH PASSWORD = 'PASSWORD';
      
    3. ユーザーを作成します。

      CREATE USER USER_NAME FOR LOGIN YOUR_LOGIN;
      
    4. db_datareader ロールを割り当てます。

      EXEC sp_addrolemember 'db_datareader', 'USER_NAME';
      
    5. VIEW DATABASE STATE 権限を付与します。

      GRANT VIEW DATABASE STATE TO 'USER_NAME';
      
    6. このユーザーを master データベースに追加します。

      USE master;
      CREATE USER USER_NAME FOR LOGIN YOUR_LOGIN;
      

トランザクション ログの CDC メソッドに必要な追加手順

このセクションで説明する手順は、トランザクション ログ のCDC メソッドで使用するためにソース SQL Server データベースを構成する場合にのみ必要です。

  1. ソース データベースに接続し、ユーザーに db_owner ロールと db_denydatawriter ロールを割り当てます。

    USE DATABASE_NAME;
    EXEC sp_addrolemember 'db_owner', 'USER_NAME';
    EXEC sp_addrolemember 'db_denydatawriter', 'USER_NAME';
    
  2. sys.fn_dblog 関数に SELECT 権限を付与します。

    USE master;
    GRANT SELECT ON sys.fn_dblog TO USER_NAME;
    
  3. ユーザーを msdb データベースに追加し、次の権限を割り当てます。

    USE msdb;
    CREATE USER USER_NAME FOR LOGIN YOUR_LOGIN;
    GRANT SELECT ON dbo.sysjobs TO USER_NAME;
    
  4. master データベース内のユーザーに次の権限を割り当てます。

      USE master;
      GRANT VIEW SERVER STATE TO YOUR_LOGIN;
    
  5. ソースで変更を有効にする保持期間を設定します。

    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 秒間(1 日)変更を保持します。sp_cdc_start_job 'capture プロシージャを実行すると、設定が開始されます。

  6. データベースでクリーンアップ ジョブまたはキャプチャ ジョブが実行されている場合は、これらのジョブを停止します。詳細については、変更データ キャプチャの管理とモニタリングをご覧ください。

  7. ログの切り捨ての防止対策を設定します。

    CDC のリーダーがログの読み取りに十分な時間を確保しながら、ログの切り捨てによって保存容量が不足しないようにするには、ログの切り捨ての防止対策を設定します。

    1. SQL Server クライアントを使用してデータベースに接続します。
    2. ログの切り捨てを防止するために指定した期間にアクティブなトランザクションを実行するストアド プロシージャを作成します。

      CREATE PROCEDURE dbo.DatastreamLogTruncationSafeguard @transaction_logs_retention_time INT
      AS
      BEGIN
      
      DECLARE @transactionLog TABLE (beginLSN BINARY(10), endLSN BINARY(10))
      INSERT @transactionLog EXEC sp_repltrans
      
      DECLARE @currentDateTime DATETIME = GETDATE()
      DECLARE @cutoffDateTime DATETIME = DATEADD(MINUTE, -@transaction_logs_retention_time, @currentDateTime)
      
      DECLARE @firstValidLSN BINARY(10) = NULL
      DECLARE @lastValidLSN BINARY(10) = NULL
      DECLARE @firstTxnTime DATETIME = NULL
      DECLARE @lastTxnTime DATETIME = NULL
      
      SELECT TOP 1
          @lastTxnTime = t.logStartTime,
          @lastValidLSN = t.beginLSN
      FROM (
        SELECT
          beginLSN AS beginLSN,
          (SELECT TOP 1 [begin time]
          FROM fn_dblog(stuff(stuff(CONVERT(CHAR(24), beginLSN, 1), 19, 0, ':'), 11, 0, ':'), DEFAULT)) AS logStartTime
        FROM @transactionLog
      ) t
      ORDER BY t.beginLSN DESC
      
      -- If all transactions are before cutoff, clear everything
      IF (@lastTxnTime < @cutoffDateTime)
      BEGIN
          EXEC sp_repldone NULL, NULL, 0, 0, 1
      END
      ELSE
      BEGIN
          -- Find the earliest transaction
          SELECT TOP 1
            @firstTxnTime = t.logStartTime,
            @firstValidLSN = ISNULL(@firstValidLSN, t.beginLSN)
          FROM (
            SELECT
              beginLSN AS beginLSN,
              (SELECT TOP 1 [begin time]
              FROM fn_dblog(stuff(stuff(CONVERT(CHAR(24), beginLSN, 1), 19, 0, ':'), 11, 0, ':'), DEFAULT)) AS logStartTime
            FROM @transactionLog
          ) t
          ORDER BY t.beginLSN ASC
      
          IF (@firstTxnTime < @cutoffDateTime)
          BEGIN
              -- Identify the earliest and latest LSNs within VLogs before cutoff
              SELECT
                @firstValidLSN = SUBSTRING(MAX(t.lsnMarkers), 1, 10),
                @lastValidLSN = SUBSTRING(MAX(t.lsnMarkers), 11, 10)
              FROM (
                SELECT MIN(beginLSN + endLSN) AS lsnMarkers
                FROM @transactionLog
                GROUP BY SUBSTRING(beginLSN, 1, 4)
              ) t
              WHERE (
                SELECT TOP 1 [begin time]
                FROM fn_dblog(stuff(stuff(CONVERT(CHAR(24), t.lsnMarkers, 1), 19, 0, ':'), 11, 0, ':'), DEFAULT)
                WHERE Operation = 'LOP_BEGIN_XACT'
              ) < @cutoffDateTime
      
              EXEC sp_repldone @firstValidLSN, @lastValidLSN, 0, 0, 0
          END
        END
      END;
      
    3. 別のストアド プロシージャを作成します。今回は、前の手順で作成したストアド プロシージャを指定した間隔に従って実行するジョブを作成します。

      CREATE PROCEDURE [dbo].[SetUpDatastreamJob] @transaction_logs_retention_time INT
      AS
      BEGIN
      
      DECLARE @database_name VARCHAR(MAX)
        SET @database_name =  (SELECT DB_NAME());;
      
        DECLARE @command_str VARCHAR(MAX);
        SET @command_str = CONCAT('Use ', @database_name,'; exec dbo.DatastreamLogTruncationSafeguard @transaction_logs_retention_time = ' + CAST(@transaction_logs_retention_time AS VARCHAR(10)));
      
        DECLARE @job_name VARCHAR(MAX);
      SET @job_name =
        CONCAT(@database_name, '_', 'DatastreamLogTruncationSafeguardJob1')
          DECLARE @current_time INT
        = CAST(FORMAT(GETDATE(), 'HHmmss') AS INT);
      
        -- Schedule the procedure to run after every 5 minutes.
        IF NOT EXISTS (
          SELECT * FROM msdb.dbo.sysjobs
          WHERE name = @job_name
        )
        BEGIN
          EXEC msdb.dbo.sp_add_job
          @job_name = @job_name,
          @enabled = 1,
          @description = N'Execute the procedure every 5 minutes.' ;
      
          EXEC msdb.dbo.sp_add_jobstep
          @job_name =  @job_name,
          @step_name = N'Execute_DatastreamLogTruncationSafeguard',
          @subsystem = N'TSQL',
          @command = @command_str;
      
            DECLARE @schedule_name_1 VARCHAR(MAX);
          SET @schedule_name_1 = CONCAT(@database_name, '_', 'DatastreamEveryFiveMinutesSchedule')
      
          EXEC msdb.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 = 5,
          @active_start_time = @current_time;
      
          EXEC msdb.dbo.sp_attach_schedule
          @job_name = @job_name,
          @schedule_name = @schedule_name_1 ;
      
          -- Add a schedule that runs the stored procedure on the SQL Server Agent startup.
          DECLARE @schedule_name_agent_startup VARCHAR(MAX);
          SET @schedule_name_agent_startup = CONCAT(@database_name, '_', 'DatastreamSqlServerAgentStartupSchedule')
      
          EXEC msdb.dbo.sp_add_schedule
          @schedule_name = @schedule_name_agent_startup,
          @freq_type = 64,  -- start on SQL Server Agent startup
          @active_start_time = @current_time;
      
          EXEC msdb.dbo.sp_attach_schedule
          @job_name = @job_name,
          @schedule_name = @schedule_name_agent_startup ;
      
          EXEC msdb.dbo.sp_add_jobserver
          @job_name = @job_name,
          @server_name = @@servername ;
        END
      END;
      
    4. データストリーム ジョブを作成するストアド プロシージャを実行します。

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

      INT は、ログを保持する分数に置き換えます。次に例を示します。

      • 値が 60 の場合、保持期間は 1 時間です。
      • 値が 24 * 60 の場合、保持期間は 1 日です。
      • 値が 3 * 24 * 60 の場合、保持期間は 3 日です。

次のステップ