AWS ML CI/CD와 CDC 파이프라인 구축 (4) CDC 파이프라인: Debezium 로컬 구축과 실험

docker compose 기반 MySQL, Kafka, Debezium, PostgreSQL 구성부터 초기 snapshot과 실시간 CDC 검증, 스키마 변경과 incremental snapshot 실험까지 CDC 파이프라인 로컬 구축 과정을 정리합니다.

Posted 2026/05/25

By 강호승

5 min read

AWS ML CI/CD와 CDC 파이프라인 구축 프로젝트 시리즈의 4편입니다. 프로젝트 소개는 0편에 있습니다.

3편까지 ML CI/CD 축을 다뤘고, 이번 편은 2026-07-25에 진행한 다른 한 축, CDC 파이프라인의 구축과 실험 기록입니다. 운영 DB(MySQL)의 변경을 분석용 DB(PostgreSQL)로 실시간 동기화하는 구간을 로컬 docker compose로 구축하고, 초기 snapshot, 실시간 반영, 스키마 변경, 재동기화까지 검증했습니다. AWS에 올릴 때는 MySQL이 RDS, Kafka가 MSK, Debezium과 sink가 MSK Connect로 바뀔 뿐 구성요소는 동일하므로, 검증은 전부 무료인 로컬에서 했습니다.

구성: 컨테이너 4개

cdc/docker-compose.yml 하나로 4개 container를 기동합니다.

flowchart LR
    MYSQL[("MySQL 8.0<br/>operations DB")] -->|binlog| DBZ["Kafka Connect 3.0<br/>Debezium source"]
    DBZ --> KAFKA["Kafka 3.9.1<br/>ops.operations.*"]
    KAFKA --> SINK["JDBC sink<br/>upsert"]
    SINK --> PG[("PostgreSQL 16<br/>analytics DB")]

mysql (mysql:8.0): 소스 DB. MySQL 8.0은 binlog가 기본 활성(ROW 포맷)이라 --server-id=1 --binlog-format=ROW만 명시했습니다. 초기화 스크립트는 mysql-init/을 entrypoint 디렉터리에 마운트해서 실행합니다.
kafka (apache/kafka:3.9.1): KRaft 모드 단일 브로커로, ZooKeeper 없이 broker와 controller 역할을 한 노드가 겸합니다. 단일 브로커 데모라 내부 토픽 복제계수는 1로 낮췄습니다.
connect (quay.io/debezium/connect:3.0): Kafka Connect. 8083 포트로 REST API를 열고, 자신의 설정과 offset과 상태를 Kafka 내부 토픽에 저장합니다. 커넥터가 재시작해도 binlog를 어디까지 읽었는지가 offset 토픽에 남아 이어서 읽습니다.
postgres (postgres:16): 분석용 타깃 DB.

Debezium은 설치하는 프로그램이 아니라 Kafka Connect 위에서 도는 connector 플러그인입니다. 설정 JSON을 REST API로 등록하면 그때부터 동작하고, compose에 올라가는 것은 Debezium이 포함된 Kafka Connect container입니다.

구성 파일은 4개입니다.

파일	역할
`cdc/docker-compose.yml`	MySQL + Kafka(KRaft) + Kafka Connect(Debezium) + PostgreSQL 기동
`cdc/mysql-init/01-schema.sql`	소스 스키마와 샘플 데이터, debezium 계정 권한, signal 테이블
`cdc/connectors/source-mysql.json`	Debezium MySQL source connector 설정
`cdc/connectors/sink-postgres.json`	JDBC sink connector 설정

커넥터 설정 2개

source 쪽(source-mysql.json)의 핵심 설정은 다음과 같습니다.

connector.class: io.debezium.connector.mysql.MySqlConnector. 전용 계정 debezium으로 접속합니다.
topic.prefix=ops, database.include.list=operations: operations DB의 변경이 ops.operations.<테이블> 토픽으로 발행됩니다.
snapshot.mode=initial: 커넥터 최초 등록 시 기존 데이터 전체를 읽고, 그 시점의 binlog 위치부터 스트리밍을 시작합니다.
signal.data.collection=operations.dbz_signal: incremental snapshot 명령을 받을 signal 테이블 지정입니다.
schema.history.internal.kafka.topic: DDL 이력을 영속화하는 schema history 토픽입니다.

sink 쪽(sink-postgres.json)의 핵심 설정입니다.

connector.class: io.debezium.connector.jdbc.JdbcSinkConnector. topics.regex로 vehicles, sensor_readings, service_orders 세 테이블의 토픽을 구독합니다.
insert.mode=upsert, primary.key.mode=record_key: 레코드 키(소스 테이블 PK) 기준 upsert입니다. 같은 이벤트가 중복 전달돼도 결과가 같습니다.
delete.enabled=true: 소스의 DELETE를 타깃 행 삭제로 반영합니다.
schema.evolution=basic: 이벤트에 새 컬럼이 등장하면 타깃에 ALTER TABLE ADD COLUMN을 자동 실행합니다.
RegexRouter transform: 토픽 이름 ops.operations.vehicles를 타깃 테이블 이름 vehicles로 라우팅합니다.

기동과 커넥터 등록

  
cd cdc
docker compose up -d              # 4개 컨테이너 기동
docker compose ps                 # 4개 Up 확인
curl localhost:8083               # Kafka Connect 준비 확인

curl -X POST -H "Content-Type: application/json" \
  --data @connectors/source-mysql.json  localhost:8083/connectors
curl -X POST -H "Content-Type: application/json" \
  --data @connectors/sink-postgres.json localhost:8083/connectors

curl -s localhost:8083/connectors/source-operations-mysql/status   # RUNNING 확인
curl -s localhost:8083/connectors/sink-analytics-postgres/status

두 status 호출 모두 RUNNING을 반환하면 파이프라인이 가동 중입니다.

기본 검증: 초기 snapshot과 실시간 CDC

먼저 초기 snapshot입니다. 커넥터를 등록만 했는데 소스에 미리 넣어 둔 기존 3행이 PostgreSQL에 넘어와 있었습니다. snapshot.mode=initial이 등록 시점에 기존 데이터를 읽어 보낸 결과입니다.

  
# snapshot: 등록만 했는데 기존 3행이 PostgreSQL에 넘어와 있음
docker exec cdc-postgres psql -U analytics -c 'SELECT * FROM vehicles;'

다음은 실시간 CDC입니다. 소스에 INSERT를 넣으면 수 초 내에 타깃에서 조회됩니다.

  
docker exec cdc-mysql mysql -uroot -proot operations \
  -e "INSERT INTO vehicles VALUES ('KMHXX00XXXX000004', 2024);"
docker exec cdc-postgres psql -U analytics -c 'SELECT * FROM vehicles;'

실험 A: 스키마 변경 대응

운영 DB의 ALTER TABLE은 서비스 배포마다 일어나는 일입니다. 대응이 없으면 파이프라인이 중단되거나 새 컬럼 값이 조용히 유실됩니다. 소스에 컬럼을 추가하고 새 컬럼을 쓰는 INSERT를 실행해 봤습니다.

  
docker exec cdc-mysql mysql -uroot -proot operations \
  -e "ALTER TABLE vehicles ADD COLUMN region VARCHAR(10);
      INSERT INTO vehicles VALUES ('KMHXX00XXXX000005', 2025, 'seoul');"
docker exec cdc-postgres psql -U analytics -c 'SELECT * FROM vehicles ORDER BY vin;'

결과는 다음과 같습니다.

타깃 PostgreSQL에 region 컬럼이 자동 생성됐습니다.
신규 행의 값 seoul이 유실 없이 도착했습니다.
기존 행의 region은 NULL입니다.
파이프라인은 중단 없이 계속 동작했습니다.

동작 원리는 2단계입니다. 캡처 측에서는 DDL도 binlog에 기록되므로 Debezium이 이를 읽어 내부 스키마 모델을 갱신하고(schema history 토픽에 영속화) 이후 이벤트에 새 구조를 실어 보냅니다. 반영 측에서는 sink의 schema.evolution=basic이 새 컬럼을 보고 타깃에 ALTER TABLE ADD COLUMN을 실행합니다. 단, 자동의 범위는 컬럼 추가 같은 가산적 변경까지입니다. 컬럼 삭제나 이름 변경, 비호환 타입 변경은 타깃 선반영 같은 수동 절차로 다뤄야 합니다.

실험 B: incremental snapshot으로 재동기화

CDC는 구독 시작 이후의 변경만 흘려보내므로, 이미 쌓여 있던 데이터나 장애로 유실된 데이터는 별도 경로로 옮겨야 합니다. Debezium의 incremental snapshot은 운영 중 언제든 signal 테이블에 명령 행을 넣어 요청할 수 있고, 테이블을 PK 순 청크로 나눠 읽으면서 스트리밍과 병행합니다.

타깃 데이터를 일부러 삭제해 소스와 사본의 불일치 상황을 재연한 뒤, signal 테이블에 재스냅샷 명령을 넣었습니다.

  
# 사고 재연: 타깃 데이터 삭제 (소스는 그대로, 불일치 상태)
docker exec cdc-postgres psql -U analytics -c 'DELETE FROM sensor_readings;'

# 재스냅샷 신호 = signal 테이블에 INSERT
docker exec cdc-mysql mysql -uroot -proot operations \
  -e "INSERT INTO dbz_signal VALUES ('resync-2', 'execute-snapshot',
      '{\"data-collections\":[\"operations.sensor_readings\"]}');"

최종적으로 소스를 멈추지 않고 3행 전량이 복구됐습니다. 다만 여기까지 오는 데 트러블슈팅이 2건 있었습니다.

트러블슈팅 1: INSERT command denied

첫 시도는 INSERT command denied for table 'dbz_signal'로 실패했습니다. Debezium 계정은 읽기 전용이 원칙이지만, incremental snapshot은 청크 전후의 기준점이 되는 watermark 행을 Debezium이 signal 테이블에 직접 INSERT하며 진행합니다. 그래서 signal 테이블에 한해서는 쓰기 권한이 필요합니다.

  
docker exec cdc-mysql mysql -uroot -proot \
  -e "GRANT INSERT, UPDATE, DELETE ON operations.dbz_signal TO 'debezium'@'%'; FLUSH PRIVILEGES;"

트러블슈팅 2: 실패한 snapshot이 진행 중 상태로 고착

권한을 고친 뒤 신호를 다시 넣어도 snapshot이 진행되지 않았습니다. 권한 문제로 실패한 snapshot이 상태기계상 진행 중으로 남아 있어서, 신호 재전송만으로는 복구되지 않는 상태였습니다. 복구 절차는 태스크 재시작입니다.

  
curl -s -X POST localhost:8083/connectors/source-operations-mysql/tasks/0/restart
# 오프셋에 저장된 진행 상태를 읽어 자동 재개, 3행 전량 복구 확인
docker exec cdc-postgres psql -U analytics -c 'SELECT count(*) FROM sensor_readings;'

재시작하면 offset에 영속화된 진행 상태를 읽어 자동으로 재개됩니다. 프로세스가 죽어도 offset 기반으로 이어서 진행된다는 장애 복구 메커니즘을 문서가 아니라 실측으로 확인한 셈입니다.

로그 확인에 쓴 명령

검증과 트러블슈팅 내내 두 명령을 반복해서 썼습니다. 하나는 Connect 로그에서 snapshot 진행과 에러를 거르는 명령이고, 하나는 Kafka 토픽에 실제로 쌓인 이벤트를 직접 읽는 명령입니다.

  
docker compose logs connect --since 5m | grep -iE "snapshot|denied|ERROR"
docker exec cdc-kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic ops.operations.sensor_readings \
  --from-beginning --timeout-ms 8000        # Kafka 토픽 내용 직접 확인

sink가 반영을 못 해도 이벤트가 Kafka에는 남아 있는지, 캡처 자체가 안 된 것인지를 이 두 명령으로 구분할 수 있었습니다.

배운 것

DB는 서버와 클라이언트로 나뉩니다. 데이터를 소유한 것은 서버 프로세스이고, mysql이나 psql은 거기에 접속하는 창구일 뿐입니다. docker exec cdc-mysql mysql ... 같은 명령은 container 안의 클라이언트로 같은 container의 서버에 접속하는 것입니다.
CDC는 변경만 나릅니다. 그래서 도입 시점에 이미 쌓여 있던 데이터와 장애로 유실된 데이터는 snapshot이라는 별도 경로가 필요하고, incremental snapshot은 그 경로를 운영 무중단으로 제공합니다.
Debezium과 Kafka의 전달 보장은 at-least-once라서 장애 복구 과정에서 같은 이벤트가 중복 전달될 수 있습니다. 이 중복은 sink의 PK 기준 upsert 멱등성으로 흡수합니다.

다음 편에서는 콘솔에서 수동으로 만든 AWS 리소스를 Terraform으로 코드화한 작업과 프로젝트 마무리를 다룹니다.

다음 글: AWS ML CI/CD와 CDC 파이프라인 구축 (5) Terraform 코드화와 마무리

Machine Learning, ML Projects