Posted inデータベース

DebeziumとApache KafkaによるChange Data Capture：PostgreSQLの変更をリアルタイムにストリーミング

Database tutorial - IT technology blog
Database tutorial - IT technology blog

現実の課題：ダウンストリームシステムとの同期維持

本番環境のデータベースが単独で動作することは稀です。標準的なチェックアウトフローを考えてみましょう。ユーザーが1回クリックするだけで、PostgreSQLに注文レコードが書き込まれ、Redisのキャッシュセッションが無効化され、Elasticsearchのカタログが再インデックスされ、StripeへのWebhookがトリガーされます。

チームは通常、アプリケーションコード内で直接二重書き込みを行うか、SELECT * FROM orders WHERE updated_at > :last_poll_time のような定期的なクエリを30秒ごとに実行する2つのシンプルなパターンから始めます。初期のプロトタイピング段階ではこれでも機能しますが、トラフィックが毎秒数百件の書き込みに拡大すると、急速に破綻します。

  • サイレントなデータの不整合（Data Drift）： アプリケーションワーカーがPostgreSQLへのコミット直後、Elasticsearchを呼び出す前にクラッシュした場合、検索インデックスは古いままになります。手動のデータ修復スクリプトを実行しない限り、その不整合は永続的に残ります。
  • データベースI/Oの飽和： 5〜10秒ごとにポーリングを行うと、PostgreSQLは大規模なインデックス範囲を繰り返しスキャンせざるを得なくなります。テーブルが1,000万行に達すると、これらのクエリがコアなトランザクション処理に必要なディスクスループットとCPUサイクルを消費してしまいます。
  • 物理削除（Hard Delete）の見落とし： updated_at でフィルタリングするポーリングクエリでは、物理削除された行（DELETE FROM orders WHERE id = 42）を決して検出できません。全テーブルに煩雑な論理削除フラグを導入するか、削除追跡用の専用テーブルを別途構築するしかなくなります。

根本原因の分析：なぜ従来の同期戦略は破綻するのか

こうした同期の失敗は、2つのアーキテクチャ上の制約に起因しています。それは、分散アトミックトランザクションの欠如と、PostgreSQL内部のエンジンログを完全にバイパスしていることです。

1. 二重書き込みの競合状態（Race Condition）

同一のリクエストハンドラ内でプライマリデータベースとセカンダリキューの両方に書き込むと、スプリットブレインのリスクが必然的に生じます。データベースのコミットとキューへのパブリッシュの間でネットワークが切断されると、イベントは失われます。安易にリトライを行えば、メッセージの重複や順序の逆転が発生するリスクがあります。理論上は2フェーズコミット（2PC）プロトコルで調整可能ですが、レイテンシが大幅に増加し（1トランザクションあたり50〜100msかかることも多い）、ネットワーク分断時に予測不能な障害を引き起こします。

2. ポーリングによる中間状態の変化の欠落

注文ステータスが2秒の間に PENDING から PROCESSING を経て SHIPPED へ遷移する場合を想定してください。ポーリングクエリが10秒ごとにしか実行されない場合、中間の PROCESSING 状態は完全に消え去ってしまいます。これにより、分析パイプラインや不正検知サービスは重要な状態遷移を見落とし、不完全な監査履歴しか残せなくなります。

データ同期アプローチの比較

本番環境における4つの代表的な変更伝播パターンの比較は以下のとおりです。

アプローチ レイテンシ データベース負荷 削除の処理 信頼性
アプリケーションの二重書き込み 低（<50ms） 低 可能 低い（データの不整合や部分障害が発生しやすい）
定期的なポーリング（ETL） 高（10秒〜数時間） 高（CPUとディスクI/Oを継続的に消費） 不可（論理削除が必要） 中程度（中間状態を見落とす）
トリガーベースのテーブル 低（<100ms） 高（書き込みの増幅が2倍） 可能 良好（スキーマの保守が複雑）
ログベースのCDC（Debezium + Kafka） 1秒未満（<10ms） ほぼゼロ（シーケンシャルなWALを読み取り） 可能 極めて高い（厳格な順序保証と原子性）

ログベースのChange Data Capture（CDC）は、SQLレイヤーを完全にバイパスします。PostgreSQLはディスクへのページ書き出しを行う前に、すべての挿入・更新・コミットを先行書き込みログ（WAL: Write-Ahead Log）へ順次書き込みます。CDCエンジンはこのログを直接読み取ります。これにより、ミリ秒未満の変更通知、実テーブルデータへのクエリ負荷ゼロ、厳格なトランザクション順序の保証、そしてすべての行操作に対する100%の再現性を実現できます。

テストデータセットの初期構築時や、ダウンストリームコンシューマー向けの初期シードファイルを準備する際、フォーマットの不一致がよく発生します。データインポートのためにCSVをJSONへ素早く変換する必要がある場合は、toolcraft.app/ja/tools/data/csv-to-json を利用しています。ブラウザ上で実行されるため、データがマシン外に送信される心配がありません。スキーマのベースラインが整ったら、リアルタイムパイプラインのセットアップに進みましょう。

DebeziumとKafkaによるログベースCDCの実装

Debeziumは、標準のレプリケーションクライアントとして、組み込みの pgoutput ロジカルデコーディングプラグインを使用してPostgreSQLに接続します。バイナリ形式のWAL変更を読み取り、行の値を構造化イベントにデシリアライズして、専用のApache Kafkaトピックにパブリッシュします。

ステップ1：論理レプリケーション用にPostgreSQLを設定する

PostgreSQLで行レベルの変更詳細をWALファイルに書き込ませる必要があります。postgresql.conf ファイルで以下のパラメータを設定します。

wal_level = logical
max_wal_senders = 4
max_replication_slots = 4

専用のレプリケーションユーザーとサンプルテーブルを作成します。

CREATE USER cdc_user WITH REPLICATION PASSWORD 'secret_cdc_pass';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO cdc_user;

CREATE TABLE customers (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- UPDATEおよびDELETE操作時に変更前の行状態全体をキャプチャする
ALTER TABLE customers REPLICA IDENTITY FULL;

ステップ2：Docker Composeで環境を立ち上げる

PostgreSQL 15、Apache Kafka（ZooKeeper不要のKRaftモード）、およびDebezium Kafka Connectを実行する完全な docker-compose.yml ファイルは以下のとおりです。

version: '3.8'
services:
  postgres:
    image: postgres:15-alpine
    environment:
      POSTGRES_DB: appdb
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: rootpassword
    command: ["postgres", "-c", "wal_level=logical"]
    ports:
      - "5432:5432"

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092'
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:29093'
      KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:29092,CONTROLLER://0.0.0.0:29093,PLAINTEXT_HOST://0.0.0.0:9092'
      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk'
    ports:
      - "9092:9092"

  connect:
    image: debezium/connect:2.4
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:29092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: my_connect_configs
      OFFSET_STORAGE_TOPIC: my_connect_offsets
      STATUS_STORAGE_TOPIC: my_connect_statuses
    depends_on:
      - kafka
      - postgres

ステップ3：Debezium PostgreSQLコネクタを登録する

ポート8083のKafka Connect RESTインターフェースにコネクタ設定をPOSTします。

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
  http://localhost:8083/connectors/ -d '{
  "name": "postgres-cdc-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "plugin.name": "pgoutput",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "rootpassword",
    "database.dbname": "appdb",
    "topic.prefix": "pg_cdc",
    "table.include.list": "public.customers"
  }
}'

ステップ4：変更ストリームを検証する

PostgreSQLに新しい行を挿入します。

INSERT INTO customers (name, email) VALUES ('Alex Rivers', '[email protected]');

Kafkaトピック pg_cdc.public.customers から直接生成されたメッセージをコンシューム（受信）します。

docker exec -it $(docker ps -qf "name=kafka") kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic pg_cdc.public.customers \
  --from-beginning

Debeziumは、トランザクションメタデータとともに変更前後の状態を詳細に記述した包括的なエンベロープを出力します。

{
  "before": null,
  "after": {
    "id": 1,
    "name": "Alex Rivers",
    "email": "[email protected]",
    "updated_at": 1700000000000
  },
  "source": {
    "version": "2.4.0.Final",
    "connector": "postgresql",
    "name": "pg_cdc",
    "ts_ms": 1700000000123,
    "db": "appdb",
    "schema": "public",
    "table": "customers",
    "txId": 582
  },
  "op": "c",
  "ts_ms": 1700000000456
}

op フィールドはイベントタイプを明確に示します。c は作成（挿入）、u は更新、d は削除、r は初期スナップショット読み取りを表します。ダウンストリームのコンシューマーは、id をキーにしつつ source.txId を確認することで、冪等性を担保できます。

本番環境における重要な考慮事項

  • レプリケーションスロットの肥大化： 書き込みトラフィックが継続している状態（例: 2,000 writes/sec）でDebeziumが数時間切断されると、PostgreSQLは未消費のWALログをディスク上に保持し続けます。ディスク枯渇によるシステム停止を防ぐため、必ず max_slot_wal_keep_size（例: 10GB）を設定し、pg_replication_slots.active を監視してください。
  • ツームストーン（Tombstone）イベント： 削除操作の後、Debeziumはペイロードがnullのレコードを発行します。これにより、Kafkaのログクリーナーがコンパクション時に古いレコードキーを削除できるようになります。ダウンストリームのコンシューマーが NullPointerException などの例外をスローせず、nullペイロードを適切に処理できることを確認してください。
  • スキーママイグレーション： ALTER TABLE を使ったNULL許容カラムの追加は、コネクタを再起動することなくスムーズにストリーミングされます。ただし、スキーマの進化に柔軟に対応できるよう、ダウンストリームコンシューマー側でJSONペイロードを防御的にパースする実装が必要です。
Share:
Tags: