8.5. スキーマ変更後のキャプチャーテーブルの更新

手順

1. データベースを更新するアプリケーションを一時停止します。
2. Debezium コネクターがストリーミングされていない変更イベントレコードをすべてストリーミングするまで待ちます。
3. Debezium コネクターを停止します。
4. すべての変更をソーステーブルスキーマに適用します。
5. パラメーター @capture_instance の一意の値で sys.sp_cdc_enable_table の手順を使用して、更新ソーステーブルの新しいキャプチャーテーブルを作成します。
6. ステップ 1 で一時停止したアプリケーションを再開します。
7. Debezium コネクターを起動します。
8. Debezium コネクターが新しいキャプチャーテーブルからストリーミングを開始したら、古いキャプチャーインスタンス名に設定されたパラメーター @capture_instance でストアドプロシージャー sys.sp_cdc_disable_table を実行して、古いキャプチャーテーブルを削除します。

手順

1. すべての変更をソーステーブルスキーマに適用します。
2. パラメーター @capture_instance に一意の値を指定して sys.sp_cdc_enable_table ストアドプロシージャーを実行し、更新元テーブルに新しいキャプチャテーブルを作成します。
3. Debezium が新しいキャプチャーテーブルからのストリーミングを開始したら、パラメーター @capture_instance に古いキャプチャーインスタンス名を設定して、sys.sp_cdc_disable_table ストアドプロシージャーを実行することで、古いキャプチャーテーブルを削除することができます。

1. 次のクエリーを実行して customers ソーステーブルのスキーマを変更し、phone_number フィールドを追加します。

   ALTER TABLE customers ADD phone_number VARCHAR(32);

2. sys.sp_cdc_enable_table ストアドプロシージャーを実行して、新しいキャプチャーインスタンスを作成します。

   EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'customers', @role_name = NULL, @supports_net_changes = 0, @capture_instance = 'dbo_customers_v2';
   GO

3. 次のクエリーを実行して、customers テーブルに新しいデータを挿入します。

   INSERT INTO customers(first_name,last_name,email,phone_number) VALUES ('John','Doe','john.doe@example.com', '+1-555-123456');
   GO

   Kafka Connect ログは、以下のメッセージのようなエントリーで設定の更新を報告します。

   connect_1    | 2019-01-17 10:11:14,924 INFO   ||  Multiple capture instances present for the same table: Capture instance "dbo_customers" [sourceTableId=testDB.dbo.customers, changeTableId=testDB.cdc.dbo_customers_CT, startLsn=00000024:00000d98:0036, changeTableObjectId=1525580473, stopLsn=00000025:00000ef8:0048] and Capture instance "dbo_customers_v2" [sourceTableId=testDB.dbo.customers, changeTableId=testDB.cdc.dbo_customers_v2_CT, startLsn=00000025:00000ef8:0048, changeTableObjectId=1749581271, stopLsn=NULL]   [io.debezium.connector.sqlserver.SqlServerStreamingChangeEventSource]
   connect_1    | 2019-01-17 10:11:14,924 INFO   ||  Schema will be changed for ChangeTable [captureInstance=dbo_customers_v2, sourceTableId=testDB.dbo.customers, changeTableId=testDB.cdc.dbo_customers_v2_CT, startLsn=00000025:00000ef8:0048, changeTableObjectId=1749581271, stopLsn=NULL]   [io.debezium.connector.sqlserver.SqlServerStreamingChangeEventSource]
   ...
   connect_1    | 2019-01-17 10:11:33,719 INFO   ||  Migrating schema to ChangeTable [captureInstance=dbo_customers_v2, sourceTableId=testDB.dbo.customers, changeTableId=testDB.cdc.dbo_customers_v2_CT, startLsn=00000025:00000ef8:0048, changeTableObjectId=1749581271, stopLsn=NULL]   [io.debezium.connector.sqlserver.SqlServerStreamingChangeEventSource]

   最終的には、phone_number フィールドがスキーマに追加され、その値が Kafka トピックに書き込まれたメッセージに表示されます。

   ...
        {
           "type": "string",
           "optional": true,
           "field": "phone_number"
        }
   ...
       "after": {
         "id": 1005,
         "first_name": "John",
         "last_name": "Doe",
         "email": "john.doe@example.com",
         "phone_number": "+1-555-123456"
       },

4. sys.sp_cdc_disable_table ストアドプロシージャーを実行して、古いキャプチャーインスタンスを削除します。

   EXEC sys.sp_cdc_disable_table @source_schema = 'dbo', @source_name = 'dbo_customers', @capture_instance = 'dbo_customers';
   GO