Debezium을 사용한 Cloud SQL to BigQuery CDC 파이프라인 구축

아키텍처

CDC

CDC란 Change Data Capture로 DBMS에서 데이터가 변경되면 transaction log를 통해 해당 항목을 추적하고 이러한 변경에 대응해야 하는 다른 시스템 및 서비스에 알림을 전송하는 검증된 데이터 통합 패턴이고, Debezium은 대표적인 CDC 툴이다.

Cloud SQL 생성

MySQL로 생성해주고 외부에서 접속을 해주기 위해 접속을 승인할 네트워크 대역을 지정해준다.

해당 ip는 아래에서 생성해줄 내 VM ip를 확인하여 지정해준 것이다.

아래 명령어로 Cloud SQL 접속

workbench와 같은 툴을 사용해도 되지만 어차피 Debezium 서버도 만들어야 하므로 Ubuntu VM을 하나 만들어주고 거기서 접속했다.

# mysql client 설치
apt install mysql-client-core-8.0

# 접속
mysql -h <Cloud SQL ip> -u root p

test DB를 만들어주고 아래 명령어로 people 테이블 생성

create table people (id INT, first_name varchar(45), last_name varchar(45), email varchar(45), zipcode INT, city varchar(45), country varchar(45), __deleted varchar(45));

값을 몇개 넣어줬다.

빅쿼리 테이블, Pub/Sub 생성

그리고 빅쿼리에 Cloud SQL의 DB에 대해 싱크할 테이블을 하나 만들어준다.

Cloud SQL에 있는 스키마와 같게 설정해주는 게 맞지만 해주지 않아볼 것이다.

Pub/Sub을 통해 받아올 구독에서 주제 스키마 사용 옵션을 안하고 빅쿼리에서도 스키마를 지정해주지 않아도 된다는 글을 봤었다.
빅쿼리에 STRING으로 data라는 이름의 열 하나만 넣어주면 스키마 정의 없이도 받아오는 데이터들을 그대로 스키마까지 빅쿼리에 적재한다는 것이었다.

Pub/Sub 스키마 역시 같은 스키마로 생성

{
  "type": "record",
  "name": "Avro",
  "fields": [
    {
      "name": "id",
      "type": "int"
    },
    {
      "name": "first_name",
      "type": "string"
    },
    {
      "name": "last_name",
      "type": "string"
    },
    {
      "name": "email",
      "type": "string"
    },
    {
      "name": "zipcode",
      "type": "int"
    },
    {
      "name": "city",
      "type": "string"
    },
    {
      "name": "country",
      "type": "string"
    },
    {
      "name": "deleted",
      "type": "string"
    }
  ]
}

이 스키마를 갖고 주제를 만들어줄 것인데 여기서 중요한 점이 2가지있다.
1. 주제 이름은 servername.dbname.tablename으로 해줘야 한다. Debezium Server for MySQL이 이 규칙으로 주제를 인식하기 때문이다.
2. 기본 구독 추가는 체크하지 않는다. 왜냐하면 Pub/Sub의 구독 만들기에서 빅쿼리로 push하는 구독을 생성할 수 있기 때문이다.

(주제 스키마 사용 체크를 해제해주자.)
하지만 Pub/Sub의 서비스 계정에 빅쿼리에 쓰는 권한이 없으면 아래와 같은 에라가 난다.

내 프로젝트 IAM에서 빅쿼리에 대한 권한이 없다는 위의 Pub/Sub 서비스 계정에 빅쿼리 편집자 권한을 추가해줬다.

Debezium 설치

Debezium 서버 설치

wget https://repo1.maven.org/maven2/io/debezium/debezium-server-dist/1.9.5.Final/debezium-server-dist-1.9.5.Final.tar.gz

tar xzvf debezium-server-dist-1.9.5.Final.tar.gz

기본으로 application.properties.example이 있는데 그걸 백업용으로 이름을 바꿔주고 새 application.properties 파일 생성

application.properties

debezium.sink.type=pubsub
debezium.sink.pubsub.project.id=<프로젝트 ID>
debezium.source.connector.class=io.debezium.connector.mysql.MySqlConnector
debezium.source.database.hostname=<DB IP>
debezium.source.database.port=3306
debezium.source.database.user=root
debezium.source.database.password=<DB 생성 시 만들어준 PW>
debezium.source.database.server.id=1
debezium.source.database.server.name=<Cloud SQL 인스턴스 이름>
debezium.source.database.include.list=<MySQL DB 이름>
debezium.source.table.include.list=<MySQL DB 이름>.<MySQL 테이블 이름>
debezium.source.database.history=io.debezium.relational.history.FileDatabaseHistory
debezium.source.database.history.file.filename=<history.dat파일 경로>
debezium.source.offset.storage.file.filename=<offsets.dat파일 경로>
debezium.source.offset.flush.interval.ms=0
debezium.source.transforms=unwrap
debezium.source.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
debezium.source.transforms.unwrap.delete.handling.mode=rewrite
debezium.source.key.converter.schemas.enable=false
debezium.source.value.converter.schemas.enable=false

run을 했지만 아래와 같은 오류가 나는 건 java를 설치해주지 않았기 때문.

자바 설치

wget https://download.oracle.com/java/18/latest/jdk-18_linux-x64_bin.tar.gz

tar xvf jdk-18_linux-x64_bin.tar.gz

/etc/profile에서 JAVA_HOME 설정을 해주고 다시 실행했지만 빅쿼리 테이블에 쌓이지 않는다.

application.properties를 참고한 가이드를 따르지 않고 examples로 있었던 형태에서 조금 변형하여 설정해줬다.

application.properties

debezium.sink.type=pubsub
debezium.sink.pubsub.project.id=<프로젝트 ID>
debezium.source.connector.class=io.debezium.connector.mysql.MySqlConnector
debezium.source.offset.storage.file.filename=<offsets.dat파일 경로>
debezium.source.offset.flush.interval.ms=0
debezium.source.database.hostname=<DB IP>
debezium.source.database.port=3306
debezium.source.database.user=root
debezium.source.database.password=<PW>
debezium.source.database.dbname=<DB 이름>
debezium.source.database.server.name=<Cloud SQL 인스턴스 이름>
debezium.source.database.include.list=<DB 이름>
debezium.source.database.include.list=<DB 이름>.<테이블 이름>
quarkus.log.console.json=false

Debezium 서버가 배포된다.

debezium과 CDC에 무지했고 무작정 해봤던 거라.. 에러가 몇 번 발생했었는데 아래엔 개인적으로 발생했던 문제들이 포함되어있다.

에러1

Debezium run 트러블 슈팅 -> 참고
밑에 보이는 io.deb.con.com.BaseSourceTask를 application.properties에 추가해주면 될 듯하다.

아래처럼 추가

debezium.source.database.history=io.deb.con.com.BaseSourceTask

에러2
해결방법을 찾지 못했다.

하지만 신기한 건 나중에 확인해보니 데이터가 적재됐다(?)
어디서 된거지,,

Pub/Sub에서 데이터를 받아올 때 빅쿼리 테이블에 data열이 있으면 JSON으로 떨어지는 데이터가 형식 그대로 쌓이는 것 같다. BigQuery 구독 참고
하지만 형식이 JSON 형태 그대로 위의 사진처럼 떨어지므로 주제 스키마 사용을 해주는 것이 좋아보인다.

[BigQuery로 CDC 파이프라인 구축 참고]

• https://medium.com/google-cloud/streaming-from-google-cloud-pub-sub-to-bigquery-without-the-middlemen-327ef24f4d15
• https://medium.com/dev-genius/realtime-change-data-capture-to-bigquery-no-kafka-no-dataflow-fb4a6994441b