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

## 概要

本記事では、Azure Database for PostgreSQL をソースとして、Confluent で提供される Debezium を用いて CDC（Change Data Capture）を実行する手順と、その実行結果についてご紹介します。Azure Database for PostgreSQL に対する 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 for PostgreSQL の環境構築

#### 1. Azure Database for PostgreSQL Flexible Server の作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/97367204-d38c-9519-14f5-fa501741e712.png)

#### 2. `wal_level` を `LOGICAL` に設定

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2be20533-7a49-4c6d-aa35-7c6b2cd1bfc3.png)

#### 3. スキーマ作成

```sql
DROP SCHEMA IF EXISTS debezium_test CASCADE;
CREATE SCHEMA debezium_test;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1ec1d34d-3dd7-56c6-36b8-44c3eee3e9ef.png)

#### 4. PUBLICATION の作成

```sql
CREATE PUBLICATION debezium_publication FOR ALL TABLES;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fd34c093-2cef-3d39-49fd-68f323dd2482.png)

#### 5. `REPLICATION` 権限の付与

```sql
ALTER ROLE debezium WITH REPLICATION;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2e25bb45-0141-6cd4-72a7-79d91b442025.png)

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

以下の画面例のように、Confluent Cloud で Debezium コネクタを作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8d451162-61a5-26c6-4693-b08b44b5a2ea.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8530404e-d0b2-c8e3-b6c9-85d265ebeb18.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d6add875-b002-a2c1-1cd6-d18e63e9016c.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d32d0c70-c16b-a034-7993-938362ff9dd5.png)

---

## 動作検証

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

#### 事前準備

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

```sql
-- テーブル作成
CREATE TABLE debezium_test.customers (
    id         BIGINT PRIMARY KEY,
    first_name VARCHAR(255) NOT NULL,
    last_name  VARCHAR(255) NOT NULL,
    email      VARCHAR(255) NOT NULL,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0643ef1a-727d-1781-9f8b-6681e3af9665.png)

#### INSERT 操作（create events）

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

```sql
-- 通常の INSERT
INSERT INTO debezium_test.customers (id, first_name, last_name, email)
VALUES
    (1, 'Anne', 'Kretchmar', 'annek@example.org');
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/e4cbf741-91c0-6467-df27-ed786a069a07.png)

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

```json
{
  "before": null,
  "after": {
    "id": 1,
    "first_name": "Anne",
    "last_name": "Kretchmar",
    "email": "annek@example.org",
    "created_at": "2025-01-05T06:49:37.681496Z"
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "postgresql",
    "name": "debezium",
    "ts_ms": 1736059777682,
    "snapshot": "false",
    "db": "postgres",
    "sequence": "[null,\"67376224\"]",
    "schema": "debezium_test",
    "table": "customers",
    "txId": 1060,
    "lsn": 67376224,
    "xmin": null
  },
  "op": "c",
  "ts_ms": 1736059777909,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c6917261-8bd8-cf1a-ec31-7a3b4dd2490b.png)

#### UPDATE 操作（update events）

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

```sql
-- 既存レコードの更新
UPDATE debezium_test.customers
   SET email = 'updated_anne@example.org'
 WHERE id = 1;
```

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

```json
{
  "before": null,
  "after": {
    "id": 1,
    "first_name": "Anne",
    "last_name": "Kretchmar",
    "email": "updated_anne@example.org",
    "created_at": "2025-01-05T06:49:37.681496Z"
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "postgresql",
    "name": "debezium",
    "ts_ms": 1736059837649,
    "snapshot": "false",
    "db": "postgres",
    "sequence": "[\"67376544\",\"83896440\"]",
    "schema": "debezium_test",
    "table": "customers",
    "txId": 1075,
    "lsn": 83896440,
    "xmin": null
  },
  "op": "u",
  "ts_ms": 1736059837776,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a40b63da-942c-d277-642c-f29a64fb5b5f.png)

#### DELETE 操作（delete events）

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

```sql
-- レコード削除
DELETE FROM debezium_test.customers
 WHERE id = 1;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f2fe5a6d-546c-a3f3-235d-4f8e848f371d.png)

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

```json
{
  "before": {
    "id": 1,
    "first_name": "",
    "last_name": "",
    "email": "",
    "created_at": "1970-01-01T00:00:00.000000Z"
  },
  "after": null,
  "source": {
    "version": "2.4.2.Final",
    "connector": "postgresql",
    "name": "debezium",
    "ts_ms": 1736059892606,
    "snapshot": "false",
    "db": "postgres",
    "sequence": "[\"83896616\",\"83899008\"]",
    "schema": "debezium_test",
    "table": "customers",
    "txId": 1089,
    "lsn": 83899008,
    "xmin": null
  },
  "op": "d",
  "ts_ms": 1736059893043,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5b4c35f6-7557-4573-eb4a-2a1e00f1dd92.png)

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

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

```sql
-- 通常の INSERT
INSERT INTO debezium_test.customers (id, first_name, last_name, email)
VALUES
    (1, 'Anne', 'Kretchmar', 'annek@example.org');

UPDATE debezium_test.customers
   SET id = 10
 WHERE id = 1;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/484fba96-66af-67e3-9745-c41e749ef9fa.png)

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

```json
{
  "before": {
    "id": 1,
    "first_name": "",
    "last_name": "",
    "email": "",
    "created_at": "1970-01-01T00:00:00.000000Z"
  },
  "after": null,
  "source": {
    "version": "2.4.2.Final",
    "connector": "postgresql",
    "name": "debezium",
    "ts_ms": 1736059959049,
    "snapshot": "false",
    "db": "postgres",
    "sequence": "[\"83902280\",\"83902280\"]",
    "schema": "debezium_test",
    "table": "customers",
    "txId": 1106,
    "lsn": 83902280,
    "xmin": null
  },
  "op": "d",
  "ts_ms": 1736059959282,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/cad791ad-31f6-afbf-2002-0565a02b7ec5.png)


```json
""
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6a7a3058-6b0e-a9d0-572a-742450db979a.png)

```json
{
  "before": null,
  "after": {
    "id": 10,
    "first_name": "Anne",
    "last_name": "Kretchmar",
    "email": "annek@example.org",
    "created_at": "2025-01-05T06:52:36.387167Z"
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "postgresql",
    "name": "debezium",
    "ts_ms": 1736059959049,
    "snapshot": "false",
    "db": "postgres",
    "sequence": "[\"83902280\",\"83902280\"]",
    "schema": "debezium_test",
    "table": "customers",
    "txId": 1106,
    "lsn": 83902280,
    "xmin": null
  },
  "op": "c",
  "ts_ms": 1736059959282,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d2d8125b-98aa-171a-652f-5e17dedb9b35.png)


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

#### 事前準備

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

```sql
-- 例: orders テーブル
CREATE TABLE debezium_test.orders (
    order_id    BIGINT PRIMARY KEY,
    customer_id BIGINT NOT NULL,
    amount      BIGINT,
    created_at  TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE debezium_test.tmp_orders (
    order_id    BIGINT PRIMARY KEY,
    customer_id BIGINT NOT NULL,
    amount      BIGINT,
    created_at  TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c146c45b-0886-39db-e3b1-557fad7ed489.png)

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

INSERT INTO debezium_test.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/8565770f-d584-f82b-1602-a0db0dc2ee71.png)

#### DELETE 操作 (JOIN 例)

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

```sql
DELETE FROM debezium_test.orders o
 USING debezium_test.tmp_orders c
 WHERE o.order_id = c.order_id;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/45303bde-d85b-15b5-d664-237930fe2083.png)

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


```json
{
  "before": {
    "order_id": 1,
    "customer_id": 0,
    "amount": null,
    "created_at": "1970-01-01T00:00:00.000000Z"
  },
  "after": null,
  "source": {
    "version": "2.4.2.Final",
    "connector": "postgresql",
    "name": "debezium",
    "ts_ms": 1736060074117,
    "snapshot": "false",
    "db": "postgres",
    "sequence": "[\"83997784\",\"83998040\"]",
    "schema": "debezium_test",
    "table": "orders",
    "txId": 1137,
    "lsn": 83998040,
    "xmin": null
  },
  "op": "d",
  "ts_ms": 1736060074330,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f4f8e0e9-e746-5fc1-1274-0c70919f43f1.png)

```json
""
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5a16c1a3-33e7-6480-c700-c7d037635800.png)

#### INSERT SELECT 操作

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

```sql
INSERT INTO debezium_test.orders
SELECT
    order_id,
    customer_id,
    amount,
    created_at
  FROM debezium_test.tmp_orders;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c7efc36f-5eb0-0c80-0e16-88323bee568d.png)

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

```json
{
  "before": null,
  "after": {
    "order_id": 1,
    "customer_id": 1,
    "amount": 100,
    "created_at": "2025-01-05T06:54:29.435440Z"
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "postgresql",
    "name": "debezium",
    "ts_ms": 1736060196651,
    "snapshot": "false",
    "db": "postgres",
    "sequence": "[\"83998160\",\"100667016\"]",
    "schema": "debezium_test",
    "table": "orders",
    "txId": 1167,
    "lsn": 100667016,
    "xmin": null
  },
  "op": "c",
  "ts_ms": 1736060196778,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8edf9162-20e4-342c-6923-44ab91a8b2fb.png)

#### MERGE 操作

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

```sql
TRUNCATE TABLE debezium_test.tmp_orders;
INSERT INTO debezium_test.tmp_orders (order_id, customer_id, amount)
VALUES
    (2, 2, 200),
    (3, 3, 300);
```

```sql
MERGE INTO debezium_test.orders AS target
USING debezium_test.tmp_orders AS source
ON (target.customer_id = source.customer_id)
WHEN MATCHED THEN
  UPDATE SET
    amount = source.amount,
    customer_id = source.customer_id,
    created_at = CURRENT_TIMESTAMP
WHEN NOT MATCHED THEN
  INSERT (order_id, customer_id, amount)
  VALUES (source.order_id, source.customer_id, source.amount);
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5b9c2efc-218b-d891-4495-5ca95287102e.png)

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

```json
{
  "before": null,
  "after": {
    "order_id": 3,
    "customer_id": 3,
    "amount": 300,
    "created_at": "2025-01-05T06:58:02.160738Z"
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "postgresql",
    "name": "debezium",
    "ts_ms": 1736060282161,
    "snapshot": "false",
    "db": "postgres",
    "sequence": "[\"100674592\",\"100674696\"]",
    "schema": "debezium_test",
    "table": "orders",
    "txId": 1190,
    "lsn": 100674696,
    "xmin": null
  },
  "op": "c",
  "ts_ms": 1736060282378,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/98d7a6e6-e5ba-51ce-125c-c2ef16e3daaa.png)


```json
{
  "before": null,
  "after": {
    "order_id": 2,
    "customer_id": 2,
    "amount": 200,
    "created_at": "2025-01-05T06:58:02.160738Z"
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "postgresql",
    "name": "debezium",
    "ts_ms": 1736060282161,
    "snapshot": "false",
    "db": "postgres",
    "sequence": "[\"100674592\",\"100674592\"]",
    "schema": "debezium_test",
    "table": "orders",
    "txId": 1190,
    "lsn": 100674592,
    "xmin": null
  },
  "op": "u",
  "ts_ms": 1736060282378,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5372c635-f796-0173-6ff4-c08cbea66fa0.png)

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

#### 事前準備

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

```sql
-- PK なしテーブル
CREATE TABLE debezium_test.logs (
    log_message TEXT,
    created_at  TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
);
```

PK が設定されていないテーブルに対しては、`UPDATE` を実行しようとすると以下のエラーが発生する場合があります。

```text
cannot update table "logs" because it does not have a replica identity and publishes updates
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dd7d23f9-0a54-7993-e484-51ba60a5b12c.png)

このエラーを回避するには、`REPLICA IDENTITY` を `FULL` に設定します。ただし、INSERT 操作のみであれば必須ではありません。

```sql
ALTER TABLE debezium_test.logs REPLICA IDENTITY FULL;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fd43c246-4691-2ebd-5930-c081b2ec7c34.png)

#### INSERT 操作（create events）

```sql
INSERT INTO debezium_test.logs (log_message)
VALUES
   ('System boot');
```

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

```json
{
  "before": null,
  "after": {
    "log_message": "System boot",
    "created_at": "2025-01-05T07:00:54.979111Z"
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "postgresql",
    "name": "debezium",
    "ts_ms": 1736060454979,
    "snapshot": "false",
    "db": "postgres",
    "sequence": "[\"100674896\",\"117448792\"]",
    "schema": "debezium_test",
    "table": "logs",
    "txId": 1232,
    "lsn": 117448792,
    "xmin": null
  },
  "op": "c",
  "ts_ms": 1736060455160,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/99b4c209-0e92-fa8b-ac33-1fb6d790d067.png)

#### UPDATE 操作（update events）

Primary Key が無いテーブルでは、以下のようにレコードを特定して UPDATE します。

```sql
UPDATE debezium_test.logs
   SET log_message = '[UPDATED] System boot'
 WHERE log_message = 'System boot';
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0948ea03-1f38-96b4-7b42-bbf198a61e07.png)

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

```json
{
  "before": {
    "log_message": "System boot",
    "created_at": "2025-01-05T07:00:54.979111Z"
  },
  "after": {
    "log_message": "[UPDATED] System boot",
    "created_at": "2025-01-05T07:00:54.979111Z"
  },
  "source": {
    "version": "2.4.2.Final",
    "connector": "postgresql",
    "name": "debezium",
    "ts_ms": 1736060585737,
    "snapshot": "false",
    "db": "postgres",
    "sequence": "[\"117448920\",\"117456544\"]",
    "schema": "debezium_test",
    "table": "logs",
    "txId": 1263,
    "lsn": 117456544,
    "xmin": null
  },
  "op": "u",
  "ts_ms": 1736060585803,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2bdd6ce5-6389-5fb9-d1ff-fb4bf4d39096.png)

#### DELETE 操作（delete events）

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

```sql
DELETE FROM debezium_test.logs
 WHERE log_message = '[UPDATED] System boot';
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4a80825e-f3e8-c775-495b-a7bf9267be39.png)

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

```json
{
  "before": {
    "log_message": "[UPDATED] System boot",
    "created_at": "2025-01-05T07:00:54.979111Z"
  },
  "after": null,
  "source": {
    "version": "2.4.2.Final",
    "connector": "postgresql",
    "name": "debezium",
    "ts_ms": 1736060654552,
    "snapshot": "false",
    "db": "postgres",
    "sequence": "[\"117456720\",\"117460176\"]",
    "schema": "debezium_test",
    "table": "logs",
    "txId": 1280,
    "lsn": 117460176,
    "xmin": null
  },
  "op": "d",
  "ts_ms": 1736060654671,
  "transaction": null
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9d89ed6c-c55c-1e4e-e028-d60a93410beb.png)

## まとめ

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