背景：Dual-writeと定期実行バッチが破綻を招く理由
毎分2,000件の注文が発生するECサイトを運用している場面を想像してみてください。PostgreSQLは信頼できる単一の情報源（Source of Truth）として機能し、ACID特性に準拠して各トランザクションを確実に記録します。同時にプロダクトチームは、商品検索や注文履歴の検索をレスポンス50ms未満でユーザーに提供するため、Elasticsearchを必要としています。
この課題に対して真っ先に思い浮かびがちなのがDual-write（二重書き込み）です。アプリケーションがPostgreSQLへの書き込みを完了した後、続けてElasticsearchのAPIを呼び出して反映する構成です。一見シンプルに思えますが、ネットワーク障害やElasticsearchの3秒のタイムアウトが発生しただけで、2つのデータストア間に即座にデータの不整合が生じます。これをカバーしようとすると、リトライや補償トランザクションの処理が複雑化し、コードベースが肥大化してしまいます。
次に検討されるのが、5分ごとに定期実行するcronバッチです。しかし、テーブルのデータ量が数千万行規模に達すると、SELECT ... WHERE updated_at > ...といったクエリがデータベースのIOPSを枯渇させます。さらに、検索用データには常に最低5分の遅延がつきまといます。
こうした課題を根本から解決するのがChange Data Capture（CDC）アーキテクチャです。データベースに負荷の高い検索クエリを強いるのではなく、CDCはPostgreSQLのWrite-Ahead Log（WAL）を直接読み取ります。追加・更新・削除（INSERT、UPDATE、DELETE）の各操作を、データベースのCPU負荷をほぼゼロに抑えながらリアルタイムに検知します。
Debezium、Kafka、Kafka Connectの3つを組み合わせることで、このパイプラインをスムーズに処理できます。DebeziumがWALをイベントストリームとして読み出し、Kafkaが高信頼な分散メッセージングを担い、Kafka Connectが数百ミリ秒の低遅延でElasticsearchへデータを転送します。
Docker Composeによる環境構築
各ツールをローカル環境に個別にインストールする必要はありません。Docker Composeを使用すれば、わずか数分で一連のクラスタ環境を立ち上げることができます。
1. docker-compose.ymlの準備
作業ディレクトリを作成し、以下の内容でdocker-compose.ymlを保存します。
version: '3.8'
services:
postgres:
image: quay.io/debezium/postgres:15-alpine
ports:
- "5432:5432"
environment:
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=postgres
- POSTGRES_DB=inventory
zookeeper:
image: confluentinc/cp-zookeeper:7.5.0
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
kafka:
image: confluentinc/cp-kafka:7.5.0
depends_on:
- zookeeper
ports:
- "9092:9092"
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
kafka-connect:
image: debezium/connect:2.4
depends_on:
- kafka
- postgres
ports:
- "8083:8083"
environment:
BOOTSTRAP_SERVERS: kafka:9092
GROUP_ID: 1
CONFIG_STORAGE_TOPIC: my_connect_configs
OFFSET_STORAGE_TOPIC: my_connect_offsets
STATUS_STORAGE_TOPIC: my_connect_statuses
elasticsearch:
image: docker.elastic.co/elasticsearch/elasticsearch:8.10.2
ports:
- "9200:9200"
environment:
- discovery.type=single-node
- xpack.security.enabled=false2. クラスタサービスの起動
バックグラウンドでコンテナを起動します：
docker compose up -dKafka Connectが各種プラグインを読み込み、ポート8083でREST APIが利用可能になるまで30〜45秒ほど待ちます。
CDCパイプラインの設定
Debeziumがトランザクションログを抽出できるようにするには、PostgreSQLでwal_level = logicalが有効になっている必要があります。今回使用するquay.io/debezium/postgresイメージではあらかじめ設定されていますが、既存のサーバーやAWS RDSなどのPostgreSQLインスタンスを使用する場合は設定値を確認してください。
1. PostgreSQLでの初期データ作成
顧客テーブルcustomersを作成し、サンプルデータを2件投入します：
docker compose exec -it postgres psql -U postgres -d inventory -c "
CREATE TABLE customers (
id SERIAL PRIMARY KEY,
first_name VARCHAR(255),
last_name VARCHAR(255),
email VARCHAR(255) UNIQUE,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
INSERT INTO customers (first_name, last_name, email) VALUES
('Nguyen', 'An', '[email protected]'),
('Tran', 'Binh', '[email protected]');
"2. Debezium PostgreSQL Source Connectorの登録
ポート8083のKafka Connect REST API経由でConnectorの設定を送信します：
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
http://localhost:8083/connectors/ -d '{
"name": "postgres-source-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": "postgres",
"database.dbname": "inventory",
"topic.prefix": "cdc",
"table.include.list": "public.customers"
}
}'リクエストを受け取ると、Debeziumはまず既存データのスナップショットを取得します。それ以降の変更データは、自動的にKafkaトピックcdc.public.customersへパブリッシュされます。
3. Unwrap SMTによるElasticsearchへのデータ転送
デフォルト状態のDebeziumメッセージには、スキーマ情報や変更前後の状態（before／after）などを含む複雑なエンベロープが含まれています。Elasticsearch側では最新のペイロードのみが必要となるため、ExtractNewRecordState（Unwrap SMT）変換プラグインを使い、インデックス登録前に純粋なデータオブジェクトのみを抽出します：
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" \
http://localhost:8083/connectors/ -d '{
"name": "elastic-sink-connector",
"config": {
"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"tasks.max": "1",
"topics": "cdc.public.customers",
"connection.url": "http://elasticsearch:9200",
"type.name": "_doc",
"key.ignore": "false",
"schema.ignore": "true",
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.drop.tombstones": "false"
}
}'ConfluentのSinkプラグインを使用しない場合は、PythonやGoなどでKafkaメッセージをコンシュームしてElasticsearchの_bulk APIを呼び出す軽量ワーカーを作成して代用することも可能です。
動作検証と運用ポイント
パイプライン全体の接続が完了しました。実際の同期遅延を検証してみましょう。
1. Connectorの稼働状況を確認
両方のConnectorが正常に動作しているか確認します：
curl -s http://localhost:8083/connectors/postgres-source-connector/status | jq .
curl -s http://localhost:8083/connectors/elastic-sink-connector/status | jq .stateフィールドの値がRUNNINGであることを確認してください。もしFAILEDになっている場合は、docker compose logs kafka-connectでコンテナログを確認し、認証エラーやプラグインの不足がないかトラブルシューティングを行います。
2. リアルタイムデータ同期の検証
PostgreSQLに新しい顧客レコードを挿入します：
docker compose exec -it postgres psql -U postgres -d inventory -c "
INSERT INTO customers (first_name, last_name, email) VALUES ('Le', 'Cuong', '[email protected]');
"ほぼ一瞬（通常150ms〜300ms程度の遅延）で、Elasticsearch側で検索クエリを実行してみます：
curl -s http://localhost:9200/cdc.public.customers/_search?q=last_name:Cuong | jq .hits.hitsアプリケーション側に同期コードを一切追加することなく、新規顧客「Le Cuong」のレコードが検索インデックスに反映されていることが確認できます。
3. 本番運用で不可欠な3つの重要メトリクス
CDCを本番環境で運用する際は、以下のリスクを継続的に監視する必要があります：
- PostgreSQLのレプリケーションスロット遅延（Replication Slot Lag）：最も注意すべき落とし穴です。Kafka Connectクラスタが不意に停止した場合、PostgreSQLは未読み出しのWALファイルを蓄積し続けます。これにより、わずか数時間でディスク容量が20GBから200GBへと急増し、プライマリデータベースが停止する危険があります。
pg_replication_slotsを監視し、WALの保持容量に対するアラートを必ず設定してください。 - Kafkaのコンシューマーラグ（Consumer Lag）：生成された最新オフセットと、Elasticsearch Sinkが処理したオフセットの差分を計測します。ラグが増加し続けている場合、Elasticsearchへの書き込みがボトルネックになっています。Connectorのタスク数を増やすか、Elasticsearch側のリフレッシュ間隔（refresh interval）のチューニングを検討してください。
- Kafka ConnectのJMXメトリクス監視：Prometheus JMX Exporterを組み込み、毎秒のレコード読み取りレート（poll rate）やオフセットコミット時間（commit time）を監視します。ユーザーから問い合わせが入る前に、ボトルネックを早期検知できる体制を整えましょう。
