Debezium 실무 스터디 가이드

1. Debezium란?

Debezium는 오픈소스 Change Data Capture (CDC) 플랫폼입니다. 데이터베이스의 트랜잭션 로그(예: MySQL의 binlog, PostgreSQL의 WAL)를 모니터링하여 데이터 변경 사항을 실시간으로 캡처하고, 이를 이벤트 스트림(주로 Apache Kafka)으로 전송하는 역할을 합니다.

마이크로서비스 아키텍처, 데이터 레이크/웨어하우스 적재, 캐시/검색엔진 동기화 등 실시간 데이터 파이프라인의 핵심 컴포넌트로 사용됩니다.

2. 왜 써야 하는가?

기존 데이터 동기화 방식과 Debezium의 로그 기반 CDC를 비교합니다.

3. 핵심 개념

Debezium는 Kafka Connect의 소스 커넥터(Source Connector)로 동작합니다. 아래는 전형적인 아키텍처입니다.

주요 컴포넌트

• 커넥터(Connector): 특정 DBMS용 플러그인 (예: debezium-connector-mysql).
• 태스크(Task): 커넥터의 실제 작업 인스턴스. 병렬 처리를 위해 여러 태스크 실행 가능.
• 오프셋(Offset): 커넥터가 읽은 로그의 위치를 저장하여 장애 복구 시 중단 지점부터 재개.
• 스키마 레지스트리(Schema Registry): 이벤트의 Avro 스키마를 중앙 관리 (Confluent Schema Registry와 연동).

4. 퀵스타트 (MySQL + Docker)

Docker Compose를 사용해 MySQL, Zookeeper, Kafka, Kafka Connect (Debezium)를 한 번에 실행합니다.

단계 1: docker-compose.yml 작성

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  mysql:
    image: mysql:8.0
    environment:
      MYSQL_ROOT_PASSWORD: debezium
      MYSQL_USER: debezium
      MYSQL_PASSWORD: dbz
    ports:
      - "3306:3306"
    command: --binlog-format=ROW --default-authentication-plugin=mysql_native_password

  connect:
    image: debezium/connect:latest
    depends_on:
      - kafka
      - mysql
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_statuses
    ports:
      - "8083:8083"

단계 2: 인프라 실행 및 커넥터 등록

아래 명령어를 터미널에서 실행합니다.

# 1. 인프라 실행
docker-compose up -d

# 2. MySQL에 샘플 데이터베이스 및 테이블 생성 (다른 터미널에서)
docker exec -it <mysql-container-id> mysql -u root -pdebezium

MySQL 클라이언트에서:

CREATE DATABASE inventory;
USE inventory;

CREATE TABLE customers (
  id INT PRIMARY KEY AUTO_INCREMENT,
  first_name VARCHAR(50),
  last_name VARCHAR(50),
  email VARCHAR(100)
);

INSERT INTO customers (first_name, last_name, email) VALUES
('John', 'Doe', 'john@example.com'),
('Jane', 'Smith', 'jane@example.com');

단계 3: Debezium MySQL 커넥터 등록

Kafka Connect REST API를 호출해 커넥터를 생성합니다.

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "inventory-connector",
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "tasks.max": "1",
      "database.hostname": "mysql",
      "database.port": "3306",
      "database.user": "root",
      "database.password": "debezium",
      "database.server.id": "184054",
      "database.server.name": "dbserver1",
      "database.include.list": "inventory",
      "table.include.list": "inventory.customers",
      "database.history.kafka.bootstrap.servers": "kafka:9092",
      "database.history.kafka.topic": "dbhistory.inventory",
      "include.schema.changes": "true"
    }
  }'

단계 4: 이벤트 소비 (Python 예제)

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'dbserver1.inventory.customers',  # 토픽명: {server}.{database}.{table}
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

print("Listening for changes...")
for message in consumer:
    event = message.value
    op = event['op']  # 'c'=create, 'u'=update, 'd'=delete
    source = event['source']
    after = event.get('after', {})
    
    print(f"Operation: {op} | Table: {source['table']} | Data: {after}")

5. 핵심 설정 레퍼런스

MySQL 커넥터를 기준으로 한 필수 및 주요 설정 목록입니다.

6. 실무 팁 & 트러블슈팅

자주 겪는 문제

성능 튜닝 팁

• 병렬 처리: 파티션 키를 명시적으로 설정하거나 tasks.max를 늘려 여러 테이블을 병렬로 처리.
• 스냅샷 최적화: 대용량 테이블의 초기 스냅샷 시 snapshot.mode=initial_only 후 증분 모드로 전환 고려.
• 모니터링: JMX 메트릭 (예: MaxQueueSize, MilliSecondsBehindSource)을 Grafana 대시보드로 시각화.
• 오프셋 관리: 정기적으로 오프셋 토픽의 로그 압축 상태를 확인하고, 커넥터 재시작 시 오프셋 초기화에 주의.

# 커넥터 상태 확인
curl -s http://localhost:8083/connectors/inventory-connector/status | jq .

# 커넥터 설정 조회
curl -s http://localhost:8083/connectors/inventory-connector | jq .

# Kafka 토픽 메시지 확인 (최근 5개)
docker exec <kafka-container> kafka-console-consumer \
  --bootstrap-server localhost:9092 \
  --topic dbserver1.inventory.customers \
  --from-beginning --max-messages 5

7. 참고 자료

• 📘 Debezium 공식 문서 - 가장 상세하고 최신의 정보원
• 🧪 Debezium 예제 저장소 - 다양한 DB와 시나리오별 샘플 코드
• 🔗 Kafka KIPs - Kafka의 변화와 최신 기능 이해
• ✍️ Confluent 블로그 - CDC와 스트리밍 데이터 파이프라인 실무 사례
• 추천 도서: "Kafka: The Definitive Guide" (CDC와 Debezium 관련 장 포함)