Claude Code CDC実装｜データ変更検知とリアルタイム同期

データベースの変更を検知し、他システムへリアルタイムに同期するCDC（Change Data Capture）は、マイクロサービスやデータ分析基盤を運用する多くの企業で必須の技術要件となっています。一方で、CDC の設計には複数のアプローチがあり、ログベース・トリガーベース・ポーリングベースの選択、イベントストリームの冪等性保証、ネットワーク障害時のリトライ戦略など、検討すべき技術課題が山積しています。私自身、デジライズでお客様のデータ基盤構築を支援する中で、CDC 設計の適切なパターン選択が、システムの可用性と運用負荷に直結することを実感してきました。本記事では、Claude Code を活用した CDC 実装の設計パターンと、環境ごとに異なる要件への対応指針を、エラーハンドリング戦略を含めて技術的に解説します。

本記事の結論: CDC 実装は「ログベース」「トリガーベース」「ポーリングベース」の3パターンから要件に応じて選択し、Claude Code でイベントストリーム処理とエラーハンドリングを自動実装することで、構築期間を短縮できる

CDC実装の3つの設計パターンと選択基準

CDC の実装には、データベースの種類・変更頻度・遅延許容度・運用体制により、適切なアプローチが異なります。Claude Code を用いた実装でも、この基本的な設計判断は人間が行う必要があります。

1. ログベースCDC（PostgreSQL WAL / MySQL binlog） — データベースのトランザクションログを購読し、INSERT / UPDATE / DELETE を検知する方式。遅延が最小（ミリ秒〜秒単位）で、データベースへの負荷が軽い反面、ログフォーマットの解析が必要。本番環境での利用が多い。

2. トリガーベースCDC — データベーストリガーで変更イベントを専用テーブルに記録し、別プロセスが読み取る方式。実装がシンプルで既存DBに依存しないが、書き込みトランザクションへのオーバーヘッドが発生する。中小規模の環境で採用されることが多い。

3. ポーリングベースCDC — 定期的に updated_at カラムを参照し差分を取得する方式。実装が最も容易だが、遅延が大きく（分単位）、頻繁なポーリングはDB負荷となる。リアルタイム性が不要な集計用途に適する。

Claude Code に「PostgreSQL の WAL を Kafka に流すコンポーネントを実装して」と依頼する場合、Debezium や pg_recvlogical などの既存ツールとの統合コードを生成させることが現実的です。完全に独自実装する場合は、ログフォーマット解析のロジックが複雑になるため、段階的な検証が必要です。

既存のデータウェアハウス連携については、Claude Code データウェアハウス連携｜ETL自動化とバッチ最適化で詳述しています。CDC はそこで扱うバッチ連携を「リアルタイム化」する技術として位置づけられます。

イベントストリーム処理の実装パターン

CDC で検知した変更イベントを、Kafka・Kinesis・Pub/Sub などのストリーム基盤に流し、複数の消費者（Consumer）が並行処理する構成が一般的です。Claude Code を用いた実装でも、以下の技術課題への対処が必要です。

イベント順序保証とパーティショニング

同一レコードへの複数回更新が発生する場合、イベントの処理順序が前後すると最終的なデータ状態が不整合になります。Kafka であればパーティションキーを主キーに設定し、同一レコードの変更が同一パーティションに集約されることで順序を保証できます。

# Claude Code に生成させる Kafka Producer の例（パーティションキー指定）
from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

def send_change_event(table_name, operation, record_id, payload):
    # record_id をパーティションキーとして送信
    producer.send(
        topic=f'cdc.{table_name}',
        key=str(record_id).encode('utf-8'),
        value={
            'operation': operation,  # INSERT / UPDATE / DELETE
            'timestamp': payload.get('updated_at'),
            'data': payload
        }
    )
    producer.flush()

Claude Code に「Kafka のパーティション設計を考慮したイベント送信処理を実装して」と依頼すると、上記のようなキー指定ロジックを含むコードが生成されます。ただし、パーティション数の決定や、リバランス時の処理停止時間の見積もりは、環境のスループット要件により異なるため、人間が判断する必要があります。

冪等性とデデュプリケーション

ネットワーク障害などでイベントが重複送信された場合、同じ変更を複数回適用すると不整合が生じます。Consumer 側で冪等性を保証する実装が必要です。

重複処理のリスク: 同一の UPDATE イベントを2回適用すると、カウンタ型のカラム（在庫数など）が意図しない値になる可能性がある。イベントID によるデデュプリケーションが必須

Claude Code に以下のような指示を与えることで、Redis を用いたデデュプリケーション処理を生成できます。

「Kafka Consumer でイベントを受信し、Redis に処理済みイベントIDを記録して重複を防ぐ Python コードを実装してください。イベントIDは event_id フィールドに含まれ、TTL は24時間とします。」

生成例:

import redis
from kafka import KafkaConsumer
import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)
consumer = KafkaConsumer(
    'cdc.orders',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    event = message.value
    event_id = event.get('event_id')
    
    # Redis で重複チェック（SETNX で原子的に確認）
    if r.setnx(f'processed:{event_id}', '1'):
        r.expire(f'processed:{event_id}', 86400)  # 24時間 TTL
        
        # 実際の処理（データベース更新など）
        apply_change(event)
    else:
        print(f"Duplicate event {event_id}, skipping")

この実装では、SETNX の原子性により、複数 Consumer が並行稼働しても重複処理が防止されます。ただし、Redis 自体の可用性設計（Sentinel / Cluster）は別途必要です。

エラーハンドリングとリトライ戦略

CDC パイプラインでは、ネットワーク障害・データベース接続エラー・スキーマ不整合など、多様な障害が発生します。Claude Code を用いた実装でも、以下のリトライ戦略を明示的に設計する必要があります。

エラー種別	リトライ可否	対応方針
一時的なネットワーク障害	可	Exponential Backoff で最大3回リトライ
データベース接続タイムアウト	可	接続プールの再構築後にリトライ
スキーマ不一致（カラム欠落）	不可	Dead Letter Queue へ移動し、アラート発報
主キー重複エラー（Upsert 失敗）	不可	既存レコードを確認し、UPDATE に切り替え

Claude Code に「Kafka Consumer でリトライ処理を実装してください。一時エラーは Exponential Backoff で3回まで再試行し、恒久エラーは DLQ トピックに送信します」と依頼すると、以下のような実装が生成されます。

import time
from kafka import KafkaProducer, KafkaConsumer
from kafka.errors import KafkaError

dlq_producer = KafkaProducer(bootstrap_servers=['localhost:9092'])

def process_with_retry(event, max_retries=3):
    for attempt in range(max_retries):
        try:
            # 実際の処理（データベース書き込みなど）
            apply_change_to_db(event)
            return True
        except TemporaryError as e:
            wait_time = 2 ** attempt  # 1秒, 2秒, 4秒
            print(f"Retry {attempt+1}/{max_retries} after {wait_time}s: {e}")
            time.sleep(wait_time)
        except PermanentError as e:
            # DLQ へ送信
            dlq_producer.send('cdc.dlq', value=event)
            print(f"Permanent error, sent to DLQ: {e}")
            return False
    
    # 最大リトライ回数を超えた場合も DLQ へ
    dlq_producer.send('cdc.dlq', value=event)
    return False

この実装では、TemporaryError と PermanentError の分類を人間が定義する必要があります。例えば、PostgreSQL の OperationalError を Temporary、IntegrityError を Permanent と判定するなど、データベースドライバーのエラークラスに応じた分岐が必要です。

スキーマ進化への対応とバージョン管理

CDC パイプラインでは、ソーステーブルにカラムが追加・削除された際、Consumer 側のスキーマ定義を同期する必要があります。Avro や Protobuf などのスキーマレジストリを用いることで、後方互換性を保ちながらスキーマを進化させることができます。

Claude Code に「Confluent Schema Registry と Avro を使った Kafka Producer を実装してください。スキーマは user_schema.avsc ファイルから読み込みます」と依頼すると、以下のような実装が生成されます。

from confluent_kafka import avro
from confluent_kafka.avro import AvroProducer

# Avro スキーマの読み込み
value_schema = avro.load('user_schema.avsc')

avro_producer = AvroProducer({
    'bootstrap.servers': 'localhost:9092',
    'schema.registry.url': 'http://localhost:8081'
}, default_value_schema=value_schema)

def send_avro_event(user_data):
    avro_producer.produce(
        topic='cdc.users',
        value=user_data  # スキーマに準拠した dict
    )
    avro_producer.flush()

スキーマの後方互換性ルール（カラム追加時は default 値を設定、削除時は optional 化）は、Schema Registry の互換性設定（BACKWARD / FORWARD / FULL）により制御されます。これらの運用ポリシーは、環境の変更頻度と Consumer の更新タイミングにより異なるため、要件定義時に決定する必要があります。

マイクロサービス間のオーケストレーションにおける CDC の活用例は、Claude Code マイクロサービスオーケストレーション｜Saga パターン実装で紹介しています。

モニタリングと遅延検知の実装

CDC パイプラインでは、変更検知から最終的な同期完了までの End-to-End 遅延を監視する必要があります。Claude Code を用いて、各ステージでのタイムスタンプ記録と Prometheus メトリクス出力を実装できます。

「Kafka Consumer で処理遅延を計測し、Prometheus のヒストグラムとして出力する Python コードを実装してください。イベントには source_timestamp フィールドが含まれます」という指示で以下が生成されます。

from prometheus_client import Histogram, start_http_server
import time

# Prometheus メトリクス定義
processing_latency = Histogram(
    'cdc_processing_latency_seconds',
    'Time from source event to processing completion',
    buckets=[0.1, 0.5, 1.0, 5.0, 10.0, 30.0]
)

start_http_server(8000)  # Prometheus のスクレイプ用エンドポイント

for message in consumer:
    event = message.value
    source_ts = event.get('source_timestamp')
    
    start_time = time.time()
    apply_change(event)
    
    # 遅延を計測（秒単位）
    latency = time.time() - source_ts
    processing_latency.observe(latency)

この実装により、Grafana などで遅延の分布をモニタリングできます。アラート閾値（例: P95 遅延が10秒超過）は、ビジネス要件により異なるため、運用開始後に調整します。

遅延の主要因: データベースのロック待ち・Kafka パーティションのリバランス・Consumer の処理遅延など、複数の要因が重なる。メトリクスは各ステージで取得し、ボトルネックを特定する

データベース移行時のCDC活用パターン

既存システムから新システムへのデータベース移行時、CDC を用いてダウンタイムを最小化するアプローチがあります。初期データコピー後、CDC で差分を継続同期し、切り替え時点で同期を停止する戦略です。

1. 初期データコピー — 既存DBのスナップショットを新DBへ一括コピー（pg_dump / mysqldump など）

2. CDC 同期開始 — コピー完了時点から CDC を起動し、差分変更を新DBへ反映

3. 遅延収束確認 — CDC の処理遅延が十分小さくなる（数秒以内）まで待機

4. 切り替え — アプリケーションの書き込み先を新DBへ変更し、CDC を停止

Claude Code に「PostgreSQL から MySQL への移行用に、Debezium で CDC を設定し、差分データを変換して MySQL に投入する Python コードを実装してください」と依頼することで、基本的な変換ロジックを生成できます。ただし、データ型の差異（PostgreSQL の JSONB → MySQL の JSON、ARRAY 型の扱いなど）は環境ごとに異なるため、手動での調整が必要です。

データベース移行全般の戦略については、Claude Code データベース移行｜スキーマ変換とデータ整合性で詳しく解説しています。

まとめ

CDC 実装では、ログベース・トリガーベース・ポーリングベースの3パターンから、遅延許容度とデータベース負荷のバランスを考慮して選択します。Claude Code を活用することで、イベントストリーム処理・冪等性保証・エラーハンドリングの定型コードを迅速に生成できますが、パーティション設計・リトライ戦略・スキーマ進化ポリシーは環境の要件により異なるため、人間が判断する必要があります。

3パターン

CDC 設計アプローチ

4ステージ

移行時の CDC 活用手順

要件次第

遅延・負荷のトレードオフ

デジライズでは、Claude Code を活用した CDC 実装の設計支援を、研修とコンサルティングの2本柱で提供しています。お客様の環境に応じたパターン選択・エラーハンドリング戦略の策定・モニタリング設計まで、実装と並行してサポートします。CDC パイプラインの構築や既存システムの移行にお悩みの際は、無料相談をご活用ください。具体的な要件をお伺いし、最適な実装アプローチをご提案いたします。

数値・事例・提供条件について

デジライズの実績は社内集計値です。特に明記のない数値付き事例は、匿名加工された実例をもとにしたモデルケースです。導入効果は企業や業務によって異なります。各サービスの料金・機能・提供条件は記事の公開・更新時点の情報であり、変更されるため、最新情報はAnthropic公式サイトなどの一次情報をご確認ください。

株式会社デジライズ

Claude Codeの法人導入・研修・業務活用を支援しています。

法人導入支援サービスを見る

自社での活用を、具体的に。

業務の整理から研修・導入・運用まで、ご相談ください。

資料請求・お問い合わせ

CDC実装の3つの設計パターンと選択基準

イベントストリーム処理の実装パターン

イベント順序保証とパーティショニング

冪等性とデデュプリケーション

エラーハンドリングとリトライ戦略

スキーマ進化への対応とバージョン管理

モニタリングと遅延検知の実装

データベース移行時のCDC活用パターン

まとめ

関連記事

自社での活用を、具体的に。

あわせて読みたい記事

Claude Code アクセシビリティ対応｜WCAG準拠コード生成ガイド

Claude Code 同時セッション管理｜複数チームでの並行利用設計

Claude Code モデル選択フレームワーク｜Opus/Sonnet/Haiku 使い分けガイド