Mengonfigurasi database SQL Server yang dikelola sendiri untuk CDC

Halaman ini menjelaskan cara mengonfigurasi change data capture (CDC) untuk mengalirkan data dari database SQL Server yang dikelola sendiri ke tujuan yang didukung, seperti BigQuery atau Cloud Storage.

1. Aktifkan CDC untuk database sumber Anda. Untuk melakukannya, hubungkan ke database dan jalankan perintah berikut di prompt SQL atau di terminal:

   USE [DATABASE_NAME]
   GO
   EXEC sys.sp_cdc_enable_db
   GO
   

   Ganti DATABASE_NAME dengan nama database sumber Anda.

2. Aktifkan CDC pada tabel yang perubahannya perlu Anda ambil:

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

   Catatan: Anda harus menjalankan perintah untuk setiap tabel yang ingin Anda aktifkan CDC-nya.

   Ganti kode berikut:

   • DATABASE_NAME: nama database sumber Anda
   • SCHEMA_NAME: nama skema tempat tabel berada
   • TABLE_NAME: nama tabel yang ingin Anda aktifkan CDC-nya

3. Mulai SQL Server Agent dan pastikan SQL Server Agent selalu berjalan. Jika SQL Server Agent tetap tidak aktif dalam jangka waktu yang lama, log dapat terpotong, sehingga menyebabkan hilangnya data perubahan secara permanen yang tidak dibaca oleh Datastream.

   Untuk mengetahui informasi tentang cara menjalankan SQL Server Agent, lihat Memulai, menghentikan, atau memulai ulang instance SQL Server Agent.

   Catatan: Sebaiknya aktifkan restart otomatis untuk layanan SQL Server Agent agar agen tidak nonaktif dalam waktu yang lebih lama.

4. Aktifkan isolasi snapshot.

   Saat mengisi ulang data dari database SQL Server, Anda harus memastikan snapshot yang konsisten. Jika Anda tidak menerapkan setelan yang dijelaskan di bagian ini, perubahan yang dilakukan pada database selama proses pengisian ulang dapat menyebabkan duplikat atau hasil yang salah, terutama untuk tabel tanpa kunci utama.

   Mengaktifkan isolasi snapshot akan membuat tampilan sementara database Anda di awal proses pengisian ulang. Hal ini memastikan bahwa data yang disalin tetap konsisten, meskipun pengguna lain membuat perubahan pada tabel aktif pada saat yang sama. Mengaktifkan isolasi snapshot mungkin memiliki sedikit dampak performa, tetapi penting untuk ekstraksi data yang andal.

   Untuk mengaktifkan isolasi snapshot:

   1. Hubungkan ke database Anda menggunakan klien SQL Server.
   2. Jalankan perintah berikut:

   ALTER DATABASE DATABASE_NAME SET ALLOW_SNAPSHOT_ISOLATION ON;
   

   Ganti DATABASE_NAME dengan nama database Anda.

5. Buat pengguna Datastream:

   1. Hubungkan ke database sumber dan masukkan perintah berikut:

      USE DATABASE_NAME;
      

   2. Buat login untuk digunakan saat menyiapkan profil koneksi di Datastream.

      CREATE LOGIN YOUR_LOGIN WITH PASSWORD = 'PASSWORD';
      

   3. Buat pengguna:

      CREATE USER USER_NAME FOR LOGIN YOUR_LOGIN;
      

   4. Tetapkan peran db_datareader kepada mereka:

      EXEC sp_addrolemember 'db_datareader', 'USER_NAME';
      

   5. Beri mereka izin VIEW DATABASE STATE:

      GRANT VIEW DATABASE STATE TO USER_NAME;
      

   6. Tambahkan pengguna ini ke database master:

      USE master;
      CREATE USER USER_NAME FOR LOGIN YOUR_LOGIN;
      

Langkah tambahan yang diperlukan untuk metode CDC log transaksi

Langkah-langkah yang dijelaskan di bagian ini hanya diperlukan saat Anda mengonfigurasi database SQL Server sumber untuk digunakan dengan metode CDC log transaksi.

1. Hubungkan ke database sumber dan tetapkan peran db_owner dan db_denydatawriter kepada pengguna Anda:

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

2. Beri izin SELECT untuk fungsi sys.fn_dblog.

   USE master;
   GRANT SELECT ON sys.fn_dblog TO USER_NAME;
   

3. Tambahkan pengguna Anda ke database msdb dan tetapkan izin berikut kepadanya:

   USE msdb;
   CREATE USER USER_NAME FOR LOGIN YOUR_LOGIN;
   GRANT SELECT ON dbo.sysjobs TO USER_NAME;
   

4. Tetapkan izin berikut kepada pengguna Anda di database master:

     USE master;
     GRANT VIEW SERVER STATE TO YOUR_LOGIN;
   

5. Tetapkan periode retensi yang Anda inginkan agar perubahan tersedia di sumber Anda.

   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'
   

   Parameter @pollinginterval diukur dalam detik dengan nilai yang direkomendasikan ditetapkan ke 86399. Artinya,log transaksi menyimpan perubahan selama 86.399 detik (satu hari). Menjalankan prosedur sp_cdc_start_job 'capture akan memulai setelan.

6. Jika ada tugas pembersihan atau pengambilan yang berjalan di database Anda, hentikan tugas tersebut. Untuk mengetahui informasi selengkapnya, lihat Mengelola dan memantau pengambilan data perubahan.

7. Siapkan pengamanan pemangkasan log.

   Untuk memastikan bahwa pembaca CDC memiliki cukup waktu untuk membaca log sekaligus memungkinkan pemangkasan log untuk mencegah penggunaan ruang penyimpanan, Anda dapat menyiapkan pengamanan pemangkasan log:

   1. Hubungkan ke database menggunakan klien SQL Server.

   2. Buat prosedur tersimpan yang menjalankan transaksi aktif untuk jangka waktu yang Anda tentukan untuk mencegah pemotongan log:

      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. Buat prosedur tersimpan lainnya. Kali ini, Anda membuat tugas yang menjalankan prosedur tersimpan yang Anda buat di langkah sebelumnya sesuai dengan irama yang ditentukan:

      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. Jalankan prosedur tersimpan yang membuat tugas Datastream.

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

      Ganti INT dengan jumlah menit yang Anda inginkan untuk mempertahankan log. Contoh:

      • Nilai 60 menetapkan waktu retensi menjadi 1 jam
      • Nilai 24 * 60 menetapkan waktu retensi ke 1 hari
      • Nilai 3 * 24 * 60 menetapkan waktu retensi menjadi 3 hari

Langkah berikutnya

• Pelajari lebih lanjut cara kerja Datastream dengan sumber SQL Server.