이 문서에서는 Kafka Connect용 PostgreSQL용 Cloud SQL 소스 커넥터를 만드는 방법을 설명합니다.
PostgreSQL용 Cloud SQL 소스 커넥터는 Debezium PostgreSQL 커넥터의 인스턴스입니다. PostgreSQL용 Cloud SQL 데이터베이스에서 행 수준 변경사항을 읽고 Managed Service for Apache Kafka 클러스터의 주제에 씁니다.
이 커넥터의 사용 사례는 다음과 같습니다.
- 행 수준 데이터베이스 변경사항을 실시간으로 모니터링합니다.
- 데이터베이스 변경 이벤트를 이벤트 기반 아키텍처에 통합합니다.
- 행 삽입 또는 삭제와 같은 데이터베이스 이벤트에 응답합니다.
- 데이터베이스 변경사항을 다른 시스템에 복사합니다.
시작하기 전에
PostgreSQL용 Cloud SQL 소스 커넥터를 만들기 전에 다음이 있는지 확인하세요.
데이터베이스가 있는 PostgreSQL용 Cloud SQL 인스턴스 이러한 리소스를 만드는 방법을 알아보려면 Google Cloud 콘솔을 사용하여 PostgreSQL용 Cloud SQL 데이터베이스 만들기 및 쿼리하기를 참고하세요.
필수 역할 및 권한
커넥터를 만드는 데 필요한 권한을 얻으려면 관리자에게 프로젝트에 대한 관리형 Kafka 커넥터 편집자 (roles/managedkafka.connectorEditor) IAM 역할을 부여해 달라고 요청하세요.
역할 부여에 대한 자세한 내용은 프로젝트, 폴더, 조직에 대한 액세스 관리를 참조하세요.
이 사전 정의된 역할에는 커넥터를 만드는 데 필요한 권한이 포함되어 있습니다. 필요한 정확한 권한을 보려면 필수 권한 섹션을 펼치세요.
필수 권한
커넥터를 만들려면 다음 권한이 필요합니다.
-
커넥터를 만듭니다.
managedkafka.connectors.create
커스텀 역할이나 다른 사전 정의된 역할을 사용하여 이 권한을 부여받을 수도 있습니다.
Cloud SQL에서 읽기 권한 부여
관리형 Kafka 서비스 계정에는 PostgreSQL용 Cloud SQL에 액세스할 수 있는 권한이 있어야 합니다. 서비스 계정에 다음 IAM 역할을 부여합니다.
- Cloud SQL 클라이언트 (
roles/cloudsql.client) - Cloud SQL 인스턴스 사용자 (
roles/cloudsql.instanceUser)
관리형 Kafka 서비스 계정의 형식은 다음과 같습니다.
service-PROJECT_NUMBER@gcp-sa-managedkafka.,
여기서 PROJECT_NUMBER은 Connect 클러스터의 프로젝트 번호입니다.
Connect 클러스터가 Managed Service for Apache Kafka 클러스터와 다른 프로젝트에 있는 경우 다른 프로젝트에서 Connect 클러스터 만들기를 참고하세요.
데이터베이스 구성
커넥터를 만들기 전에 데이터베이스 복제를 구성하고 커넥터가 데이터베이스로 인증할 수 있도록 해야 합니다. 다음 섹션에서는 이러한 단계를 설명합니다.
논리 디코딩 사용 설정
PostgreSQL용 Cloud SQL 소스 커넥터는 PostgreSQL의 논리적 디코딩 기능을 사용합니다. PostgreSQL용 Cloud SQL 인스턴스에서 논리적 디코딩을 사용 설정하려면 다음 단계를 따르세요.
콘솔
Cloud SQL > 인스턴스로 이동합니다.
인스턴스의 이름을 클릭합니다.
수정을 클릭합니다.
플래그 및 매개변수를 펼칩니다.
데이터베이스 플래그 추가를 클릭합니다.
플래그 선택 목록에서
cloudsql.logical_decoding을 선택합니다.값으로
On을 선택합니다.완료를 클릭합니다.
저장을 클릭합니다.
자세한 내용은 논리 복제 및 디코딩 설정을 참고하세요.
변경 데이터 캡처 (CDC) 구성
인스턴스에서 논리적 디코딩을 사용 설정한 후 복제할 테이블에 변경 데이터 캡처(CDC)를 사용 설정합니다.
테이블에서 CDC를 사용 설정하려면 CREATE PUBLICATION SQL 문을 실행합니다. 이 문은 복제할 테이블 그룹을 정의하는 게시물을 만듭니다.
옵션 1. 데이터베이스의 모든 테이블에 대한 변경사항을 복제하는 게시를 만듭니다.
CREATE PUBLICATION dbz_publication FOR ALL TABLES;옵션 2. 특정 테이블 집합에 대한 게시를 만듭니다.
CREATE PUBLICATION dbz_publication FOR TABLE TABLE_LIST;TABLE_LIST를"schema_name"."table_name"형식의 쉼표로 구분된 목록으로 바꿉니다. 스키마와 테이블 이름을 다음과 같이 큰따옴표로 묶으면 이름에 특수문자나 대문자가 포함된 경우 구문 오류를 방지할 수 있습니다.
기본적으로 커넥터는 게시 이름에 dbz_publication을 사용합니다. 이름이 다른 간행물을 사용하려면 간행물 이름을 참고하세요.
관리형 Kafka 서비스 계정의 사용자 계정 만들기
PostgreSQL용 Cloud SQL 소스 커넥터는 IAM 데이터베이스 인증을 사용하여 데이터베이스에 연결합니다. IAM 데이터베이스 인증을 사용 설정하려면 다음과 같이 관리 Kafka 서비스 계정을 Cloud SQL 인스턴스에 추가합니다.
콘솔
Cloud SQL > 인스턴스로 이동
인스턴스의 이름을 클릭합니다.
탐색 창에서 사용자를 클릭합니다.
사용자 계정 추가를 클릭합니다.
사용자 계정 추가 창에서 Cloud IAM을 선택합니다.
IAM 주 구성원 필드에 다음을 입력합니다.
service-PROJECT_NUMBER@gcp-sa-managedkafka.PROJECT_NUMBER를 연결 클러스터의 프로젝트 번호로 바꿉니다.추가를 클릭합니다.
gcloud
gcloud sql users create 명령어를 실행합니다.
gcloud sql users create service-PROJECT_NUMBER@gcp-sa-managedkafka.iam \
--instance=INSTANCE_NAME \
--type=cloud_iam_service_account
다음을 바꿉니다.
PROJECT_NUMBER: Connect 클러스터의 프로젝트 번호입니다.INSTANCE_NAME: PostgreSQL용 Cloud SQL 인스턴스의 이름입니다.
데이터베이스 사용자 이름의 길이 제한으로 인해 . 서픽스가 사용자 이름에서 삭제되므로 사용자 이름은 service-PROJECT_NUMBER@gcp-sa-managedkafka.iam입니다. IAM 사용자 계정을 참조하는 SQL 쿼리를 실행할 때는 잘린 이름을 지정하세요.
사용자 계정 구성
IAM 사용자 계정을 만든 후 cloudsqlsuperuser 역할이 있는 사용자 (예: 기본 postgres 사용자)로 데이터베이스에 연결하고 다음 SQL 쿼리를 실행합니다.
콘솔
사용자가 미리 쓰기 로그를 읽을 수 있도록 합니다.
ALTER USER "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam" WITH REPLICATION;사용자에게 테이블에 대한
SELECT권한을 부여합니다.GRANT SELECT ON ALL TABLES IN SCHEMA "SCHEMA_NAME" TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";또는 개별 테이블에
SELECT권한을 부여할 수 있습니다. 이 옵션을 선택하는 경우 커넥터의table.include.list구성 속성을 허용된 테이블 목록으로 설정해야 합니다. 다음 SQL 쿼리는 단일 테이블에SELECT권한을 부여합니다.GRANT SELECT ON TABLE SCHEMA_NAME.TABLE_NAME TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";각 테이블에 대해 사용자에게 테이블의 스키마에 대한 액세스 권한을 부여합니다. 표가 기본
public스키마에 있으면 이 단계를 건너뛸 수 있습니다.GRANT USAGE ON SCHEMA SCHEMA_NAME TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";
네트워킹 구성
PostgreSQL용 Cloud SQL 소스 커넥터는 다음과 같은 방법으로 Cloud SQL 인스턴스에 연결할 수 있습니다.
- 비공개 IP
- Private Service Connect
- 공개 IP
이러한 옵션에 대한 자세한 내용은 Cloud SQL에 연결하는 방법 선택을 참고하세요. 보안 권장사항에 따라 비공개 IP 또는 Private Service Connect를 사용하는 것이 좋습니다. 이러한 옵션은 외부 IP 주소에 연결할 필요가 없기 때문입니다.
다음 표에는 각 옵션의 네트워크 요구사항이 나와 있습니다.
| IP 주소 유형 | 요구사항 |
|---|---|
| 비공개 IP | 인스턴스의 비공개 IP를 구성합니다. 자세한 내용은 비공개 IP 구성을 참고하세요. |
| Private Service Connect |
|
| 공개 IP |
|
PostgreSQL용 Cloud SQL 소스 커넥터 만들기
PostgreSQL용 Cloud SQL 소스 커넥터를 만들려면 다음 단계를 따르세요.
커넥터가 초기화되면 다음 작업을 실행합니다.
- 데이터베이스의 초기 스냅샷을 만듭니다.
- 행이 있는 모든 테이블에 대해 Kafka 주제를 만듭니다.
- 각 데이터베이스 행에 대해 해당 주제로 변경 이벤트를 전송합니다.
커넥터가 실행되는 동안 변경 이벤트가 주제로 계속 전송됩니다. 초기 스냅샷에 대한 자세한 내용은 Debezium 문서의 스냅샷을 참고하세요.
콘솔
Google Cloud 콘솔에서 클러스터 연결 페이지로 이동합니다.
커넥터를 만들려는 Connect 클러스터를 클릭합니다.
커넥터 만들기를 클릭합니다.
커넥터 이름에 문자열을 입력합니다.
커넥터 이름을 지정하는 방법에 대한 가이드라인은 Managed Service for Apache Kafka 리소스 이름 지정 가이드라인을 참고하세요.
커넥터 플러그인에서 PostgreSQL용 Cloud SQL 소스를 선택합니다.
인스턴스 목록에서 Cloud SQL 인스턴스를 선택합니다.
데이터베이스 목록에서 Cloud SQL 데이터베이스를 선택합니다.
주제 접두사 필드에 Kafka 주제 이름에 사용할 접두사를 입력합니다. PostgreSQL용 Cloud SQL 소스 커넥터마다 고유한 접두사를 선택합니다.
선택사항: 테이블 이름 필드에 변경 데이터를 읽어올 테이블 목록을 쉼표로 구분하여 입력합니다(
"schema_name"."table_name"형식). 이 필드를 비워 두면 커넥터가 데이터베이스의 모든 비시스템 테이블에서 변경 데이터를 읽습니다.선택사항: 구성 상자에서 구성 속성을 추가하거나 기본 속성을 수정합니다. 자세한 내용은 커넥터 구성을 참고하세요.
다음 속성의 기본값을 재정의해야 할 수 있습니다.
선택사항: 작업 재시작 정책을 선택합니다. 자세한 내용은 작업 다시 시작 정책을 참고하세요.
만들기를 클릭합니다.
gcloud
-
Google Cloud 콘솔에서 Cloud Shell을 활성화합니다.
Google Cloud 콘솔 하단에 Cloud Shell 세션이 시작되고 명령줄 프롬프트가 표시됩니다. Cloud Shell은 Google Cloud CLI가 사전 설치된 셸 환경으로, 현재 프로젝트의 값이 이미 설정되어 있습니다. 세션이 초기화되는 데 몇 초 정도 걸릴 수 있습니다.
gcloud managed-kafka connectors create명령어를 실행합니다.gcloud managed-kafka connectors create CONNECTOR_ID \ --location=LOCATION \ --connect-cluster=CONNECT_CLUSTER_ID \ --config-file=CONFIG_FILE다음을 바꿉니다.
CONNECTOR_ID: 커넥터의 ID 또는 이름입니다. 커넥터 이름을 지정하는 방법에 대한 가이드라인은 Managed Service for Apache Kafka 리소스 이름 지정 가이드라인을 참고하세요. 커넥터의 이름은 변경할 수 없습니다.LOCATION: 커넥터를 만드는 위치입니다. 이 위치는 Connect 클러스터를 만든 위치와 동일해야 합니다.CONNECT_CLUSTER_ID: 커넥터가 생성된 Connect 클러스터의 ID입니다.CONFIG_FILE: 커넥터의 YAML 구성 파일 경로입니다.
다음은 PostgreSQL용 Cloud SQL 소스 커넥터의 구성 파일 예시입니다.
connector.class: io.debezium.connector.postgresql.PostgresConnector database.dbname: DATABASE_NAME driver.cloudSqlInstance: INSTANCE_ID driver.enableIamAuth: "true" driver.ipTypes: IP_TYPES driver.sslmode: disable key.converter: org.apache.kafka.connect.json.JsonConverter key.converter.schemas.enable: "false" plugin.name: pgoutput slot.name: SLOT_NAME table.include.list: TABLE_LIST topic.prefix: TOPIC_PREFIX value.converter: org.apache.kafka.connect.json.JsonConverter value.converter.schemas.enable: "true"다음을 바꿉니다.
INSTANCE_ID: 데이터베이스가 포함된 Cloud SQL 인스턴스의 ID입니다. 형식은 다음과 같습니다.PROJECT_ID:REGION:INSTANCE_NAME
DATABASE_NAME: 읽어올 Cloud SQL 데이터베이스의 이름입니다.IP_TYPES: 쉼표로 구분된 IP 주소 유형 목록SLOT_NAME: 생성할 복제 슬롯의 이름입니다.TABLE_LIST: 변경 데이터를 읽어올 테이블의 쉼표로 구분된 목록입니다("schema_name"."table_name"형식).TOPIC_PREFIX: Kafka 주제 이름에 사용할 접두사입니다.
커넥터 구성
이 섹션에서는 커넥터에서 설정할 수 있는 몇 가지 구성 속성을 설명합니다. 전체 목록은 Debezium 문서의 PostgreSQL용 Debezium 커넥터를 참고하세요.
IP 주소 유형
driver.ipTypes 속성은 커넥터가 데이터베이스에 연결하는 데 사용하는 IP 주소의 유형을 지정합니다.
PRIVATE: 비공개 IPPSC: Private Service ConnectPUBLIC: 공개 IP
driver.ipTypes 속성에는 선호하는 순서대로 IP 유형의 쉼표로 구분된 목록이 포함됩니다(예: driver.ipTypes=PRIVATE,PUBLIC).
자세한 내용은 네트워킹 구성을 참고하세요.
게시 이름
기본적으로 커넥터는 dbz_publication이라는 게시물에서 스트리밍하려고 시도합니다.
다른 게시를 지정하려면 구성에 publication.name=PUBLICATION_NAME을 추가합니다. 여기서 PUBLICATION_NAME은 게시 이름입니다. 예: publication.name=my_publication
복제 슬롯
PostgreSQL은 복제 슬롯을 사용하여 데이터베이스 테이블 변경사항을 스트리밍합니다. 기본적으로 커넥터는 debezium이라는 복제 슬롯을 만듭니다. 다른 슬롯 이름을 사용하려면 slot.name 속성을 설정하세요.
동일한 데이터베이스에 대해 커넥터 인스턴스를 두 개 만드는 경우 각 커넥터에 고유한 슬롯 이름을 지정해야 합니다.
기본적으로 커넥터는 데이터 손실을 방지하기 위해
slot.drop.on.stop 속성을 false로 설정합니다. 커넥터를 영구적으로 삭제할 때는 커넥터가 사용 중인 복제 슬롯을 수동으로 삭제해야 합니다. 복제 슬롯 이름은 slot.name 속성을 사용하여 다르게 구성하지 않는 한 기본적으로 debezium입니다.
소스 PostgreSQL 데이터베이스 서버에서 WAL 디스크 사용량을 모니터링하도록 알림을 설정하고 사용하지 않는 복제 슬롯을 삭제하는 것이 좋습니다.
표 필터
기본적으로 커넥터는 데이터베이스의 모든 비시스템 테이블에서 변경 데이터를 캡처합니다. 캡처되는 테이블을 필터링하려면 다음 설정 중 하나 이상을 지정하세요.
schema.include.list. 포함할 스키마 목록입니다.schema.exclude.list. 제외할 스키마 목록입니다.schema.include.list와 함께 사용할 수 없습니다.table.include.list. 포함할 테이블 목록입니다.table.exclude.list. 제외할 테이블 목록입니다.table.include.list와 함께 사용할 수 없습니다.
주제 이름
기본적으로 커넥터는 topic_prefix.schema.table_name이라는 이름 지정 규칙을 사용하여 Kafka 주제를 만듭니다. 여기서 topic.prefix은 topic.prefix 구성의 값입니다.
자세한 내용은 Debezium 문서의 주제 이름을 참고하세요.