mysql 데이터 변경 실시간 확인 진행

mysql 데이터 변경을 실시간으로 확인하고 싶어서 search하고 정리했다.

mysql 데이터 변경 실시간 확인

이를 기반으로 실제로 진행해보려고 한다.

우선 나는 docker로 zookeeper, kafka, kafka-connect, mysql을 올릴 것이다

그래서 docker-compose.yml을 아래와 같이 작성했다

docker-compose.yml

version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"
  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    ports:
      - "9092:9092"
  kafka-connect:
    image: debezium/connect:latest
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: "1"
      CONFIG_STORAGE_TOPIC: my_connect_configs
      OFFSET_STORAGE_TOPIC: my_connect_offsets
      STATUS_STORAGE_TOPIC: my_connect_statuses
      KEY_CONVERTER_SCHEMAS_ENABLE: "false"
      VALUE_CONVERTER_SCHEMAS_ENABLE: "false"
    depends_on:
      - kafka
  mysql:
    image: mysql:8.0
    environment:
      MYSQL_ROOT_PASSWORD: root
      MYSQL_DATABASE: robotworld
    ports:
      - "3306:3306"
    volumes:
      - mysql_data:/var/lib/mysql
    command:
      --server-id=1
      --log-bin=mysql-bin
      --binlog-format=ROW
      --binlog-row-image=FULL
volumes:
  mysql_data:
    driver: local

 

mysql

• volumes : 데이터를 유지하기 위해 MySQL 데이터 디렉터리를 호스트 디렉터리 또는 Docker 볼륨에 마운트
• command : MySQL 서버 실행 시 필요한 binlog 설정을 추가
  • --server-id=1 : MySQL 서버의 고유한 ID를 지정
  • --log-bin=mysql-bin : 바이너리 로그 파일의 기본 이름을 지정하여 binlog를 활성화
  • --binlog-format=ROW : 행 기반의 binlog 형식을 설정
  • --binlog-row-image=FULL : 변경된 데이터의 전체 행을 기록

volumes

• mysql_data: MySQL 데이터 파일을 저장할 Docker 볼륨 이름
• driver: local
  • 로컬 드라이버를 사용하여 Docker 볼륨 생성
  • 기본적으로 Docker가 제공하는 로컬 스토리지 사용

mysql에 binlog 활성하기

위 설정으로 생성된 mysql container에 접속해서 정상적으로 binlog가 활성화 되었는지 확인한다.

• docker exec -it [comtainer id] bash
• mysql -u root -p
• SHOW VARIABLES LIKE 'log_bin';
  • ON이면 설정 완료

• SHOW VARIABLES LIKE 'binlog_format';
  • ROW이면 설정 완료

 

Kafka Connect API를 통해 Debezium MySQL 커넥터를 등록

등록하기 전, Kafka Connect가 제대로 실행중인지 확인

curl http://localhost:8083/connectors

 

>> 빈 리스트 혹은 이미 등록된 목록을 반환하면 정상적으로 동작하는 것임

 

Kafka Connect API는 RESTful API이므로 curl 또는 Postman과 같은 도구를 사용하려 HTTP 요청으로 확인한다.

curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
  "name": "mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "root",
    "database.password": "root",
    "database.server.id": "1",
    "database.server.name": "mysql_server",
    "database.whitelist": "robotworld",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.robotworld",
    "include.schema.changes": "true",
    "database.serverTimezone": "UTC",
    "topic.prefix": "mysql_server"
  }
}'

 

얘 보내고, 커넥터 성공하면 이런 응답 받음

{
  "name": "mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "root",
    "database.password": "root",
    "database.server.id": "1",
    "database.server.name": "mysql_server",
    "database.whitelist": "robotworld",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.robotworld",
    "include.schema.changes": "true",
    "database.serverTimezone": "UTC",
    "topic.prefix": "mysql_server",
    "name": "mysql-connector"
  },
  "tasks": [],
  "type": "source"
}

 

커넥터 연결 확인

curl -X GET http://localhost:8083/connectors/mysql-connector/status

 

커넥터 제거

curl -X DELETE http://localhost:8083/connectors/mysql-connector

 

이제 MySQL 데이터베이스에서 변경 사항이 발생하면 해당 내용이 Kafka로 스트리밍 된다.

• Kafka 토픽에서 데이터 확인: MySQL에서 데이터베이스(robotworld)에 삽입, 업데이트 또는 삭제 작업을 수행하면 해당 변경 사항이 Kafka 토픽에 반영. Kafka CLI 또는 Kafka Console Consumer를 사용하여 이 데이터 확인/사용 가능

docker exec -it <kafka-container-id> kafka-console-consumer --bootstrap-server kafka:9092 --topic mysql_server.robotworld.<your_table_name> --from-beginning

• Kafka 토픽 확인: Kafka Console Consumer를 사용하여 데이터가 토픽에 정상적으로 들어오는지 확인 가능 !