✍️ 필사 모드: CDC (Change Data Capture) 完全ガイド 2025: Debezium, Kafka, データ同期, Outbox Pattern

while True:
    new_records = db.query("SELECT * FROM users WHERE updated_at > ?", last_sync)
    sync_to_warehouse(new_records)
    last_sync = now()
    sleep(60)

[DB] -> [WAL/Binlog] -> [CDC tool] -> [Kafka] -> [Consumers]
                            ^
                リアルタイム、非侵襲的

INSERT INTO users (name) VALUES ('Alice');
-> [WAL]: tx_id=123, op=INSERT, table=users, data={name:'Alice'}
-> [Disk page write]

# postgresql.conf
wal_level = logical
max_wal_senders = 10
max_replication_slots = 10

-- ユーザ作成
CREATE USER cdc_user REPLICATION LOGIN PASSWORD 'secret';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO cdc_user;

-- Publication 作成 (どのテーブルを複製するか)
CREATE PUBLICATION my_pub FOR TABLE users, orders;
-- 全テーブル対象
CREATE PUBLICATION my_pub FOR ALL TABLES;

-- Replication slot 作成 (変更を保存する場所)
SELECT pg_create_logical_replication_slot('my_slot', 'pgoutput');

# my.cnf
log_bin = mysql-bin
binlog_format = ROW  # CDC には ROW 必須
binlog_row_image = FULL
server_id = 1
expire_logs_days = 7

const changeStream = db.collection('users').watch()
changeStream.on('change', (change) => {
  console.log(change)
  // { operationType: 'insert', fullDocument: {...}, ... }
})

[PostgreSQL] -> [Debezium connector] -> [Kafka Connect] -> [Kafka topics] -> [Consumers]

version: '3.7'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    depends_on: [zookeeper]
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092

  connect:
    image: debezium/connect:2.5
    depends_on: [kafka, postgres]
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_statuses

  postgres:
    image: debezium/example-postgres:2.5
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres

curl -X POST http://connect:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "postgres-connector",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "database.hostname": "postgres",
      "database.port": "5432",
      "database.user": "cdc_user",
      "database.password": "secret",
      "database.dbname": "mydb",
      "database.server.name": "mydb",
      "table.include.list": "public.users,public.orders",
      "plugin.name": "pgoutput",
      "publication.name": "my_pub",
      "slot.name": "my_slot"
    }
  }'

mydb.public.users   <- users テーブルの変更
mydb.public.orders  <- orders テーブルの変更

{
  "before": null,
  "after": {
    "id": 123,
    "name": "Alice",
    "email": "alice@example.com"
  },
  "source": {
    "version": "2.5",
    "connector": "postgresql",
    "name": "mydb",
    "ts_ms": 1681545600000,
    "snapshot": "false",
    "db": "mydb",
    "schema": "public",
    "table": "users",
    "txId": 12345,
    "lsn": 23456789
  },
  "op": "c",
  "ts_ms": 1681545600000
}

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'mydb.public.users',
    bootstrap_servers=['kafka:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    event = message.value
    op = event['op']
    if op == 'c':
        print(f"new user: {event['after']}")
    elif op == 'u':
        print(f"update: {event['before']} -> {event['after']}")
    elif op == 'd':
        print(f"delete: {event['before']}")

def create_order(order):
    db.save(order)              # 1
    kafka.send("orders", order) # 2

CREATE TABLE outbox (
  id UUID PRIMARY KEY,
  aggregate_type VARCHAR(255),
  aggregate_id VARCHAR(255),
  event_type VARCHAR(255),
  payload JSONB,
  created_at TIMESTAMP DEFAULT NOW()
);

def create_order(order):
    with db.transaction():
        db.save(order)
        db.save(OutboxEvent(
            aggregate_type='Order',
            aggregate_id=order.id,
            event_type='OrderCreated',
            payload=order.to_json()
        ))

while True:
    events = db.query("""
        SELECT * FROM outbox
        WHERE published = false
        ORDER BY created_at
        LIMIT 100
    """)
    for event in events:
        kafka.send(event.aggregate_type, event.payload)
        db.execute("UPDATE outbox SET published = true WHERE id = ?", event.id)
    sleep(1)

curl -X POST http://connect:8083/connectors \
  -d '{
    "name": "outbox-connector",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "table.include.list": "public.outbox",
      "transforms": "outbox",
      "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
      "transforms.outbox.route.by.field": "aggregate_type",
      "transforms.outbox.table.field.event.id": "id",
      "transforms.outbox.table.field.event.payload": "payload"
    }
  }'

[Extract] -> [Transform] -> [Load]
(バッチ・日次)  (遅い)    (大容量)

[PostgreSQL] -> [Debezium] -> [Kafka] -> [Stream Processor] -> [Snowflake]

[PostgreSQL] -> [Debezium] -> [Kafka] -> [Elasticsearch Sink Connector] -> [Elasticsearch]

curl -X POST http://connect:8083/connectors \
  -d '{
    "name": "es-sink",
    "config": {
      "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
      "topics": "mydb.public.products",
      "connection.url": "http://elasticsearch:9200",
      "type.name": "_doc",
      "key.ignore": "false",
      "schema.ignore": "true",
      "behavior.on.malformed.documents": "warn"
    }
  }'

def get_user(user_id):
    user = cache.get(f"user:{user_id}")
    if user is None:
        user = db.query(...)
        cache.set(f"user:{user_id}", user, ttl=3600)
    return user

def update_user(user_id, data):
    db.execute("UPDATE users SET ... WHERE id=?", user_id)
    cache.delete(f"user:{user_id}")  # 忘れたら? 他サービスが更新したら?

consumer = KafkaConsumer('mydb.public.users')

for message in consumer:
    event = message.value
    user_id = event['after']['id'] if event['op'] != 'd' else event['before']['id']
    cache.delete(f"user:{user_id}")

[MySQL (現行)] -> [Debezium] -> [Kafka] -> [PostgreSQL (新)]

-- アクティブなスロット確認
SELECT slot_name, active, restart_lsn
FROM pg_replication_slots;

-- 使ってないスロット削除
SELECT pg_drop_replication_slot('unused_slot');

# Connector 状態
curl http://connect:8083/connectors/postgres-connector/status

# メトリクス (JMX)
debezium_metrics_QueueRemainingCapacity
debezium_metrics_NumberOfEventsFiltered