대규모 공공데이터 실시간 동기화를 위한 CDC(Change Data Capture)와 Debezium·Kafka 증분 파이프라인 아키텍처

1. 공공데이터 배치 폴링(Polling) 아키텍처의 구조적 한계

수천만 건 규모의 전국 요양기관, 대중교통 노선, 지자체 인허가 데이터 등 대규모 공공데이터를 외부 서비스와 연동할 때 전통적으로 가장 많이 활용되어 온 방식은 주기적 배치 폴링(Periodic Batch Polling)입니다. 일정 시간 간격(예: 1시간 또는 매일 자정)마다 원천 데이터베이스를 대상으로 SELECT * FROM table WHERE updated_at > :last_sync_time 쿼리를 수행하여 증분 데이터를 추출하는 기법입니다.

그러나 서비스 규모가 확장됨에 따라 배치 폴링 방식은 다음과 같은 치명적인 아키텍처 결함을 노출합니다.

• 원천 데이터베이스 리소스 고갈 및 락(Lock) 경합: 대량의 데이터를 스캔하는 풀 테이블 쿼리는 CPU 사용률을 급등시키고 공유 락(Shared Lock) 또는 인덱스 락을 유발하여 실시간 서비스 트랜잭션의 지연(Latency)을 초래합니다.
• 데이터 변경 누락(Silent Missing) 위험: 삭제(DELETE)된 레코드는 updated_at 조건으로 감지할 수 없어 영구적인 정합성 불일치를 유발하며, 긴 트랜잭션이 커밋되기 전에 배치가 실행되면 해당 변경분이 다음 폴링 윈도우에서 영구 누락될 수 있습니다.
• 실시간성 부재(High Latency): 데이터가 실제로 변경된 시점과 배치 주기가 도래하는 시점 사이에 수 분에서 수 시간의 데이터 지연이 발생하여 실시간 의사결정 파이프라인 구축이 불가능합니다.

2. CDC(Change Data Capture) 아키텍처 원리와 트랜잭션 로그 마이닝

Debezium 공식 문서에 명시된 바와 같이, CDC(Change Data Capture)는 데이터베이스에 SQL 질의를 던지지 않고 데이터베이스의 내부 물리 트랜잭션 로그(MySQL의 Binary Log, PostgreSQL의 Write-Ahead Log)를 직접 실시간 스트리밍으로 읽어내는 아키텍처입니다.

CDC의 핵심 동작 메커니즘은 다음과 같습니다.

• 비침해적(Non-intrusive) 로그 스트리밍: 트랜잭션이 커밋되는 즉시 스토리지 엔진이 기록하는 바이너리 로그 이벤트를 소켓 수준에서 직접 소비하므로, 데이터베이스 엔진에 쿼리 파싱이나 인덱스 스캔 부하를 전혀 주지 않습니다.
• 완벽한 변경 이력(Before/After Snapshot): 레코드의 신규 생성(c), 수정(u), 삭제(d)뿐만 아니라 변경 직전 상태(before)와 변경 직후 상태(after)를 모두 캡슐화한 엔벨로프(Envelope) 형태로 전달하여 데이터 복원력을 극대화합니다.
• 순서 보장(Ordered Stream): 바이너리 로그는 트랜잭션이 직렬화된 순서대로 기록되므로, 분산 환경에서도 원천 데이터의 변경 순서가 100% 무결하게 유지됩니다.

3. Debezium과 Apache Kafka를 결합한 분산 이벤트 파이프라인

CDC 아키텍처를 대규모 엔터프라이즈 환경에서 지탱하기 위해 Apache Kafka의 Exactly-Once 의미론과 Debezium 커넥터가 유기적으로 결합됩니다.

• 카프카 파티셔닝과 순서성 보장: Debezium은 원천 테이블의 기본 키(Primary Key)를 Kafka 메시지의 키로 지정합니다. Kafka는 동일한 키를 가진 모든 이벤트를 동일한 토픽 파티션(Partition)으로 라우팅하므로, 특정 공공데이터 엔티티에 대한 변경 이벤트의 선후 관계가 절대 뒤바뀌지 않습니다.
• 오프셋 커밋(Offset Commit)과 장애 복구: Debezium 엔진은 소비한 바이너리 로그의 파일명(file)과 포지션(pos)을 카프카 내부 오프셋 토픽에 지속적으로 저장합니다. 커넥터 장애 발생 시 마지막 커밋 위치부터 정확히 스트리밍을 재개하여 데이터 유실을 방지합니다.
• 이중 쓰기(Dual Write) 문제 원천 차단: 애플리케이션 레벨에서 DB 저장과 카프카 발행을 별도로 수행할 때 발생하는 네트워크 단절 정합성 불일치를 원천적으로 제거하고, 데이터베이스 커밋 자체가 단일 진실 공급원(SSOT) 역할을 수행합니다.

4. 실전 파이프라인 구현: Debezium 엔벨로프 파싱 및 멱등적 싱크 엔진

다음은 카프카 토픽으로 인입되는 Debezium CDC 이벤트를 수신하여 바이너리 로그 오프셋을 검증하고, 레이크하우스 및 검색 인덱스에 멱등적으로 반영하는 스트리밍 컨슈머 구현체입니다.

import time
import json
import logging
from typing import Dict, Any, Optional, List
from dataclasses import dataclass
from enum import Enum

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] (%(name)s) %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger('DebeziumCDCStreamEngine')

class CDCOperation(str, Enum):
    CREATE = 'c'
    UPDATE = 'u'
    DELETE = 'd'
    READ = 'r'

@dataclass
class BinlogOffset:
    file_name: str
    position: int

class CDCEventConsumer:
    def __init__(self, target_sink_table: str):
        self.target_table = target_sink_table
        self.committed_offset = BinlogOffset(file_name='mysql-bin.000001', position=0)
        self.idempotency_set = set()

    def process_envelope(self, event_envelope: Dict[str, Any]) -> bool:
        source_meta = event_envelope.get('source', {})
        binlog_file = source_meta.get('file', '')
        binlog_pos = source_meta.get('pos', 0)
        tx_id = source_meta.get('txId', '')

        # Exactly-Once를 위한 오프셋 멱등성 검증
        if binlog_file == self.committed_offset.file_name and binlog_pos <= self.committed_offset.position:
            logger.warning(f'중복 이벤트 스킵: {binlog_file}:{binlog_pos}')
            return False

        op_type = event_envelope.get('op')
        before_data = event_envelope.get('before')
        after_data = event_envelope.get('after')

        try:
            if op_type == CDCOperation.CREATE.value:
                self._apply_insert(after_data)
            elif op_type == CDCOperation.UPDATE.value:
                self._apply_update(before_data, after_data)
            elif op_type == CDCOperation.DELETE.value:
                self._apply_delete(before_data)
            else:
                logger.info(f'스냅샷 READ 레코드 처리: {after_data.get(\'id\')}')

            # 트랜잭션 오프셋 원자적 커밋 갱신
            self.committed_offset = BinlogOffset(file_name=binlog_file, position=binlog_pos)
            logger.info(f'CDC 동기화 완주 (커밋 오프셋: {binlog_file}:{binlog_pos}, TX: {tx_id})')
            return True

        except Exception as e:
            logger.error(f'CDC 레코드 싱크 실패: {str(e)}')
            self._route_to_dlq(event_envelope, str(e))
            return False

    def _apply_insert(self, record: Dict[str, Any]):
        rec_id = record.get('id')
        logger.info(f'Target DataLake INSERT 반영: ID={rec_id}, Name={record.get(\'facility_name\')}')

    def _apply_update(self, before: Dict[str, Any], after: Dict[str, Any]):
        rec_id = after.get('id')
        logger.info(f'Target DataLake UPDATE 반영: ID={rec_id}, 이전상태={before.get(\'status\')} -> 신규={after.get(\'status\')}')

    def _apply_delete(self, before: Dict[str, Any]):
        rec_id = before.get('id')
        logger.info(f'Target DataLake SOFT DELETE 반영: ID={rec_id}')

    def _route_to_dlq(self, failed_event: Dict[str, Any], reason: str):
        logger.error(f'데드 레터 큐(DLQ) 적재: 이유={reason}')

if __name__ == '__main__':
    consumer = CDCEventConsumer('public_healthcare_facilities')
    sample_cdc_event = {
        'schema': {'type': 'struct'},
        'payload': {
            'before': {'id': 1001, 'facility_name': '서울의료원', 'status': 'OPEN'},
            'after': {'id': 1001, 'facility_name': '서울의료원', 'status': 'MAINTENANCE'},
            'source': {'version': '2.4.0.Final', 'connector': 'mysql', 'file': 'mysql-bin.000003', 'pos': 1542, 'txId': 8849},
            'op': 'u',
            'ts_ms': int(time.time() * 1000)
        }
    }
    consumer.process_envelope(sample_cdc_event['payload'])

5. 성능 분석 및 프로덕션 운영 고려사항

5,000만 건 규모의 공공데이터를 대상으로 배치 폴링 방식과 Debezium CDC 스트리밍 방식의 자원 소모 및 지연 시간을 실측한 벤치마크 결과는 다음과 같습니다.

프로덕션 CDC 파이프라인 운영 시 다음 사항을 반드시 설계에 반영해야 합니다.

• 바이너리 로그 보관 주기(expire_logs_days) 확보: 카프카 클러스터 장애나 컨슈머 지연(Lag)이 길어질 경우를 대비하여 원천 MySQL의 Binlog 보관 주기를 최소 3~7일 이상 확보하여 로그 잘림(Log Truncation) 사고를 방지해야 합니다.
• 스키마 진화(Schema Evolution) 대응: 원천 테이블에 컬럼이 추가되거나 타입이 변경될 때 컨슈머가 역직렬화 오류로 중단되지 않도록 Confluent Schema Registry(Avro/JSON Schema)를 도입하여 상하위 호환성(Full Compatibility)을 강제해야 합니다.
• 초기 스냅샷(Initial Snapshot) 부하 제어: 대용량 테이블 초기 적재 시 snapshot.mode=initial 실행 도중 발생하는 부하를 분산하기 위해 snapshot.select.statement.overrides를 지정하여 청크 단위로 스냅샷을 분할 읽기 해야 합니다.