DB to Kafka CDC 환경구성

[dnlpys]

• db의 변동사항을 감지하여 카프카 프로듀싱

• DB환경: postgres - HA Proxy로 환경구성

  • postgres - HA Proxy 설정값

    TZ=Asia/Seoul
    POSTGRESQL_POSTGRES_PASSWORD=
    POSTGRESQL_USERNAME=
    POSTGRESQL_PASSWORD=
    POSTGRESQL_DATABASE=
    POSTGRESQL_NUM_SYNCHRONOUS_REPLICAS=
    REPMGR_PRIMARY_HOST=
    REPMGR_PARTNER_NODES=
    REPMGR_NODE_NAME=
    REPMGR_NODE_NETWORK_NAME=
    REPMGR_USERNAME=
    REPMGR_PASSWORD=
    REPMGR_PRIORITY=

    PGPOOL_BACKEND_NODES=
    PGPOOL_BACKEND_APPLICATION_NAMES=
    PGPOOL_SR_CHECK_USER=
    PGPOOL_SR_CHECK_PASSWORD=
    PGPOOL_ENABLE_LDAP=
    PGPOOL_POSTGRES_USERNAME=
    PGPOOL_POSTGRES_PASSWORD=
    PGPOOL_ADMIN_USERNAME=
    PGPOOL_ADMIN_PASSWORD=
    PGPOOL_ENABLE_LOAD_BALANCING=
    PGPOOL_POSTGRES_CUSTOM_USERS=
    PGPOOL_POSTGRES_CUSTOM_PASSWORDS=
    PGPOOL_AUTO_FAILBACK=

• 기존 kafka-cluster(kraft mode)

  • docker-compose.yaml

    # kafka-1
    dnlpys9999@datapipeline1:~/app/kafka$ cat docker-compose.yaml
    version: '3.8'
    services:
      kafka:
        image: confluentinc/cp-kafka:7.5.3
        hostname: kafka-1
        container_name: kafka-1
        ports:
          - "9092:9092"  # 브로커 포트
          - "9093:9093"  # 컨트롤러 포트
        environment:
          KAFKA_BROKER_ID: 1
          KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://10.178.0.5:9092
          KAFKA_CONTROLLER_QUORUM_VOTERS: "1@10.178.0.5:9093,2@10.178.0.6:9093,3@10.178.0.7:9093"
          KAFKA_PROCESS_ROLES: broker,controller
          KAFKA_LOG_DIRS: /var/lib/kafka/data
          CLUSTER_ID: ${KAFKA_CLUSTER_ID:-MkU3OEVBNTcwNTJENDM2Qk}
          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
          KAFKA_LISTENER_NAMES: PLAINTEXT,CONTROLLER
          KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER  # 추가
        volumes:
          - /home/dnlpys9999/app/kafka-data:/var/lib/kafka/data

    # kafka-2
    dnlpys9999@datapipeline2:~/app/kafka$ cat docker-compose.yaml
    version: '3.8'
    services:
      kafka:
        image: confluentinc/cp-kafka:7.5.3
        hostname: kafka-2
        container_name: kafka-2
        ports:
          - "9092:9092"  # 브로커 포트
          - "9093:9093"  # 컨트롤러 포트
        environment:
          KAFKA_BROKER_ID: 2
          KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://10.178.0.6:9092
          KAFKA_CONTROLLER_QUORUM_VOTERS: "1@10.178.0.5:9093,2@10.178.0.6:9093,3@10.178.0.7:9093"
          KAFKA_PROCESS_ROLES: broker,controller
          KAFKA_LOG_DIRS: /var/lib/kafka/data
          CLUSTER_ID: ${KAFKA_CLUSTER_ID:-MkU3OEVBNTcwNTJENDM2Qk}
          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
          KAFKA_LISTENER_NAMES: PLAINTEXT,CONTROLLER
          KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER  # 추가
        volumes:
          - /home/dnlpys9999/app/kafka-data:/var/lib/kafka/data

    # kafka-3
    dnlpys9999@datapipeline3:~/app/kafka$ cat docker-compose.yaml
    version: '3.8'
    services:
      kafka:
        image: confluentinc/cp-kafka:7.5.3
        hostname: kafka-3
        container_name: kafka-3
        ports:
          - "9092:9092"  # 브로커 포트
          - "9093:9093"  # 컨트롤러 포트
        environment:
          KAFKA_BROKER_ID: 3
          KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://10.178.0.7:9092
          KAFKA_CONTROLLER_QUORUM_VOTERS: "1@10.178.0.5:9093,2@10.178.0.6:9093,3@10.178.0.7:9093"
          KAFKA_PROCESS_ROLES: broker,controller
          KAFKA_LOG_DIRS: /var/lib/kafka/data
          CLUSTER_ID: ${KAFKA_CLUSTER_ID:-MkU3OEVBNTcwNTJENDM2Qk}
          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
          KAFKA_LISTENER_NAMES: PLAINTEXT,CONTROLLER
          KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER  # 추가
        volumes:
          - /home/dnlpys9999/app/kafka-data:/var/lib/kafka/data

각 인스턴스의 docker-compose.yaml 파일에 Kafka Connect 서비스를 추가하고, Debezium PostgreSQL 커넥터를 설정

version: '3.8'
services:
  kafka:
    image: confluentinc/cp-kafka:7.5.3
    hostname: kafka-1
    container_name: kafka-1
    ports:
      - "9092:9092"  # 브로커 포트
      - "9093:9093"  # 컨트롤러 포트
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://10.178.0.5:9092
      KAFKA_CONTROLLER_QUORUM_VOTERS: "1@10.178.0.5:9093,2@10.178.0.6:9093,3@10.178.0.7:9093"
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LOG_DIRS: /var/lib/kafka/data
      CLUSTER_ID: ${KAFKA_CLUSTER_ID:-MkU3OEVBNTcwNTJENDM2Qk}
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_LISTENER_NAMES: PLAINTEXT,CONTROLLER
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER  # 추가
    volumes:
      - /home/dnlpys9999/app/kafka-data:/var/lib/kafka/data
  
  kafka-connect:
    image: confluentinc/cp-kafka-connect:7.5.3
    container_name: kafka-connect
    ports:
      - "8083:8083"  # Kafka Connect REST API
    depends_on:
      - kafka
    environment:
      CONNECT_BOOTSTRAP_SERVERS: "kafka-1:9092,kafka-2:9092,kafka-3:9092"
      CONNECT_GROUP_ID: "kafka-connect-cluster"
      CONNECT_CONFIG_STORAGE_TOPIC: "connect-configs"
      CONNECT_OFFSET_STORAGE_TOPIC: "connect-offsets"
      CONNECT_STATUS_STORAGE_TOPIC: "connect-status"
      CONNECT_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_PLUGIN_PATH: "/usr/share/java"
      CONNECT_LOG4J_LOGGERS: "org.apache.kafka.connect.runtime.rest=WARN,org.reflections=WARN"
    volumes:
      - /home/dnlpys9999/app/kafka-connect-plugins:/usr/share/java
    command:
      - bash
      - -c
      - |
        confluent-hub install --no-prompt debezium/debezium-connector-postgresql:1.9.5 && \
        /etc/confluent/docker/run

1. Kafka 클러스터 상태 확인

1. 브로커 상태 확인
   각 인스턴스에서 Kafka 브로커가 제대로 실행 중인지 확인하려면, 각 서버에서 아래 명령어로 Kafka 주제(topic) 목록을 조회할 수 있습니다:

   docker exec -it kafka-1 kafka-topics --bootstrap-server 10.178.0.5:9092 --list

   각 인스턴스에서 실행하며, Kafka 브로커가 제대로 클러스터에 연결되어 있으면 동일한 주제 목록이 반환되어야 합니다.

2. Kafka 클러스터 메타데이터 확인
   클러스터에 연결된 모든 브로커들을 확인하려면, 다음 명령을 실행합니다:

   docker exec -it kafka-1 kafka-broker-api-versions --bootstrap-server 10.178.0.5:9092

   이 명령을 통해 클러스터에 연결된 각 브로커의 ID와 메타데이터를 확인할 수 있습니다.

2. Kafka Connect 상태 확인

1. Kafka Connect REST API 확인
   각 서버의 Kafka Connect가 제대로 작동하는지 확인하려면, curl 명령어를 사용하여 REST API를 호출할 수 있습니다:

   curl http://10.178.0.5:8083/ | jq

   이 명령을 실행했을 때, Kafka Connect의 상태 정보(JSON 형식)가 반환되면 정상적으로 작동하는 것입니다.

2. Kafka Connect 플러그인 설치 확인
   Kafka Connect에 설치된 Debezium 플러그인 같은 커넥터가 제대로 설치되었는지 확인하려면, 다음 명령어를 실행해 Kafka Connect에 등록된 커넥터를 확인할 수 있습니다:

   curl http://10.178.0.5:8083/connector-plugins | jq

   여기서 PostgreSQL 커넥터와 같은 Debezium 커넥터가 표시되면 설치가 잘 된 것입니다.

[dnlpys]

아직 db와 Kafka Connect 플러그인(Debezium) 연결이 안됨

[dnlpys]

다음 진행 단계

• PostgreSQL에서 Logical Replication 활성화

  • PostgreSQL에서 CDC 기능을 사용하기 위해 logical replication을 활성화
  • pg_hba.conf 파일을 수정하여 logical replication이 가능한 사용자에게 권한을 부여

• Debezium PostgreSQL Connector 설정

  • Kafka Connect REST API를 사용하여 PostgreSQL 커넥터를 설정

  # example
  {
    "name": "postgresql-connector",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "tasks.max": "1",
      "database.hostname": "server2-ip", 
      "database.port": "5432",
      "database.user": "your_replication_user",
      "database.password": "your_replication_password",
      "database.dbname": "your_db",
      "database.server.name": "dbserver1",
      "plugin.name": "pgoutput",
      "slot.name": "debezium_slot",
      "publication.name": "db_publication",
      "database.history.kafka.bootstrap.servers": "kafka-1:9092,kafka-2:9092,kafka-3:9092",
      "database.history.kafka.topic": "schema-changes.dbserver1"
    }
  }

  curl -X POST -H "Content-Type: application/json" --data @postgresql-connector.json http://server1:8083/connectors

• PostgreSQL에서 Publication 설정

  • PostgreSQL 마스터 서버(server2)에 접속하여 publication을 생성

• Kafka Topic 설정 및 데이터 스트리밍 확인

  • Kafka에서 Debezium이 스트리밍하는 데이터는 특정 토픽 확인

  kafka-console-consumer --bootstrap-server kafka-1:9092 --topic dbserver1.public.your_table_name --from-beginning

• Kafka Connect Cluster 모니터링

  • Kafka Connect REST API나 Kafka Connect UI(Kafka Manager)에서 커넥터 상태를 모니터

[dnlpys]

• 기존 gcp 환경에서 aws 환경으로 변경 완료
• kafka-connect와 CDC플러그인(debezium) 설치완료
  => 기존에는 jar파일을 직접 복사하는 방식이었으나, Dockerfile을 빌드하여 confluent-hub로 pull하여 디펜던시까지 해결
• debezium이 db을 복제할수 있게 logical replication을 활성화하는 단계에서 트러블슈팅중
  --> postgresql.conf에서 wal_level = logical 변경해야 logical replication활성화를 할수 있지만 변경하면 컨테이너가 무한 로딩
  --> 네트워크 재설정 시도
  --> host 모드 시도
  --> pgpool > postgres-0 연결 O
  --> pgpool > postgres-1 연결 X
  --> postgres-1에 접속해 repmgr 사용자가 올바르게 설정되어 있는지 확인
  --> pgpool.conf 설정에서 sr_check_user와 sr_check_password 설정이 올바르게 되어 있는지 다시 확인
  --> pgpool컨테이너 접속후 postgres-1 repmgr계정 접속은 성공

docker exec -it pgpool psql -U repmgr -h 172.31.0.17 -d postgres

[dnlpys]

• db의 변경사항을 CDC기능으로 kafka클러스터에 producing
• kafka_connect plugin Debezium으로 CDC기능구현
• 구성환경: single kafka클러스터(Kraft mode), kafka_connet-debezium, postgres

# docker-compose.yaml
version: '3'
services:
  # Kafka (KRaft 모드)
  kafka:
    image: confluentinc/cp-kafka:7.5.3
    container_name: kafka
    user: "1000:1000"  # 호스트의 사용자 ID와 그룹 ID로 변경하세요
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:9093'
      KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:9092'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT'
      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_LOG_DIRS: '/var/lib/kafka/data'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
      CLUSTER_ID: 'jaShGO9YR12vntLHlWQQBA'
    ports:
      - "9092:9092"
    volumes:
      - ./kafka_data:/var/lib/kafka/data
    networks:
      - kafka-network

  # Kafka Connect와 Debezium (KRaft 모드)
  kafka-connect:
    image: debezium/connect:1.9
    container_name: kafka-connect
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: 'kafka:9092'
      GROUP_ID: "connect-cluster"
      CONFIG_STORAGE_TOPIC: "connect-configs"
      OFFSET_STORAGE_TOPIC: "connect-offsets"
      STATUS_STORAGE_TOPIC: "connect-status"
      KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE: "false"
      CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: "false"
      INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_INTERNAL_KEY_CONVERTER_SCHEMAS_ENABLE: "false"
      CONNECT_INTERNAL_VALUE_CONVERTER_SCHEMAS_ENABLE: "false"
      OFFSET_FLUSH_INTERVAL_MS: 10000
      PLUGIN_PATH: "/usr/share/java"
      REST_ADVERTISED_HOST_NAME: kafka-connect
    depends_on:
      - kafka
    networks:
      - kafka-network

  # PostgreSQL
  postgres:
    image: postgres:15
    container_name: postgres
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgrespass
      POSTGRES_DB: inventory
    ports:
      - "5432:5432"
    volumes:
      - ./postgres_data:/var/lib/postgresql/data
      - ./postgres.conf:/etc/postgresql/postgresql.conf
      - ./init_scripts:/docker-entrypoint-initdb.d  # 스크립트를 실행하기 위해 init_scripts 추가
    command: ["postgres", "-c", "config_file=/etc/postgresql/postgresql.conf"]
    networks:
      - kafka-network
networks:
  kafka-network:
    driver: bridge

Debezium이 db상태를 확인하기 위한 설정값변경

• wal_level = logical 로 변경해줘야 논리적 복제를 해서 상태비교를 할수 있음
• 클러스터의 경우, max_wal_senders, max_replication_slots 값을 4이상으로 줄것

# postgres.conf
listen_addresses = '*'
wal_level = logical
max_wal_senders = 1
max_replication_slots = 1

외부 접속 허용을 위해 pg_hba.conf 수정

# pg_hba.conf

#!/bin/bash
echo "host    all    all    0.0.0.0/0    scram-sha-256" >> /var/lib/postgresql/data/pg_hba.conf

CDC 테스트

• DB생성

  # 테이블 생성
  CREATE TABLE public.folder_log (
      id int4 GENERATED ALWAYS AS IDENTITY (INCREMENT BY 1 MINVALUE 1 MAXVALUE 2147483647 START 1 CACHE 1 NO CYCLE) NOT NULL,
      "path" varchar(255) NOT NULL,
      created_at timestamp NOT NULL,
      CONSTRAINT folder_log_pkey PRIMARY KEY (id)
  );
  
  # 데이터 삽입
  INSERT INTO public.folder_log ("path", created_at) VALUES ('/path/to/folder1', '2024-09-12 12:34:56'); 

• 복제 슬롯(Replication Slot)

  • PostgreSQL에서 데이터베이스의 변경사항을 추적하기 위해 사용되는 메커니즘

  • 복제 슬롯(Replication Slot) 종류

    • 물리적 슬롯 (Physical Slot):
      • 물리적 슬롯은 WAL(Write Ahead Log) 기록을 사용하여 데이터베이스의 변경사항을 복제합니다.
      • 주로 고가용성(HA) 설정이나 데이터베이스 클러스터 간의 데이터 복제에 사용됩니다.
      • wal_levle=“replica”
    • 논리적 슬롯 (Logical Slot):
      • 논리적 슬롯은 데이터베이스의 변경사항을 논리적 형태로 제공하여, 특정 테이블의 변경사항만을 선택적으로 복제할 수 있습니다.
      • Debezium과 같은 CDC(Change Data Capture) 도구에서 주로 사용됩니다.
      • 논리적 슬롯은 복제 시 before와 after 이미지(변경 전후 데이터)를 제공합니다.
      • wal_levle="logical"

  • 슬롯 플러그인(Slot PlugIn)

    1. pgoutput 플러그인

    • 용도: PostgreSQL의 기본 논리적 복제 시스템에서 사용됩니다.
    • 기능: 데이터베이스 변경 사항을 실시간으로 전송하지만, **이전 상태(before image)**를 제공하지 않습니다. 즉, 데이터가 수정되었을 때 그 수정 이전의 데이터를 알 수 없습니다.
    • 사용 예: 주로 데이터 복제 및 HA(고가용성) 시스템에서 사용됩니다.

    2. test_decoding 플러그인

    • 용도: 논리적 복제를 위한 테스트 및 실험 목적으로 설계된 플러그인입니다.
    • 기능: 데이터 변경 사항을 전송할 때 **이전 상태(before image)**와 **변경된 상태(after image)**를 모두 제공합니다. 즉, 데이터를 수정할 때 변경 이전의 값과 이후의 값을 모두 캡처할 수 있습니다.
    • 사용 예: CDC(Change Data Capture) 도구, 예를 들어 Debezium과 함께 사용되어 데이터의 변화를 더 세밀하게 추적할 수 있도록 합니다.

  • 현재 구성으로는 ‘test_decoding’ 옵션 사용 불가

  • 커넥터 생성시 "plugin.name" , "slot.name" 옵션으로 자동 생성 시킬수 있음.

  # 슬롯 생성
  # SELECT pg_create_logical_replication_slot('slot_name', 'slot_type');
  docker exec -it postgres psql -U postgres -d inventory -c "SELECT pg_create_logical_replication_slot('debezium_slot', 'test_decoding');"
  
  # 슬롯 확인
  # 여러 슬롯이 있을경우, active컬럼이 t인것이 활성화
  docker exec -it postgres psql -U postgres -d inventory -c "SELECT * FROM pg_replication_slots;"
  
  # 슬롯 삭제
  docker exec -it postgres psql -U postgres -d inventory -c "SELECT pg_drop_replication_slot('debezium_slot');"

• debezium 커넥터

  # 커넥터 생성(connector name: inventory-connector)
  curl -X POST -H "Content-Type: application/json" \
  -d '{
    "name": "inventory-connector",  # 커넥터의 이름
    "config": {  # 커넥터의 설정
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",  # 사용할 커넥터 클래스
      "tasks.max": "1",  # 최대 태스크 수
      "database.hostname": "postgres",  # PostgreSQL 서버의 호스트명
      "database.port": "5432",  # PostgreSQL 서버의 포트
      "database.user": "postgres",  # 데이터베이스 사용자 이름
      "database.password": "postgrespass",  # 데이터베이스 비밀번호
      "database.dbname": "inventory",  # 데이터베이스 이름
      "database.server.name": "dbserver1",  # Kafka 내에서 식별할 서버 이름
      "plugin.name": "test_decoding",  # 사용할 플러그인 (before 이미지 수집을 위해)
      "table.whitelist": "public.folder_log",  # CDC를 수행할 테이블
      "slot.name": "debezium_slot",  # 사용될 복제 슬롯의 이름
      "database.history.kafka.bootstrap.servers": "localhost:9092",  # Kafka의 Bootstrap 서버 주소
      "database.history.kafka.topic": "dbhistory.inventory"  # 데이터베이스 변경 이력을 저장할 Kafka 주제
    }
  }' \
  http://localhost:8083/connectors  # Kafka Connect API 엔드포인트
  
  curl -X POST -H "Content-Type: application/json" \
  -d '{
    "name": "inventory-connector",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "tasks.max": "1",
      "database.hostname": "postgres",
      "database.port": "5432",
      "database.user": "postgres",
      "database.password": "postgrespass",
      "database.dbname": "inventory",
      "database.server.name": "dbserver1",
      "plugin.name": "pgoutput",
      "table.whitelist": "public.folder_log",
      "slot.name": "debezium_slot",
      "database.history.kafka.bootstrap.servers": "localhost:9092",
      "database.history.kafka.topic": "dbhistory.inventory"
    }
  }' \
  http://localhost:8083/connectors
  
  # 커넥터 삭제
  curl -X DELETE http://localhost:8083/connectors/inventory-connector
  
  # 커넥터 생태확인
  curl -X GET http://localhost:8083/connectors/inventory-connector/status

  

  커넥터가 RUNNING 상태인지 확인

• Kafka 확인

  • kafka consumer로 결과 확인

  docker exec -it kafka kafka-console-consumer --bootstrap-server localhost:9092 \
    --topic dbserver1.public.customers \
    --from-beginning

  

[Kdreamtomaster]

{"before":null,"after":{"id":3,"path":"/version3","created_at":1726144496000000},"source":{"version":"1.9.7.Final","connector":"postgresql","name":"dbserver1","ts_ms":1727077772859,"snapshot":"false","db":"inventory","sequence":"["26837704","26838096"]","schema":"public","table":"folder_log","txId":743,"lsn":26838096,"xmin":null},"op":"c","ts_ms":1727077773153,"transaction":null}

[Kdreamtomaster]

topic : dbserver1.public.folder_log

[wwkler]

• Spark에서 받았을 떄

spark가 kafka에 있는 데이터 consuming

kafka_df = spark
.read
.format("kafka")
.option("kafka.bootstrap.servers", bootstrap_servers)
.option("subscribe", topic_name)
.option("startingOffsets", "earliest")
.load()

value 컬럼을 문자열로 변환

decoded_df = kafka_df.selectExpr("CAST(value AS STRING)").select("value").collect()

decoded_df를 print 했을 떄

[
Row(value='{"before":null,"after":{"id":1,"path":"/path/to/folder1","created_at":1726144496000000},"source":{"version":"1.9.7.Final","connector":"postgresql","name":"dbserver1","ts_ms":1727077005971,"snapshot":"last","db":"inventory","sequence":"[null,\"26837024\"]","schema":"public","table":"folder_log","txId":741,"lsn":26837024,"xmin":null},"op":"r","ts_ms":1727077006101,"transaction":null}'),

Row(value='{"before":null,"after":{"id":2,"path":"/path/to/folder1","created_at":1726144496000000},"source":{"version":"1.9.7.Final","connector":"postgresql","name":"dbserver1","ts_ms":1727077140520,"snapshot":"false","db":"inventory","sequence":"[null,\"26837280\"]","schema":"public","table":"folder_log","txId":742,"lsn":26837280,"xmin":null},"op":"c","ts_ms":1727077140906,"transaction":null}'),

Row(value='{"before":null,"after":{"id":3,"path":"/version3","created_at":1726144496000000},"source":{"version":"1.9.7.Final","connector":"postgresql","name":"dbserver1","ts_ms":1727077772859,"snapshot":"false","db":"inventory","sequence":"[\"26837704\",\"26838096\"]","schema":"public","table":"folder_log","txId":743,"lsn":26838096,"xmin":null},"op":"c","ts_ms":1727077773153,"transaction":null}'),

Row(value='{"before":null,"after":{"id":4,"path":"/version4","created_at":1726922096000000},"source":{"version":"1.9.7.Final","connector":"postgresql","name":"dbserver1","ts_ms":1727142393377,"snapshot":"false","db":"inventory","sequence":"[\"26838584\",\"26839152\"]","schema":"public","table":"folder_log","txId":744,"lsn":26839152,"xmin":null},"op":"c","ts_ms":1727142393583,"transaction":null}')
]

[Kdreamtomaster]

[
    {
        “before”: null,
        “after”: {
          “id”: 1,
          “path”: “/path/to/folder1”,
          “created_at”: 1726144496000000
       },
       “source”: {
          “version”: “1.9.7.Final”,
          “connector”: “postgresql”,
          “name”: “dbserver1”,
          “ts_ms”: 1727077005971,
          “snapshot”: “last”,
          “db”: “inventory”,
          “sequence”: “[null,\“26837024\”]”,
          “schema”: “public”,
          “table”: “folder_log”,
          “txId”: 741,
          “lsn”: 26837024,
          “xmin”: null
       },
       “op”: “r”,
       “ts_ms”: 1727077006101,
       “transaction”: null
   },
   {
       “before”: null,
       “after”: {
         “id”: 2,
         “path”: “/path/to/folder1”,
         “created_at”: 1726144496000000
      },
      “source”: {
         “version”: “1.9.7.Final”,
         “connector”: “postgresql”,
         “name”: “dbserver1”,
         “ts_ms”: 1727077140520,
         “snapshot”: false”,
         ”db“: ”inventory“，
         ”sequence“：[null，\“26837280\“]”，
         ”schema“：”public“，
         ”table“：”folder_log“，
         ”txId“：742，
         ”lsn“：26837280，
         ”xmin“：null
     }，
     ”op“：”c“，
     ”ts_ms“：1727077140906，
     ”transaction“：null
   }
]

[Kdreamtomaster]

배치 작업에서 CDC를 사용하는 이유

배치 작업에서 Kafka DB CDC(Change Data Capture)를 사용하는 경우는 주로 대량의 데이터 변경 사항을 실시간으로 처리하고, 데이터 일관성을 유지하며, 시스템 간의 데이터 동기화를 효율적으로 수행하기 위해서입니다. 다음과 같은 주요 이유와 사례가 있습니다:

Kafka DB CDC의 사용 사례

• 실시간 데이터 스트리밍: Kafka CDC는 데이터베이스의 변경 사항을 실시간으로 감지하여 스트리밍 데이터 파이프라인을 구축할 수 있습니다. 이를 통해 실시간 분석, 모니터링, 이벤트 기반 마이크로서비스를 구현할 수 있습니다[1][3].

• 데이터 일관성 유지: CDC를 통해 데이터베이스의 변경 사항을 다른 시스템에 전파하여 데이터 일관성을 유지할 수 있습니다. 이는 특히 여러 시스템 간의 데이터 동기화가 필요한 경우 유용합니다[1][6].

• 대용량 데이터 마이그레이션: 대량의 데이터를 다른 데이터베이스로 마이그레이션할 때, CDC를 사용하면 변경된 데이터를 지속적으로 추적하고 전송할 수 있어 효율적인 데이터 이동을 지원합니다[2][8].

• ETL 프로세스 최적화: 기존의 배치 기반 ETL(Extract, Transform, Load) 프로세스를 실시간으로 전환하여 더 빠르고 효율적인 데이터 처리 환경을 구축할 수 있습니다. 이는 특히 빠른 데이터 처리가 필요한 환경에서 유리합니다[8].

Kafka DB CDC의 장점

• 확장성과 내결함성: Kafka는 분산 시스템으로 설계되어 있어 대규모 데이터를 처리하는 데 적합하며, 시스템 장애에도 견딜 수 있는 내결함성을 제공합니다[3][6].

• 실시간 처리: Kafka는 데이터를 실시간으로 스트리밍하여 즉각적인 반응과 의사결정을 가능하게 합니다. 이는 비즈니스 민첩성을 높이고 경쟁력을 강화하는 데 기여합니다[1][3].

• 데이터 손실 방지: Kafka는 데이터를 디스크에 영구적으로 저장하므로, 데이터 손실 없이 안정적으로 데이터를 처리하고 보관할 수 있습니다[3].

배치 작업에서 Kafka DB CDC를 활용하면 실시간 데이터 처리와 시스템 간 데이터 일관성 유지가 가능해져, 다양한 비즈니스 요구사항에 신속하게 대응할 수 있는 유연한 데이터 아키텍처를 구축할 수 있습니다.

Citations:
[1] https://risingwave.com/blog/unlocking-kafka-change-data-capture-a-beginners-guide/
[2] https://hyperconnect.github.io/2021/06/21/cdc-event-application-consuming.html
[3] https://pinggoopark.tistory.com/869
[4] https://hyperconnect.github.io/2021/01/11/cdc-platform.html
[5] https://breezymind.com/kafka-connect-basic/
[6] https://oceanobe.com/news/using-kafka-to-stream-change-data-capture-data-between-databases/984
[7] https://velog.io/%40jihwankim94/Kafka-Debezium-Postgresql-Source-Connector-%EB%A5%BC-%ED%99%9C%EC%9A%A9%ED%95%9C-CDC
[8] https://techblog.lycorp.co.jp/ko/migrating-large-data-with-kafka-and-etl

[dnlpys]

Debezium이 db상태를 확인하기 위한 설정값변경( 수정 및 추가 설명)

# 변경사항
# postgres.conf
listen_addresses = '*'
wal_level = logical
max_wal_senders = 4
max_replication_slots = 4

추가설명

wal_level 설정은 PostgreSQL의 Write-Ahead Logging (WAL) 기능의 동작 방식을 지정( 3가지 옵션 값을 지정할수 있음)

1. minimal:

• 설명: 이 모드는 WAL 기록을 최소화하여 성능을 최적화합니다. 기본적으로 데이터베이스 복구를 위한 최소한의 정보를 기록합니다.
• 사용 사례: 단순한 데이터베이스와 일반적인 트랜잭션 환경에서 사용할 수 있습니다. 복제 및 복구가 필요 없는 상황에서 최적입니다.

2. replica:

• 설명: 이 모드는 기본적인 스트리밍 복제를 지원합니다. WAL 로그에 복제와 관련된 정보를 기록하여 물리적 복제(예: 슬레이브 서버로 데이터 전송)를 가능하게 합니다.
• 사용 사례: 기본적인 데이터베이스 복제 및 고가용성을 제공하는 환경에서 사용합니다. 슬레이브 서버에서 데이터를 복제하는 데 필요한 정보를 기록합니다.

3. logical:

• 설명: 이 모드는 논리적 복제와 변경 데이터 캡처(CDC)를 지원합니다. WAL에 더 많은 정보를 기록하여 데이터 변경 사항을 추적할 수 있습니다.
• 사용 사례: Debezium과 같은 CDC 도구를 사용하여 데이터를 다른 시스템으로 스트리밍하거나 논리적 복제를 구현하는 환경에서 필요합니다.

---

max_wal_senders:

• PostgreSQL에서 WAL(Write-Ahead Logging) 데이터를 전송하는 프로세스의 최대 개수를 설정합니다.
• WAL 데이터를 전송하는 데에는 복제 작업이나 Debezium의 CDC 작업이 포함됩니다. CDC 작업을 위해 Debezium은 WAL 데이터를 사용하므로, 여러 프로세스가 동시에 WAL 데이터를 사용할 수 있게 하려면 이 값이 높아야 합니다.

max_replication_slots:

• 복제 슬롯의 최대 개수를 설정하는 값입니다. Debezium은 PostgreSQL에서 **복제 슬롯(replication slot)**을 사용하여 변경 로그를 유지하고 관리합니다. 이 값을 늘려야 여러 CDC 작업 또는 여러 커넥터가 안정적으로 실행될 수 있습니다.
• 복제 슬롯이 부족하면 CDC 작업을 제대로 수행할 수 없거나 추가적인 복제 슬롯을 생성할 수 없게 됩니다.