Kafka - PostgreSQL to MariaDB (2)

1. What is Debezium?

Debezium은 Change Data Capture를 위한 오픈 소스 분산 플랫폼

Change Data Capture - 변경된 데이터를 캡쳐

소스 시스템에서 데이터가 변경된 것을 감지, 타깃 시스템이 변경 작업에 대응하는 작업을 수행하도록 하는 프로세스

출처:  https://access.redhat.com/documentation/ko-kr/red_hat_integration/2022.q2/html/debezium_user_guide/high-level-overview-of-debezium

 

2. Demonstration

PostgreSQL to MariaDB

• Debezium Connect 구성
• Kafka 클러스터 구성
• PostgreSQL 구성

DB 초기 설정 필요

1. plugin: logical decoding plugin(decoderbufs, pgoutput) 필요

postgresql.conf 파일 수정 필요

# REPLICATION
wal_level = logical  

• decoderbufs

# MODULES
shared_preload_libraries = 'decoderbufs' 

• pgoutput(default로 설치되어있음)

Kafka Connect REST API

get connectors list

curl -i -X GET -H "Accept:application/json" localhost:8083/connectors

check connector status

curl -i -X GET -H "Accept:application/json" localhost:8083/connectors/[커넥터 이름]/status

delete connector

curl -X DELETE localhost:8083/connectors/[커넥터 이름]

Configuration

• Source(PostgreSQL)

 curl --location 'http://localhost:8083/connectors'    --header 'Accept: application/json'    --header 'Content-Type: application/json'    --data ' {
"name": "local-postgres-connector",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"database.hostname": "172.18.0.1", //Database 호스트 IP 주소
"database.port": "45432", //Database 포트
"database.user": "postgres", //Database 사용자 ID
"database.password": "1234", //Database 비밀번호
"database.dbname": "postgres", //Database 이름
"topic.prefix": "from-postgres",
"table.include.list": "public.employee", //담고자 하는 테이블명
"plugin.name": "pgoutput"
}}'

• Sink(MariaDB)

curl --location 'http://localhost:8083/connectors'    --header 'Accept: application/json'    --header 'Content-Type: application/json'    --data '{
  "name": "mariadb-upsert-connector",
  "config": {
    "connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",
    "connection.password": "1234",
    "tasks.max": "1",
    "connection.username": "root",
    "delete.enabled": "true",
    "auto.create": "false",
    "auto.evolve": "true",
    "truncate.enabled": "true",
    "connection.url": "jdbc:mysql://172.18.0.1:43306/receivekafka",
    "insert.mode": "upsert",
    "primary.key.mode": "record_key",  
    "topics": "from-postgres.public.employee",
    "schema.evolution": "basic"

  }}' | jq

 

 

 

 

추가 출처:

https://engkimbs.tistory.com/entry/691

https://velog.io/@seungyeon/카프카-무엇이고-왜-필요할까

https://docs.confluent.io/kafka/introduction.html

https://debezium.io/documentation/reference/stable/connectors/jdbc.html