‫Google משתמשת בטכנולוגיית AI כדי לתרגם תוכן לשפה המועדפת עליך. בתרגומים כאלו עשויות להיות שגיאות.

הגדרת מסד נתונים של SQL Server בניהול עצמי ל-CDC

כדי להעביר בסטרימינג עדכונים של סימון נתונים שהשתנו (CDC) ממסד נתונים של SQL Server בניהול עצמי באמצעות Datastream, אתם יכולים להשתמש בטבלאות שינויים או ביומני טרנזקציות. שתי השיטות משכפלות את השינויים ביעדים נתמכים, כמו BigQuery או Cloud Storage. לפני שיוצרים סטרימינג, צריך להפעיל את CDC ואת בידוד ה-snapshot, להריץ את SQL Server Agent, ליצור משתמש Datastream ולהגדיר את שמירת היומנים.

הגדרת מסד הנתונים של המקור

כדי להגדיר את מסד הנתונים של המקור:

מוודאים שמסד הנתונים משתמש במודל השחזור המלא. מתחברים למסד הנתונים ומריצים את הפקודה הבאה בשורת הפקודה של SQL או בטרמינל:
```
USE master
GO
ALTER DATABASE [DATABASE_NAME] SET RECOVERY FULL
GO
```
מחליפים את DATABASE_NAME בשם של מסד הנתונים במקור.

הערה: אחרי שמגדירים את מודל השחזור ל-FULL, מסד הנתונים נשאר במצב שחזור פסאודו-פשוט עד שמבצעים גיבוי מלא של מסד הנתונים. צריך לבצע גיבוי מלא של מסד הנתונים לפני שיומן העסקאות יתחיל לשמור רשומות לסימון נתונים שהשתנו.
מפעילים את CDC במסד הנתונים של המקור. כדי לעשות את זה, מתחברים למסד הנתונים ומריצים את הפקודה הבאה בשורת הפקודה של SQL או בטרמינל:
```
USE [DATABASE_NAME]
GO
EXEC sys.sp_cdc_enable_db
GO
```
מחליפים את DATABASE_NAME בשם של מסד הנתונים במקור.
מפעילים את CDC בטבלאות שבהן רוצים לתעד שינויים:
```
USE [DATABASE_NAME]
EXEC sys.sp_cdc_enable_table
@source_schema = N'SCHEMA_NAME',
@source_name = N'TABLE_NAME',
@role_name = NULL
GO
```
הערה: צריך להריץ את הפקודה לכל טבלה שרוצים להפעיל בה CDC.

מחליפים את מה שכתוב בשדות הבאים:
- ‫DATABASE_NAME: השם של מסד הנתונים במקור
- ‫SCHEMA_NAME: שם הסכימה שאליה שייכות הטבלאות
- ‫TABLE_NAME: שם הטבלה שבה רוצים להפעיל את ה-CDC
מפעילים את SQL Server Agent ומוודאים שהוא פועל כל הזמן. אם SQL Server Agent לא פועל במשך תקופה ארוכה, יכול להיות שהיומנים ייחתכו, וכתוצאה מכך הנתונים שהשתנו ולא נקראו על ידי Datastream יאבדו באופן סופי.

למידע על הפעלת SQL Server Agent, קראו את המאמר הפעלה, עצירה או הפעלה מחדש של מכונה של SQL Server Agent.

הערה: מומלץ להגדיר הפעלה מחדש אוטומטית של שירות SQL Server Agent כדי למנוע השבתה של הסוכן לזמן ארוך יותר.
הפעלת בידוד snapshot.

כשממלאים נתונים ממסד הנתונים של SQL Server, חשוב לוודא שתמונות המצב עקביות. אם לא תפעילו את ההגדרות שמתוארות בקטע הזה, שינויים שיבוצעו במסד הנתונים במהלך תהליך המילוי החוזר עלולים להוביל לכפילויות או לתוצאות שגויות. אם ה-stream כולל טבלאות בלי מפתחות ראשיים, צריך להחיל הגדרות של בידוד snapshot.

כשמפעילים בידוד snapshot, נוצרת תצוגה זמנית של מסד הנתונים בתחילת תהליך המילוי של החוסרים בנתונים. כך אפשר לוודא שהנתונים שמועתקים נשארים עקביים, גם אם משתמשים אחרים מבצעים שינויים בטבלאות בזמן אמת באותו הזמן. הפעלת בידוד ה-snapshot עשויה להשפיע מעט על הביצועים, אבל היא חיונית לחילוץ נתונים אמין.

כדי להפעיל בידוד snapshot:
1. מתחברים למסד הנתונים באמצעות לקוח של SQL Server.
2. מריצים את הפקודה הבאה:
```
ALTER DATABASE DATABASE_NAME SET ALLOW_SNAPSHOT_ISOLATION ON;
```
מחליפים את DATABASE_NAME בשם של מסד הנתונים.
יוצרים משתמש ב-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 של יומני עסקאות

השלבים שמתוארים בקטע הזה נדרשים רק כשמגדירים את מסד הנתונים של SQL Server במקור לשימוש בשיטת CDC של יומני טרנזקציות.

מתחברים למסד הנתונים של המקור ומקצים למשתמש את התפקידים db_owner ו-db_denydatawriter:

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

נותנים הרשאות של SELECT לפונקציה sys.fn_dblog.
```
USE master;
GRANT SELECT ON sys.fn_dblog TO USER_NAME;
```
מוסיפים את המשתמש למסד הנתונים msdb ומקצים לו את ההרשאות הבאות:
```
USE msdb;
CREATE USER USER_NAME FOR LOGIN YOUR_LOGIN;
GRANT SELECT ON dbo.sysjobs TO USER_NAME;
```
מקצים את ההרשאות הבאות למשתמש במסד הנתונים של master:
```
  USE master;
  GRANT VIEW SERVER STATE TO YOUR_LOGIN;
```
מגדירים את מרווח הזמן לדגימה שבו רוצים שהשינויים יהיו זמינים במקור.
```
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 מאתחלת את ההגדרות.
אם יש משימות ניקוי או תיעוד שפועלות במסד הנתונים, צריך לעצור אותן. מידע נוסף זמין במאמר ניהול ומעקב אחרי סימון נתונים שהשתנו.

הגדרת אמצעי הגנה מפני קיטוע של יומנים.

כדי לוודא שלקורא ה-CDC יש מספיק זמן לקרוא את היומנים, וגם לאפשר חיתוך של היומנים כדי למנוע ניצול של כל נפח האחסון, אפשר להגדיר אמצעי הגנה לחיתוך היומנים:

מתחברים למסד הנתונים באמצעות לקוח של SQL Server.

כדי למנוע קיטוע של היומן, יוצרים פרוצדורה מאוחסנת שמריצה טרנזקציה פעילה למשך פרק הזמן שציינתם:

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), SUBSTRING(t.lsnMarkers, 1, 10), 1), 19, 0, ':'), 11, 0, ':'), DEFAULT)
          WHERE Operation = 'LOP_BEGIN_XACT'
        ) < @cutoffDateTime

        EXEC sp_repldone @firstValidLSN, @lastValidLSN, 0, 0, 0
    END
  END
END;

יוצרים עוד תהליך מאוחסן. הפעם, יוצרים משימה שמריצה את הפרוצדורה המאוחסנת שיצרתם בשלב הקודם, לפי קצב מוגדר:

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;

מריצים את התהליך המאוחסן שיוצר את המשימה ב-Datastream.
```
DECLARE @transaction_logs_retention_time INT = (INT)
EXEC [dbo].[SetUpDatastreamJob] @transaction_logs_retention_time
```
מחליפים את INT במספר הדקות שבהן רוצים לשמור את היומנים. לדוגמה:
- הערך 60 מגדיר את זמן השמירה לשעה אחת
- הערך 24 * 60 מגדיר את משך השמירה ליום אחד
- הערך של 3 * 24 * 60 מגדיר את משך השמירה ל-3 ימים

המאמרים הבאים

איך Datastream עובד עם מקורות SQL Server