Azure Database をソースとして Confluent へマネージド Debezium により CDC（変更データキャプチャー）の検証結果

## 概要
本記事では、Azure Database をソースとして、Confluent で提供される Debezium を用いて CDC（Change Data Capture）を実行する手順と、その実行結果についてご紹介します。Azure Database に対する SQL の実行を Azure Data Studio で実施しています。

本記事は、以下のシリーズの一部です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1ff6386e-0d84-e62b-dd54-8e6164c4ba5b.png)

引用元：[Data in Motion 実現へ：Data Streaming Platform である Confluent の全貌 #Kafka - Qiita](https://qiita.com/manabian/items/f8281514e9afb9a8701c)

## 環境準備

### Azure Database の環境構築

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/eb53d640-399e-6a7b-81a2-1af489337426.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/866aa041-a41e-247b-7b6a-520f604f9e0c.png)

### データベースでの CDC の有効化

```sql
EXEC sys.sp_cdc_enable_db;
```

```sql
SELECT name, is_cdc_enabled
FROM sys.databases  
WHERE name = 'debezium';
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/94964dd7-1f8f-25ff-5932-58a3466ef0b9.png)

### サンプルのテーブルを作成

```sql
IF OBJECT_ID(N'dbo.test_01', N'U') IS NOT NULL
    DROP TABLE [dbo].[test_01];
GO
CREATE TABLE [dbo].[test_01] (
    [Id] INT NOT NULL,
    CONSTRAINT [PK_Customers] PRIMARY KEY ([Id])
);
GO
```

```sql
EXEC sys.sp_cdc_enable_table
    @source_schema = N'dbo',
    @source_name = N'test_01',
    @role_name = NULL;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c40c7fcf-22d8-9d28-c080-e7a608aab9a9.png)

本手順が必要な理由は下記記事にて説明しています。

- [Azure SQL Database に対して Microsoft SQL Server CDC Source V2 (Debezium) コネクタを設定する際の CDC schema に接続できないというエラーへの対応方法 #Kafka - Qiita](https://qiita.com/manabian/items/c05ebccd12bf07be0eac)

### Confluent Cloud での Microsoft SQL Server CDC Source V2 (Debezium) コネクタの作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1828e140-8eea-1ca4-8eaa-ba355ebaf003.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/332c5b22-b116-da02-3d44-14cf8a421039.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/52cdd175-b13f-9d02-a39e-29db7463ed57.png)

## 動作検証

### 1. Primary Key を設定しているテーブルでの動作検証

#### 事前準備

まずは Customers テーブルを作成します。

```sql
-- 既存テーブルを削除 (必要に応じて)
IF OBJECT_ID(N'dbo.Customers', N'U') IS NOT NULL
    DROP TABLE [dbo].[Customers];
GO

-- Customers テーブル作成
CREATE TABLE [dbo].[Customers] (
    [Id] INT NOT NULL,
    [FirstName] NVARCHAR(255) NOT NULL,
    [LastName]  NVARCHAR(255) NOT NULL,
    [Email]     NVARCHAR(255) NOT NULL,
    [CreatedAt] DATETIME2 NOT NULL CONSTRAINT DF_Customers_CreatedAt DEFAULT (SYSUTCDATETIME()),
    CONSTRAINT [PK__Customers] PRIMARY KEY ([Id])
);
GO

EXEC sys.sp_cdc_enable_table
    @source_schema = N'dbo',
    @source_name = N'Customers',
    @role_name = NULL;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7b14a417-3cd2-b3b8-b2d2-a439bb8510bf.png)


#### INSERT 操作（create events）

以下のようにレコードを INSERT します。

```sql
INSERT INTO [dbo].[Customers] (id, [FirstName], [LastName], [Email])
VALUES
    (1, N'Anne', N'Kretchmar', N'annek@example.org');
GO
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/84a01255-782e-1309-6fa6-3bcff3ac6216.png)

Confluent のトピックを確認すると、INSERT イベントが以下のように記録されています。

```json
{
  "before": null,
  "after": {
    "Id": 1,
    "FirstName": "Anne",
    "LastName": "Kretchmar",
    "Email": "annek@example.org",
    "CreatedAt": "1736078296932060300"
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "sqlserver",
    "name": "debezium_sql",
    "ts_ms": 1736079110102,
    "snapshot": "last",
    "db": "debezium",
    "sequence": null,
    "schema": "dbo",
    "table": "Customers",
    "change_lsn": null,
    "commit_lsn": "00000037:00000ab8:0005",
    "event_serial_no": null
  },
  "op": "r",
  "ts_ms": 1736079110117,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/80012328-ac10-0ce7-7324-2f3788d9cf2a.png)


#### UPDATE 操作（update events）

次にレコードを UPDATE します。

```sql
UPDATE [dbo].[Customers]
   SET [Email] = N'updated_anne@example.org'
 WHERE [Id] = 1;
GO
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2fd81ec0-bf3b-248f-fe7e-d78e18a1341a.png)

Confluent のトピックを確認すると、UPDATE イベントが以下のように記録されています。

```json
{
  "before": {
    "Id": 1,
    "FirstName": "Anne",
    "LastName": "Kretchmar",
    "Email": "annek@example.org",
    "CreatedAt": "1736078296932060300"
  },
  "after": {
    "Id": 1,
    "FirstName": "Anne",
    "LastName": "Kretchmar",
    "Email": "updated_anne@example.org",
    "CreatedAt": "1736078296932060300"
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "sqlserver",
    "name": "debezium_sql",
    "ts_ms": 1736079342560,
    "snapshot": "false",
    "db": "debezium",
    "sequence": null,
    "schema": "dbo",
    "table": "Customers",
    "change_lsn": "00000037:00000ea0:0005",
    "commit_lsn": "00000037:00000ea0:0007",
    "event_serial_no": 2
  },
  "op": "u",
  "ts_ms": 1736079351813,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a41dd774-5bb8-a301-df40-a6e785c985fc.png)

#### DELETE 操作（delete events）

次にレコードを DELETE します。

```sql
DELETE FROM [dbo].[Customers]
 WHERE [Id] = 1;
GO
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3be8b024-772d-8a95-d04c-001fc19b1baf.png)

Confluent のトピックを確認すると、DELETE イベントが以下のように記録されています。

```json
{
  "before": {
    "Id": 1,
    "FirstName": "Anne",
    "LastName": "Kretchmar",
    "Email": "updated_anne@example.org",
    "CreatedAt": "1736078296932060300"
  },
  "after": null,
  "source": {
    "version": "2.4.2.Final",
    "connector": "sqlserver",
    "name": "debezium_sql",
    "ts_ms": 1736079397723,
    "snapshot": "false",
    "db": "debezium",
    "sequence": null,
    "schema": "dbo",
    "table": "Customers",
    "change_lsn": "00000037:00000f10:0003",
    "commit_lsn": "00000037:00000f10:0007",
    "event_serial_no": 1
  },
  "op": "d",
  "ts_ms": 1736079401631,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5d1a1b9a-65de-8518-0e4a-ede9446defc1.png)

```json
""
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/593ee62a-6684-29d0-e0e8-ec49445bf199.png)

#### Primary Key の値を更新（Primary key updates）

Primary key（id）の値を更新する操作を実施します。

```sql
INSERT INTO [dbo].[Customers] (id, [FirstName], [LastName], [Email])
VALUES
    (1, N'Anne', N'Kretchmar', N'annek@example.org');
GO
```

```sql
UPDATE [dbo].[Customers]
   SET [Id] = 10
 WHERE [Id] = 1;
GO
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f060a2be-b945-6412-26d6-57f17e3620a8.png)

Confluent のトピックでは、id=1 のレコードを削除→id=10 のレコードを挿入する形でイベントが記録されます。

```json
{
  "before": {
    "Id": 1,
    "FirstName": "Anne",
    "LastName": "Kretchmar",
    "Email": "annek@example.org",
    "CreatedAt": "1736079615696179100"
  },
  "after": null,
  "source": {
    "version": "2.4.2.Final",
    "connector": "sqlserver",
    "name": "debezium_sql",
    "ts_ms": 1736079617827,
    "snapshot": "false",
    "db": "debezium",
    "sequence": null,
    "schema": "dbo",
    "table": "Customers",
    "change_lsn": "00000037:00001130:0007",
    "commit_lsn": "00000037:00001130:000b",
    "event_serial_no": 1
  },
  "op": "d",
  "ts_ms": 1736079622630,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/24df3aff-95eb-5375-9043-3d1c17453896.png)


```json
""
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/32a976e0-93b1-781e-5946-4878fbb9dcb1.png)


```json
{
  "before": null,
  "after": {
    "Id": 10,
    "FirstName": "Anne",
    "LastName": "Kretchmar",
    "Email": "annek@example.org",
    "CreatedAt": "1736079615696179100"
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "sqlserver",
    "name": "debezium_sql",
    "ts_ms": 1736079617827,
    "snapshot": "false",
    "db": "debezium",
    "sequence": null,
    "schema": "dbo",
    "table": "Customers",
    "change_lsn": "00000037:00001130:0007",
    "commit_lsn": "00000037:00001130:000b",
    "event_serial_no": 2
  },
  "op": "c",
  "ts_ms": 1736079622630,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/339d6c83-eb2b-a209-a474-8ba6783117e4.png)

#### TRUNCATE 操作（truncate events）

CDC を有効化したテーブルでは、TRUNCATE TABLE を実施できないため省略。

```sql
TRUNCATE TABLE [dbo].[Customers];
GO
```

> Cannot truncate table 'dbo.Customers' because it is published for replication or enabled for Change Data Capture.

### 2. 2 つのテーブルを結合した DML の動作検証

#### 事前準備

例として、`orders`テーブルと`tmp_orders`テーブルを用意します。

```sql
IF OBJECT_ID(N'dbo.orders', N'U') IS NOT NULL
    DROP TABLE [dbo].[orders];
GO

CREATE TABLE [dbo].[orders] (
    [order_id]   BIGINT NOT NULL,
    [customer_id] INT NOT NULL,
    [amount]    DECIMAL(10,2) NULL,
    [created_at] DATETIME2 NOT NULL DEFAULT (SYSUTCDATETIME()),
    CONSTRAINT [PK_Orders] PRIMARY KEY ([order_id])
);
GO

IF OBJECT_ID(N'dbo.tmp_orders', N'U') IS NOT NULL
    DROP TABLE [dbo].[tmp_orders];
GO

CREATE TABLE [dbo].[tmp_orders] (
    [order_id]   BIGINT NOT NULL,
    [customer_id] INT NOT NULL,
    [amount]    DECIMAL(10,2) NULL,
    [created_at] DATETIME2 NOT NULL DEFAULT (SYSUTCDATETIME())
);
GO

EXEC sys.sp_cdc_enable_table
    @source_schema = N'dbo',
    @source_name = N'orders',
    @role_name = NULL;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f6561a60-a42c-962a-470c-380314f89fc2.png)


```sql
INSERT INTO [dbo].[orders] (order_id, customer_id, amount)
VALUES
    (1, 1, 0);

INSERT INTO [dbo].[tmp_orders] (order_id, customer_id, amount)
VALUES
    (1, 1, 100),
    (2, 2, 0);
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/92f98161-6f51-7e78-a53d-3145dc75cdb0.png)


#### DELETE 操作 (JOIN 例)

2 つのテーブルを JOIN してordersテーブルを DELETE します。

```sql
DELETE o
FROM [dbo].[Orders] AS o
JOIN [dbo].[tmp_orders] AS c
ON o.[order_id] = c.[order_id];
GO
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9b924707-197c-39bc-f531-7a4f66247e87.png)

Confluent のトピックを確認すると、DELETE イベントが以下のように記録されています。

```json
{
  "before": {
    "order_id": 1,
    "customer_id": 1,
    "amount": "AA==",
    "created_at": "1736080086476945900"
  },
  "after": null,
  "source": {
    "version": "2.4.2.Final",
    "connector": "sqlserver",
    "name": "debezium_sql",
    "ts_ms": 1736080090250,
    "snapshot": "false",
    "db": "debezium",
    "sequence": null,
    "schema": "dbo",
    "table": "orders",
    "change_lsn": "00000039:00000340:0003",
    "commit_lsn": "00000039:00000340:0006",
    "event_serial_no": 1
  },
  "op": "d",
  "ts_ms": 1736080093632,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/747a7bf1-3be2-8a2e-755f-f17371dfd29f.png)

```json
""
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/529a9bab-ded4-9e54-4dcd-c4513544a592.png)


#### INSERT SELECT 操作

`tmp_orders`テーブルのデータをもとにordersテーブルへ INSERT を実行します。

```sql
INSERT INTO [dbo].[Orders] 
SELECT
    order_id,
    customer_id,
    amount,
    created_at
  FROM [dbo].[tmp_orders];
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3ae56367-d10a-fd82-d1a6-49c527359227.png)


```json
{
  "before": null,
  "after": {
    "order_id": 1,
    "customer_id": 1,
    "amount": "JxA=",
    "created_at": "1736080086476945900"
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "sqlserver",
    "name": "debezium_sql",
    "ts_ms": 1736080195473,
    "snapshot": "false",
    "db": "debezium",
    "sequence": null,
    "schema": "dbo",
    "table": "orders",
    "change_lsn": "00000039:00000400:0005",
    "commit_lsn": "00000039:00000400:0008",
    "event_serial_no": 1
  },
  "op": "c",
  "ts_ms": 1736080199132,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c06c030c-303b-452b-7283-0bc947d3f7c1.png)

#### MERGE 操作

MERGE 操作の動作を確認するためにデータを用意し、`tmp_orders`と`orders`を突き合わせます。

```sql
TRUNCATE TABLE [dbo].[tmp_orders];
INSERT [dbo].[tmp_orders] (order_id, customer_id, amount)
VALUES
    (2, 2, 200),
    (3, 3, 300);
```

```sql
MERGE INTO [dbo].[orders] AS c
USING [dbo].[tmp_orders] AS t
  ON (c.order_id = t.order_id) 
WHEN NOT MATCHED THEN
  INSERT (order_id, customer_id, amount)
  VALUES (t.order_id, t.customer_id, t.amount)
WHEN MATCHED THEN
  UPDATE
     SET customer_id = t.customer_id,
         amount  = t.amount;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a695297d-a632-a386-e371-8dd1d7bb604b.png)

Confluent のトピックでは、INSERT イベントと UPDATE イベントが以下のように記録されています。

```json
{
  "before": {
    "order_id": 2,
    "customer_id": 2,
    "amount": "AA==",
    "created_at": "1736080086476945900"
  },
  "after": {
    "order_id": 2,
    "customer_id": 2,
    "amount": "TiA=",
    "created_at": "1736080086476945900"
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "sqlserver",
    "name": "debezium_sql",
    "ts_ms": 1736081061943,
    "snapshot": "false",
    "db": "debezium",
    "sequence": null,
    "schema": "dbo",
    "table": "orders",
    "change_lsn": "0000003a:000009c0:0007",
    "commit_lsn": "0000003a:000009c0:000a",
    "event_serial_no": 2
  },
  "op": "u",
  "ts_ms": 1736081071634,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/30f1c394-65b7-dc00-3926-5513e027b4d6.png)


```json
{
  "before": null,
  "after": {
    "order_id": 3,
    "customer_id": 3,
    "amount": "dTA=",
    "created_at": 1736081061944533200
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "sqlserver",
    "name": "debezium_sql",
    "ts_ms": 1736081061943,
    "snapshot": "false",
    "db": "debezium",
    "sequence": null,
    "schema": "dbo",
    "table": "orders",
    "change_lsn": "0000003a:000009c0:0008",
    "commit_lsn": "0000003a:000009c0:000a",
    "event_serial_no": 1
  },
  "op": "c",
  "ts_ms": 1736081071634,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e5d6536a-f5dc-b09c-eec2-9f971571cc21.png)


### Primary Key を設定していないテーブルでの動作検証

#### 事前準備

Primary Key を持たないテーブルを作成します。

```sql
IF OBJECT_ID(N'dbo.Logs', N'U') IS NOT NULL
    DROP TABLE [dbo].[Logs];
GO

CREATE TABLE [dbo].[Logs] (
    [LogMessage] NVARCHAR(MAX) NOT NULL,
    [CreatedAt]  DATETIME2 NOT NULL CONSTRAINT DF_Logs_CreatedAt DEFAULT (SYSUTCDATETIME())
);
GO

EXEC sys.sp_cdc_enable_table
    @source_schema = N'dbo',
    @source_name = N'Logs',
    @role_name = NULL;
GO
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ae7974f4-9479-4e5b-2217-076fd7c3c200.png)

#### INSERT 操作（create events）

以下のようにレコードを INSERT します。

```sql
INSERT INTO [dbo].[Logs] ([LogMessage])
VALUES
    (N'System boot');
GO
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b5654266-6794-a761-4969-215e6e794927.png)

Confluent のトピックを確認すると、INSERT イベントが以下のように記録されています。

```json
{
  "before": null,
  "after": {
    "LogMessage": "System boot",
    "CreatedAt": 1736080507210824700
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "sqlserver",
    "name": "debezium_sql",
    "ts_ms": 1736080507220,
    "snapshot": "false",
    "db": "debezium",
    "sequence": null,
    "schema": "dbo",
    "table": "Logs",
    "change_lsn": "0000003a:000000d8:0020",
    "commit_lsn": "0000003a:000000d8:0023",
    "event_serial_no": 1
  },
  "op": "c",
  "ts_ms": 1736080510153,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/16ab6401-d9be-01e1-e78e-b8ae04a5d9ed.png)

#### UPDATE 操作（update events）

```sql
UPDATE [dbo].[Logs]
   SET [LogMessage] = N'[UPDATED] System boot'
 WHERE [LogMessage] = N'System boot';
GO
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ed2f0600-3c34-8b13-97ce-5b29dafbf4ce.png)


```json
{
  "before": {
    "LogMessage": "System boot",
    "CreatedAt": 1736080507210824700
  },
  "after": {
    "LogMessage": "[UPDATED] System boot",
    "CreatedAt": 1736080507210824700
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "sqlserver",
    "name": "debezium_sql",
    "ts_ms": 1736080563660,
    "snapshot": "false",
    "db": "debezium",
    "sequence": null,
    "schema": "dbo",
    "table": "Logs",
    "change_lsn": "0000003a:00000460:0004",
    "commit_lsn": "0000003a:00000460:0006",
    "event_serial_no": 2
  },
  "op": "u",
  "ts_ms": 1736080565135,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2736d337-70e7-8e84-7cce-968f1f3efe85.png)

#### DELETE 操作（delete events）

最後に、該当レコードを削除します。

```sql
DELETE FROM [dbo].[Logs]
 WHERE [LogMessage] = N'[UPDATED] System boot';
GO
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/18288502-e529-a2d1-f452-2af0c467dbc4.png)

Confluent のトピックでは、DELETE イベントが以下のように記録されています。

```json
{
  "before": {
    "LogMessage": "[UPDATED] System boot",
    "CreatedAt": 1736080507210824700
  },
  "after": null,
  "source": {
    "version": "2.4.2.Final",
    "connector": "sqlserver",
    "name": "debezium_sql",
    "ts_ms": 1736080615627,
    "snapshot": "false",
    "db": "debezium",
    "sequence": null,
    "schema": "dbo",
    "table": "Logs",
    "change_lsn": "0000003a:000004c8:0002",
    "commit_lsn": "0000003a:000004c8:0006",
    "event_serial_no": 1
  },
  "op": "d",
  "ts_ms": 1736080620135,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8f4e3ff3-c0c1-13e9-032b-ed6a52e8f9ee.png)


```json
""
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f7c6f430-040f-aec7-f3d2-e368290a5f94.png)

まとめ
以上が、Azure Database から Confluent（Debezium）を用いた CDC の手順と、各種 DML 操作のイベントがどのようにトピックへ記録されるかの例となります。PK（Primary Key）の有無や、複数テーブルにまたがる DML などのパターンによって、Kafka に取り込まれるイベントの内容が異なる点にご注意ください。