✍️ 필사 모드: CDC (Change Data Capture) 완전 가이드 2025: Debezium, Kafka, 데이터 동기화, Outbox 패턴

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",  // c=create, u=update, d=delete, r=read(snapshot)
  "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"새 사용자: {event['after']}")
    elif op == 'u':
        print(f"업데이트: {event['before']} → {event['after']}")
    elif op == 'd':
        print(f"삭제: {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]
   (배치, 일 1회)  (느림)   (대용량)

[PostgreSQL] → [Debezium] → [Kafka] → [Stream Processor] → [Snowflake]
                                              ↓
                                       (필요 시 transform)

[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}")  # 누가 잊으면? 다른 서비스에서 업데이트하면?

# Kafka consumer
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