変更データキャプチャ（CDC）のアーキテクチャ：Debezium と Kafka によるログベース CDC

注文テーブルは Postgres にあり、今やほかの 5 か所がそれを必要としています。検索インデックス、キャッシュ、データウェアハウス、不正検知サービス、レポート用の読み取りモデルです。アプリケーションからすべてに書き込む方法は、遅かれ早かれ途中で失敗します。テーブルを毎分ポーリングすれば削除を見逃し、負荷も増えます。ログベースの変更データキャプチャ（CDC）は別の道を取ります。データベース自身のトランザクションログ、つまりクラッシュリカバリーやレプリケーションに使われるのと同じ記録を読み、コミットされた変更を 1 つずつイベントに変えるのです。

この記事では、Debezium と Kafka によるログベースの CDC を、アーキテクチャ図とともに説明します。各手法の比較、パイプラインの仕組み、変更イベントの中身、スナップショット、順序と配信保証、スキーマ変更、プライマリを停止させる運用上の落とし穴、そして代わりにアウトボックスパターンを使うべき場面です。

変更を捕捉する 4 つの方法

ログベースの CDC は、削除を含むすべてのコミット済みの変更を、アプリケーションに手を入れず、書き込みを遅くすることもなく、コミット順に捉えられる唯一の手法です。この記事の残りはこの手法について扱います。

ログベースの CDC の仕組み

データベースは、変更を適用する前に必ずログに書き込んでいます。Postgres の write-ahead log、MySQL のバイナリログ、SQL Server のトランザクションログ、MongoDB の変更ストリームの裏にある oplog です。CDC コネクターは、このログをレプリカと同じように読みます。

Postgres では、これは論理デコーディングを意味します。

-- postgresql.conf: wal_level = logical (requires a restart)
CREATE PUBLICATION orders_pub FOR TABLE orders, order_items;

-- Include all old column values in update and delete events
ALTER TABLE orders REPLICA IDENTITY FULL;

コネクターは、どこまで読んだかを記憶するレプリケーションスロットを作成し、Postgres に組み込まれた pgoutput プラグインで変更をデコードします。既定では、更新と削除のイベントには古い主キーしか含まれません。REPLICA IDENTITY FULL を使うと、ログが大きくなる代わりに古い行全体が加わります。MySQL での相当する設定は、現行バージョンの既定である行ベースのバイナリログで、行の完全なイメージを記録します。

Kafka Connect 上の Debezium

Debezium は広く使われているオープンソースの CDC エンジンです。そのコネクターは Kafka Connect の中で動き、ログを読み、行の変更ごとに 1 つのイベントを、テーブルごとの Kafka トピックに、行の主キーをキーとして発行します。最小限の Postgres コネクターの設定は次のとおりです。

{
  "name": "orders-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "orders-db",
    "database.port": "5432",
    "database.user": "debezium",
    "database.dbname": "orders",
    "topic.prefix": "shop",
    "table.include.list": "public.orders,public.order_items",
    "publication.name": "orders_pub",
    "slot.name": "orders_cdc",
    "heartbeat.interval.ms": "10000"
  }
}

public.orders の変更は、トピック shop.public.orders に届きます。パスワードはシークレットプロバイダーを使ってファイルの外に置いてください。Kafka を運用していない場合でも、Debezium Server を使えば同じイベントをほかのブローカーやクラウドのキューに流せます。

変更イベントの構造

ペイロードだけに絞ると、更新イベントは次のようになります。

{
  "before": { "id": 1042, "status": "pending", "total": 59.90 },
  "after":  { "id": 1042, "status": "paid",    "total": 59.90 },
  "source": { "db": "orders", "table": "orders", "lsn": 24023128, "txId": 5521 },
  "op": "u",
  "ts_ms": 1793606400123
}

op フィールドは、作成なら c、更新なら u、削除なら d、スナップショット中に読み取った行なら r です。削除のあとには tombstone が続きます。同じキーで値が null のメッセージで、Kafka のログコンパクションが最終的にそのキーを完全に削除できるようにします。source ブロックは変更がログのどこから来たのかを正確に記録しており、これが下流での重複排除を可能にします。

スナップショット：すでに存在する行

新しいコネクターは、すでに数百万行あるテーブルから始まりますが、ログには最近の変更しかありません。そこでコネクターはまず初期スナップショットを取り、既存の行をすべて r イベントとして出力し、スナップショット開始時に記録したログ位置からストリーミングを始めます。こうして、すき間に落ちる変更はなくなります。

大きなテーブルでは、長時間ブロックするスナップショットは苦痛です。インクリメンタルスナップショットは、ストリーミングを続けたままテーブルをチャンクごとにコピーします。Debezium チームの解説にあるとおり、Netflix の DBLog 論文のウォーターマーク手法を使っています。シグナルテーブルを通じて起動し、再起動後も最初からやり直さずに続きから再開します。

順序と配信保証

• 行単位では、イベントの順序が保たれます。 同じ主キーのイベントはすべて同じキーを持つので、同じ Kafka パーティションに入ります。
• 行やテーブルをまたいだグローバルな順序はありません。 1 つのトランザクションで変更された 2 つのテーブルは、2 つのトピックのイベントになります。トランザクションの境界が必要なら、Debezium はトランザクションのメタデータイベントを出力でき、コンシューマーはそれを使ってイベントをまとめ直せます。
• 配信は at-least-once です。 クラッシュや再起動のあと、コネクターは一部のイベントを再送することがあります。Debezium のドキュメントもはっきりそう述べています。

コンシューマーにとっての結論は、冪等にすることです。キーで upsert し、そのキーで最後に適用したものより新しくないログ位置のイベントは無視します。

スキーマ変更

テーブルを捕捉し始めた時点で、そのスキーマはすべてのコンシューマーとの契約になります。null 許容の列を追加するのは安全ですが、列名の変更や型の変更は誰かを壊します。Avro、Protobuf、JSON Schema とスキーマレジストリを使い、互換性ルールを強制してください。そうすれば、破壊的な変更は本番ではなく CI で失敗します。さらに良いのは、どのテーブルを公開するかを決めることです。サービスの外で誰も消費していなければ、内部テーブルは自由に変更できます。

レプリケーションスロットの落とし穴

これは、ほとんどのチームが痛い目にあって学ぶ運用上の教訓です。Postgres のレプリケーションスロットは、コネクターが確認していない WAL セグメントをすべて保持します。コネクターが停止したり大きく遅れたりすると、ディスクがいっぱいになってデータベースが書き込みを受け付けなくなるまで、プライマリに WAL が溜まり続けます。max_slot_wal_keep_size の Postgres の既定値は -1 で、スロットが保持できる WAL の量に上限がないことを意味します。

同じ問題の、もっと静かな形もあります。捕捉対象のテーブルが静かで、ほかのテーブルが忙しい場合です。コネクターには確認するものがないので、スロットは一向に進みません。Debezium のハートビート設定は、まさにこのケースのためにあります。

対策は次のとおりです。

-- How much WAL each slot is holding back
SELECT slot_name, active,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS retained
FROM pg_replication_slots;

• ディスク使用率だけでなく、スロットごとに保持されている WAL にアラートを設定します。
• max_slot_wal_keep_size は、ディスクが耐えられるサイズに設定します。上限を超えたスロットは無効化され、コネクターはスナップショットを取り直す必要がありますが、ディスクがいっぱいになるよりはるかにましです。
• ハートビートを有効にし、ハートビート用のテーブルをパブリケーションに追加します。
• コネクターを廃止するときは、スロットも削除します。
• フェイルオーバーを計画します。PostgreSQL 17 は論理スロットを物理スタンバイに同期できます。自分の構成とコネクターのバージョンがこれを使っているか確認してください。

CDC か、アウトボックスパターンか

テーブルを直接捕捉すると、内部のスキーマがイベントとして外部に公開されます。代わりの方法がアウトボックスパターンです。サービスはデータと同じトランザクションの中で、明示的なイベントをアウトボックステーブルに書き込み、CDC はそのテーブルだけを発行します。Debezium にはこのための outbox event router が含まれています。

右の列については、トランザクショナルアウトボックスパターンの記事で詳しく扱っています。

CDC パイプラインの図

図の読み方は次のとおりです。

• 注文サービスは、いつもどおり Postgres に書き込みます。CDC のことは何も知りません。
• Kafka Connect フレーム内の Debezium は、レプリケーションスロットを通じて write-ahead log を読み、変更イベントを shop.public.orders トピックに発行し、スキーマをレジストリに登録します。
• 4 つのコンシューマーが、それぞれ独立してトピックを読みます。検索インデックス、Redis キャッシュ、データウェアハウス、読み取りモデルのサービスです。
• Postgres に戻る破線のエッジはスロットの確認です。WAL が溜まるのを防ぐつながりです。
• Markdown ブロックには、要約した変更イベントとスロットの遅延を調べるクエリを載せています。

CDC パイプラインを描くには

• データベースだけでなく、ログも描きます。 CDC が成功するか失敗するかは、スロットと WAL で決まります。
• トピック名とキーを示します。 「注文 ID をキーにする」と書けば、順序の説明が一言で済みます。
• コンシューマーは 1 つずつ別に描きます。 それぞれに固有の遅延、障害モード、冪等性のロジックがあります。
• 配信保証を図に書きます。 at-least-once です。

Kafka 側については Kafka アーキテクチャ図で扱っています。CDC は、データパイプラインのアーキテクチャ図で示したレイクハウスの生データ層にデータを送り込む一般的な方法であり、CQRS とイベントソーシングの図の読み取りモデルを構築する方法でもあります。

Archboard で作る

1. Google または GitHub でサインインし、図を作成します。
2. 注文サービス用にクラウドパックの コンピューティング ノードを、さらにデータパックの Postgres ノードを追加します。
3. Kafka Connect 用に基本パックの フレーム を追加し、その中に Debezium コネクター用の コンピューティング ノードを置きます。
4. トピック用にデータパックの Kafka ノードを、スキーマレジストリ用に基本パックの ボックス を追加します。
5. コンシューマーを追加します。検索インデックス用にクラウドパックの マネージドデータベース、Redis ノード、データウェアハウス ノード、読み取りモデルのサービス用に コンピューティング ノードです。
6. スロットの確認を表す破線のエッジをコネクターから Postgres へ引き、トピックにキーのラベルを付け、イベントの例を書いた基本パックの Markdown ブロックを追加します。

この例で使うダイアグラムノードは 11 個で、Free プランの範囲に収まります。

よくある質問

ログベースの変更データキャプチャとは何ですか？ Postgres の write-ahead log や MySQL のバイナリログのようなデータベースのトランザクションログを読み、コミットされた変更を 1 つずつイベントに変える仕組みです。アプリケーションを変更せずに、挿入、更新、削除を順序どおりに捕捉します。

Debezium は exactly-once の配信を保証しますか？ いいえ。通常の運用では各変更は 1 回だけ配信されますが、障害のあとには一部のイベントが再配信されることがあるため、Debezium は at-least-once の配信を明記しています。コンシューマーは冪等にすべきです。

CDC を設定したあと、Postgres のディスクがいっぱいになったのはなぜですか？ レプリケーションスロットは、コネクターが確認するまで WAL を保持します。コネクターが停止していたり、ほかのテーブルが変わる一方で捕捉対象のテーブルが静かなままだったりすると、WAL が溜まります。スロットの遅延を監視し、ハートビートを有効にし、max_slot_wal_keep_size を設定してください。

CDC は既存のデータをどう扱いますか？ スナップショットを使います。コネクターはまず既存の行をすべて出力し、そのあと開始時に記録したログ位置から変更をストリーミングします。インクリメンタルスナップショットは、ストリーミングを続けながらこれをチャンク単位で行います。

CDC とアウトボックスパターンのどちらを使うべきですか？ テーブル CDC はレプリケーション、検索、キャッシュ、分析に向いています。サービス間のイベントには、内部テーブルの行ではなく設計された契約を発行するアウトボックスパターンが向いています。

無料で始める — 1 シート、クレジットカード不要。