CDC(Change Data Capture)

CDC(Change Data Capture)는 데이터베이스의 변경 사항을 실시간으로 추적하고 이를 다른 시스템에 전달하는 기술이다. 이를 통해 데이터베이스 변경 내용을 다른 시스템이나 데이터베이스로 실시간 동기화하거나, 분석용 데이터 웨어하우스로 전송할 수 있다. CDC의 주요 목적은 데이터 일관성 유지와 실시간 데이터 통합을 효율적으로 처리하는 것이다.

데이터 웨어하우스는 여러 소스에서 가져온 구조화된 데이터와 반구조화된 데이터를 분석하고 보고하는 데 사용되는 엔터프라이즈 시스템이다.

 

1. 실시간 데이터 복제

데이터베이스의 삽입, 수정, 삭제와 같은 변경 사항을 실시간으로 캡처하여 다른 시스템으로 전송한다. 이를 통해 데이터 일관성을 유지하고, 여러 시스템 간의 데이터 동기화를 실시간으로 유지할 수 있다.

 

2. 트랜잭션 로그 기반

많은 CDC 솔루션이 데이터베이스 트랜잭션 로그를 읽어 변경 사항을 추적한다. 트랜잭션 로그는 데이터베이스가 변경될 때마다 기록되는 로그로, 데이터 변경 사항을 캡처할 수 있다.

 

3. 이벤트 스트리밍

CDC는 변경 사항을 이벤트로 생성하여 메시지 큐(Kafka, RabbitMQ 등)나 스트리밍 플랫폼을 통해 다른 시스템으로 전달할 수 있다. 이를 통해 데이터 처리를 비동기적으로 수행하고, 확장성을 높일 수 있다.

 

예를 들어, MSA로 운영할 때 서로 다른 마이크로서비스 간의 데이터 동기화 및 일관성을 유지할 수 있고 데이터베이스 변경 사항을 다른 데이터베이스로 복제하여 백업 및 고가용성 시스템을 구축할 수 있다.

 

대표적인 CDC 도구로는 Debezium, Oracle GoldenGate, AWS Database Migration Service(DMS)가 있다. 

• Debezium: 오픈 소스 CDC 플랫폼으로, 여러 데이터베이스(MySQL, PostgreSQL, MongoDB 등)를 지원한다.
• Oracle GoldenGate: Oracle에서 제공하는 CDC 솔루션으로, 다양한 데이터베이스와 시스템을 지원한다.
• AWS Database Migration Service(DMS): AWS에서 제공하는 데이터베이스 마이그레이션 및 CDC 서비스로, 여러 데이터베이스와 클라우드 서비스 간의 실시간 데이터 동기화를 지원한다.

🗣️ ChatGPT 요약
CDC는 데이터베이스의 변경 사항을 실시간으로 캡처하고 이를 다른 시스템으로 전달하는 기술로, 데이터 일관성 유지, 실시간 데이터 통합, 분석 및 마이크로서비스 통합 등에 유용하게 사용됩니다. CDC는 데이터베이스 트랜잭션 로그를 기반으로 하여 신뢰성 있게 변경 사항을 추적하고, 이벤트 스트리밍을 통해 데이터를 전달합니다.

 

만약 Debezium을 사용하여 PostgreSQL 데이터베이스에서 CDC를 구현한다면, 데이터 변경이 일어났을 때 Debezium이 해당 변경 사항을 Kafka로 스트리밍한다.(이 떄 Kafka Connect가 사용된다.) 

 

먼저 Debezium은 초기 설정 시 PostgreSQL 데이터베이스의 전체 상태를 캡처하기 위해 스냅샷을 수행한다. 이 스냅샷은 초기 데이터 상태를 Kafka로 보내는 데 사용된다. 스냅샷은 직접적으로 PostgreSQL 데이터베이스에서 읽어와 Kafka 토픽으로 전송된다.(Kafka 토픽에 저장된다.)

 

이후 PostgreSQL 데이터베이스에서 데이터가 변경될 떄, Debezium 커넥터는 PostgreSQL의 트랜잭션 로그(WAL, Write-Ahead Logging)를 모니터링하여 변경 사항을 캡처한다. 변경 데이터는 Kafka 토픽에 이벤트로 전송되어 저장된다. 각 데이터 변경 이벤트는 Kafka 토픽에 순차적으로 쌓이며, 이후 Kafka 소비자들이 이를 읽어 처리한다.

 

따라서 스냅샷 데이터와 변경 데이터 모두 Kafka 서버의 토픽에 쌓이고 관리된다.

 

참고로 PostgreSQL 데이터베이스 자체에는 기본적으로 CDC 기능이 자동으로 구현되어 있지 않다. 따라서 PostgreSQL 데이터베이스에서 CDC를 구현하려면 직접 Debezium과 같은 도구를 사용하여 설정해야 한다. 

Debezium 설정은 Debezium 커넥터를 설정하여 PostgreSQL 데이터베이스와 연동하고, PostgreSQL의 트랜잭션 로그(WAL)를 읽어 변경 사항을 캡처할 수 있도록 설정한다. 이 때 Debezium 커넥터가 Kafka 토픽으로 PostgreSQL의 변경 사항을 캡처하여 전송할 수 있도록 하려면 Debezium 커넥터와 Kafka Connect를 연동해야 한다.

 

[상세 단계]

1. Kafka 및 Zookeeper 설치

2. Debezium 커넥터를 다운로드하고 Kafka Connect와 연동

3. PostgreSQL 설정

PostgreSQL의 postgresql.conf 파일에서 복제 관련 설정을 활성화한다.

 

wal_level은 PostgreSQL의 WAL(Write-Ahead Logging) 로그의 상세 수준을 정의한다. 'minimal', 'replica', 'logical' 중 하나를 선택할 수 있다.

• minimal: 가장 기본적인 WAL 기록만 남긴다. 성능은 좋지만, PITR(Point-in-Time Recovery)나 복제를 지원하지 않는다.
• replica: 복제를 위한 정보를 포함한 WAL 기록을 남긴다. 기본적인 스트리밍 복제나 아카이브 복제를 지원한다.
• logical: 논리적 복제를 위한 모든 정보를 포함한 WAL 기록을 남긴다. 논리적 복제, Debezium과 같은 CDC 도구에서 사용된다. (Debezium 사용 시 논리적 복제를 지원하기 위해 'logical'로 설정해야 한다. 이는 PostgreSQL이 변경 사항을 논리적 형식으로 기록하여, 외부 시스템(예: Kafka)으로 전송할 수 있도록 한다.)

max_replication_slots 설정은 동시에 활성화될 수 있는 복제 슬롯의 최대 수를 정의한다. 복제 슬롯은 클라이언트가 데이터를 소비하는 것을 보장하고, 데이터가 소비될 때까지 WAL 로그가 삭제되지 않도록 한다. 이 값이 4라면 최대 4개의 복제 슬롯을 동시에 사용할 수 있다. Debezium 사용 시, 이 인스턴스 수에 따라 값을 설정한다.

 

max_wal_senders 설정은 동시에 활성화될 수 있는 WAL 송신기의 최대 수를 정의한다. WAL 송신기는 WAL 로그를 외부 시스템으로 전송하는 프로세스이다. 예를 들어 4로 설정하면 최대 4개의 WAL 송신기를 동시에 사용할 수 있다. 

🗣️ ChatGPT 요약
wal_level = logical: PostgreSQL의 WAL 로그를 논리적 형식으로 기록하여, 논리적 복제와 Debezium과 같은 CDC 도구에서 사용할 수 있도록 설정합니다.
max_replication_slots = 4: 동시에 활성화될 수 있는 복제 슬롯의 최대 수를 정의합니다. Debezium과 같은 도구에서 데이터를 안전하게 소비할 수 있도록 복제 슬롯을 사용합니다.
max_wal_senders = 4: 동시에 활성화될 수 있는 WAL 송신기의 최대 수를 정의합니다. Debezium과 같은 도구에서 PostgreSQL의 변경 사항을 캡처하기 위해 WAL 송신기를 사용합니다.

wal_level = logical
max_replication_slots = 4
max_wal_senders = 4

데이터베이스 사용자에게 필요한 권한을 부여한다.

ALTER USER <your_user> WITH REPLICATION;

4. Debezium 커넥터 구성

Kafka Connect를 통해 Debezium PostgreSQL 커넥터(소스 커넥터)를 설정한다. 아래는 설정 예시이다. 이를 통해 데이터베이스 변경 사항을 캡처하여 Kafka 토픽으로 전송한다. 

 

이 때, 소스 커넥터를 통해 Kafka 토픽으로 전송된다고 하는데 어떤 토픽으로 전송되는지 지정을 하는 부분이 없어 의문이 들었다. Debezium 커넥터는 기본적으로 데이터베이스 이름과 테이블 이름을 기반으로 Kafka 토픽을 자동으로 생성한다고 한다. Debezium이 생성하는 기본 토픽 이름 형식은 다음과 같다. 

 

이 형식이 마음에 들지 않는다면 추가 설정을 통해 토픽 이름을 커스터마이즈할 수 있다고 한다. 또한, Kafka 메시지는 JSON 형식으로, 데이터 변경 이벤트의 키와 값을 포함하여 전송한다.

<database.server.name>.<schema_name>.<table_name>

{
  "name": "inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "localhost",
    "database.port": "5432",
    "database.user": "myuser",
    "database.password": "mypassword",
    "database.dbname": "mydb",
    "database.server.name": "dbserver1",
    "plugin.name": "pgoutput"
  }
}

5. Kafka 토픽으로 데이터 스트리밍

Debezium 커넥터가 PostgreSQL의 변경 사항을 캡처하여 Kafka 토픽으로 전송한다.

Kafka 소비자가 이 토픽을 구독하여 실시간으로 변경 데이터를 처리할 수 있다.

 

이 설정은 PostgreSQL 데이터베이스에서 발생하는 변경 사항을 캡처하여 이를 Kafka로 스트리밍하도록 설정한 것이다.

 

추가적으로, Kafka 토픽의 데이터를 읽어 외부 시스템(예: 또 다른 데이터베이스)에 저장하려면 Kafka 소스 커넥터 설정처럼 싱크 커넥터를 사용하여 데이터베이스, 파일 시스템, 검색 인덱스 등으로 데이터를 전송할 수 있다. 

{
  "name": "jdbc-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "dbserver1.inventory.customers",
    "connection.url": "jdbc:mysql://localhost:3306/otherdb",
    "connection.user": "mysqluser",
    "connection.password": "mysqlpw",
    "insert.mode": "upsert",
    "auto.create": "true",
    "auto.evolve": "true",
    "delete.enabled": "true"
  }
}