SQL Server で変更の追跡を有効にする

プレビュー

LakeFlow Connect はゲートパブリックプレビュー段階です。プレビューに参加するには、Databricks アカウントチームにお問い合わせください。

この記事では、SQL Serverで変更追跡を有効にする方法について説明します。 Databricks へのインジェストには、変更追跡または CDC が必要です。選択するオプションのガイダンスについては、「変更の追跡とチェンジデータキャプチャ」を参照してください。

データベースでの変更追跡の有効化

以下を実行し、 <database-name> を変更追跡を有効にするデータベースの名前に置き換えます。 CHANGE_RETENTION を、ゲートウェイがダウンする可能性がある最大時間に設定します。この時間が経過すると、ゲートウェイを再開するには完全な更新が必要になります。

ALTER DATABASE <database-name> SET CHANGE_TRACKING = ON (CHANGE_RETENTION = 3 DAYS, AUTO_CLEANUP = ON)

テーブルで変更の追跡を有効にする

以下を実行し、 <schema-name>.<table-name> を変更追跡を有効にするスキーマとテーブルの名前に置き換えます。

ALTER TABLE <schema-name>.<table-name> ENABLE CHANGE_TRACKING

取り込まれたテーブルに対する VIEW CHANGE TRACKING の付与

ソース設定で説明されている権限に加えて、データベース・ユーザーには、取り込まれたテーブルまたは追跡対象のテーブルを含むスキーマに対するVIEW CHANGE TRACKING権限が必要です。

スキーマレベルの権限を付与するには、次のコマンドを実行します。

GRANT VIEW CHANGE TRACKING ON SCHEMA::<schema-name> TO <cdc-username>;

テーブルレベルのパーミッションを付与するには、次のコマンドを実行します。

GRANT VIEW CHANGE TRACKING ON OBJECT::<schema-name>.<table-name> TO <cdc-username>;

DDL キャプチャとスキーマの進化を設定する

SQL Server コネクタは、取り込まれたデータベースオブジェクトのデータ定義言語 (DDL) を追跡し、関連するテーブルスキーマの変更を宛先テーブルに適用したり、完全なスキーマレプリケーションの場合は新しいテーブルを追加したりできます。

DDL キャプチャを実行するには、追加のデータベースオブジェクトのセットアップが必要です (内部テーブル、ストアドプロシージャ、トリガーなど)。これを行うには、次の 2 つの方法があります。

自動: Databricks へのインジェスト用に構成されたデータベースユーザーに追加の権限を付与する必要があります。
手動: 読み取り専用ユーザーが DDL キャプチャとスキーマ進化を実行する場合は、これを選択します。データベース・オブジェクトを手動で作成する必要があります。

オプション 1: DDL キャプチャの自動セットアップ

次のコマンドを実行して、Databricks へのインジェスト用に構成されたデータベースユーザーに自動 DDL キャプチャ特権を付与します。

CREATE PROCEDURE データベース上で、次の操作を行います。
```
GRANT CREATE PROCEDURE TO <cdc-username>;
```
CREATE TABLE データベース上で、次の操作を行います。
```
GRANT CREATE TABLE TO <cdc-username>;
```

SELECT、 EXECUTE、およびスキーマの INSERT :

GRANT SELECT,INSERT,EXECUTE ON SCHEMA::dbo TO <cdc-username>;

ALTER データベース上で、次の操作を行います。

GRANT ALTER ON DATABASE::<database-name> TO <cdc-username>;

ALTER スキーマまたは取り込むすべてのテーブルで:

GRANT ALTER ON SCHEMA::[<schema-name>] TO [<cdc-username>];

オプション 2: DDL キャプチャの手動セットアップ

読み取り専用ユーザーが DDL キャプチャとスキーマの進化を実行する場合は、次のコマンドを実行して DDL キャプチャを手動で設定します。

CREATE TABLE "dbo"."replicate_io_audit_ddl_1"(
    "SERIAL_NUMBER" INT IDENTITY NOT NULL,
    "CURRENT_USER" NVARCHAR(128),
    "SCHEMA_NAME" NVARCHAR(128),
    "TABLE_NAME" NVARCHAR(128),
    "TYPE" NVARCHAR(30),
    "OPERATION_TYPE" NVARCHAR(30),
    "SQL_TXT" NVARCHAR(2000),
    "LOGICAL_POSITION" BIGINT,
    CONSTRAINT "replicate_io_audit_ddlPK" PRIMARY KEY("SERIAL_NUMBER","LOGICAL_POSITION")
)

ALTER TABLE dbo.replicate_io_audit_ddl_1 ENABLE CHANGE_TRACKING;

GRANT VIEW CHANGE TRACKING ON OBJECT::dbo.replicate_io_audit_ddl_1 TO [CdcUserName];

CREATE TRIGGER "replicate_io_audit_ddl_trigger_1"
            ON DATABASE
            AFTER ALTER_TABLE
            AS
            SET NOCOUNT ON;
        DECLARE @DbName nvarchar(255), @SchemaName nvarchar(max), @TableName nvarchar(255), @data XML, @operation NVARCHAR(30), @isCTEnabledDBLevel bit, @isCTEnabledTableLevel bit;
        SET @data = EVENTDATA();
        SET @DbName = DB_NAME();
        SET @SchemaName = @data.value(' (/EVENT_INSTANCE/SchemaName) [1] ', ' NVARCHAR(MAX) ');
        SET @TableName = @data.value(' (/EVENT_INSTANCE/ObjectName) [1] ', ' NVARCHAR(255) ');
        SET @operation = @data.value(' (/EVENT_INSTANCE/EventType) [1] ', ' NVARCHAR(30) ');
        SET @isCTEnabledDBLevel =
                (SELECT COUNT(*) FROM sys.change_tracking_databases WHERE database_id = DB_ID(@DbName));
        SET @isCTEnabledTableLevel = (SELECT COUNT(*)
                                      FROM sys.change_tracking_tables
                                      WHERE object_id = object_id('[' + @SchemaName + '].[' + @TableName + ']'));
        IF (@isCTEnabledDBLevel = 1 AND @isCTEnabledTableLevel = 1)
            BEGIN
                INSERT INTO dbo.replicate_io_audit_ddl_1 ("CURRENT_USER", "SCHEMA_NAME", "TABLE_NAME", "TYPE",
                                                          "OPERATION_TYPE", "SQL_TXT", "LOGICAL_POSITION")
                VALUES (SUSER_NAME(),
                        @data.value(' (/EVENT_INSTANCE/SchemaName) [1] ', ' NVARCHAR(128) '),
                        @data.value(' (/EVENT_INSTANCE/EventType) [1] ', ' NVARCHAR(128) '),
                        @data.value(' (/EVENT_INSTANCE/ObjectType) [1] ', ' NVARCHAR(30) '),
                        @data.value(' (/EVENT_INSTANCE/EventType) [1] ', ' NVARCHAR(30) '),
                        @data.value(' (/EVENT_INSTANCE/TSQLCommand/CommandText) [1] ', ' NVARCHAR(2000) '),
                        CHANGE_TRACKING_CURRENT_VERSION());
            END

GRANT VIEW DEFINITION ON DATABASE::[YourDatabaseName] TO [CdcUserName];