Debezium 소개 - Apache Kafka용 CDC

저는 업무를 수행하면서 새로운 기술 솔루션/소프트웨어 제품을 자주 접하는데, 이에 대한 정보는 러시아어 인터넷에서는 다소 부족합니다. 이 기사에서는 Debezium을 사용하여 널리 사용되는 두 DBMS(PostgreSQL 및 MongoDB)에서 Kafka 클러스터로 CDC 이벤트를 전송하도록 구성해야 했던 최근 실습의 예를 사용하여 그러한 공백 중 하나를 메우려고 합니다. 수행된 작업의 결과로 나타나는 이 리뷰 기사가 다른 사람들에게 도움이 되기를 바랍니다.

일반적으로 Debezium과 CDC는 무엇입니까?

데베 지움 — CDC 소프트웨어 카테고리 대표(데이터 변경 캡처), 더 정확하게는 Apache Kafka Connect 프레임워크와 호환되는 다양한 DBMS를 위한 커넥터 세트입니다.

그것 오픈 소스 프로젝트, Apache License v2.0에 따라 라이센스가 부여되고 Red Hat의 후원을 받습니다. 2016년부터 개발이 진행 중이며 현재 MySQL, PostgreSQL, MongoDB, SQL Server와 같은 DBMS에 대한 공식 지원을 제공합니다. Cassandra 및 Oracle용 커넥터도 있지만 현재는 "초기 액세스" 상태에 있으며 새 릴리스는 이전 버전과의 호환성을 보장하지 않습니다.

CDC를 기존 접근 방식(애플리케이션이 DBMS에서 직접 데이터를 읽는 경우)과 비교하면 주요 이점은 낮은 대기 시간, 높은 안정성 및 가용성으로 행 수준에서 데이터 변경 스트리밍을 구현한다는 것입니다. 마지막 두 가지 사항은 Kafka 클러스터를 CDC 이벤트용 저장소로 사용하여 달성됩니다.

또 다른 장점은 단일 모델을 사용하여 이벤트를 저장하므로 최종 애플리케이션이 서로 다른 DBMS를 운영하는 데 따른 미묘한 차이에 대해 걱정할 필요가 없다는 점입니다.

마지막으로 메시지 브로커를 사용하면 데이터 변경 사항을 모니터링하는 애플리케이션이 수평으로 확장될 수 있습니다. 동시에 데이터를 DBMS에서 직접 가져오지 않고 Kafka 클러스터에서 가져오므로 데이터 소스에 미치는 영향이 최소화됩니다.

Debezium 아키텍처 정보

Debezium을 사용하는 것은 다음과 같은 간단한 계획으로 귀결됩니다.

DBMS(데이터 소스) → Kafka Connect의 커넥터 → Apache Kafka → 소비자

예를 들어, 다음은 프로젝트 웹사이트의 다이어그램입니다.

하지만 저는 이 방식이 별로 마음에 들지 않습니다. 왜냐하면 싱크 커넥터만을 사용하는 것이 가능한 것 같기 때문입니다.

실제로는 상황이 다릅니다. 데이터 레이크를 채우는 것입니다. (위 다이어그램의 마지막 링크) 이것이 Debezium을 사용하는 유일한 방법은 아닙니다. Apache Kafka로 전송된 이벤트는 애플리케이션에서 다양한 상황을 처리하는 데 사용될 수 있습니다. 예를 들어:

• 캐시에서 관련 없는 데이터를 제거합니다.
• 알림 보내기;
• 검색 색인 업데이트;
• 일종의 감사 로그;
• ...

Java 애플리케이션이 있고 Kafka 클러스터를 사용할 필요/가능성이 없는 경우 다음을 통해 작업할 수도 있습니다. 내장 커넥터. 분명한 이점은 추가 인프라(커넥터 및 Kafka 형태)가 필요하지 않다는 것입니다. 그러나 이 솔루션은 버전 1.1부터 더 이상 사용되지 않으며 더 이상 사용이 권장되지 않습니다(이에 대한 지원은 향후 릴리스에서 제거될 수 있음).

이 문서에서는 내결함성과 확장성을 제공하는 개발자가 권장하는 아키텍처에 대해 설명합니다.

커넥터 구성

가장 중요한 가치인 데이터의 변경 사항을 추적하려면 다음이 필요합니다.

1. 버전 5.7부터 MySQL, PostgreSQL 9.6+, MongoDB 3.2+(전체 목록);
2. 아파치 카프카 클러스터;
3. Kafka Connect 인스턴스(버전 1.x, 2.x)
4. Debezium 커넥터를 구성했습니다.

처음 두 점에 대해 작업하십시오. DBMS 및 Apache Kafka의 설치 프로세스는 이 기사의 범위를 벗어납니다. 그러나 샌드박스에 모든 것을 배포하려는 사람들을 위해 예제가 포함된 공식 저장소에 미리 만들어진 도커 작성.yaml.

마지막 두 가지 사항에 대해 더 자세히 설명하겠습니다.

0. 카프카 커넥트

이 문서와 추가 문서에서는 모든 구성 예제가 Debezium 개발자가 배포한 Docker 이미지의 맥락에서 논의됩니다. 여기에는 필요한 모든 플러그인 파일(커넥터)이 포함되어 있으며 환경 변수를 사용하여 Kafka Connect 구성을 제공합니다.

Confluent에서 Kafka Connect를 사용하려는 경우 필요한 커넥터의 플러그인을 다음에 지정된 디렉터리에 독립적으로 추가해야 합니다. plugin.path 또는 환경 변수를 통해 설정 CLASSPATH. Kafka Connect 작업자 및 커넥터에 대한 설정은 작업자 시작 명령에 인수로 전달되는 구성 파일을 통해 결정됩니다. 자세한 내용은 다음을 참조하세요. 선적 서류 비치.

커넥터 버전에서 Debeizum을 설정하는 전체 프로세스는 두 단계로 수행됩니다. 각각을 살펴보겠습니다.

1. Kafka Connect 프레임워크 설정

데이터를 Apache Kafka 클러스터로 스트리밍하기 위해 다음과 같은 특정 매개변수가 Kafka Connect 프레임워크에 설정됩니다.

• 클러스터 연결을 위한 매개변수
• 커넥터 자체의 구성이 직접 저장될 주제의 이름,
• 커넥터가 실행 중인 그룹의 이름(분산 모드가 사용되는 경우)

프로젝트의 공식 Docker 이미지는 환경 변수를 사용하여 구성을 지원합니다. 이것이 우리가 사용할 것입니다. 따라서 이미지를 다운로드하십시오.

docker pull debezium/connect

커넥터를 실행하는 데 필요한 최소 환경 변수 세트는 다음과 같습니다.

• BOOTSTRAP_SERVERS=kafka-1:9092,kafka-2:9092,kafka-3:9092 — 전체 클러스터 구성원 목록을 얻기 위한 Kafka 클러스터 서버의 초기 목록
• OFFSET_STORAGE_TOPIC=connector-offsets - 커넥터가 현재 위치한 위치를 저장하기 위한 주제;
• CONNECT_STATUS_STORAGE_TOPIC=connector-status — 커넥터 및 해당 작업의 상태를 저장하기 위한 주제입니다.
• CONFIG_STORAGE_TOPIC=connector-config — 커넥터 구성 데이터 및 해당 작업을 저장하는 주제입니다.
• GROUP_ID=1 — 커넥터 작업이 실행될 수 있는 작업자 그룹의 식별자입니다. 분산 사용시 필요한 (분산) 모드.

다음 변수를 사용하여 컨테이너를 시작합니다.

docker run 
  -e BOOTSTRAP_SERVERS='kafka-1:9092,kafka-2:9092,kafka-3:9092' 
  -e GROUP_ID=1 
  -e CONFIG_STORAGE_TOPIC=my_connect_configs 
  -e OFFSET_STORAGE_TOPIC=my_connect_offsets 
  -e STATUS_STORAGE_TOPIC=my_connect_statuses  debezium/connect:1.2

Avro에 대한 참고 사항

기본적으로 Debezium은 JSON 형식으로 데이터를 작성합니다. 이는 샌드박스 및 소량의 데이터에는 허용되지만 로드가 많은 데이터베이스에서는 문제가 될 수 있습니다. JSON 변환기의 대안은 다음을 사용하여 메시지를 직렬화하는 것입니다. 아 브로 Apache Kafka의 I/O 하위 시스템에 대한 로드를 줄이는 바이너리 형식으로 변환됩니다.

Avro를 사용하려면 별도의 배포가 필요합니다. 스키마 레지스트리 (다이어그램 저장용). 변환기의 변수는 다음과 같습니다.

name: CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL
value: http://kafka-registry-01:8081/
name: CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL
value: http://kafka-registry-01:8081/
name: VALUE_CONVERTER   
value: io.confluent.connect.avro.AvroConverter

Avro 사용 및 레지스트리 설정에 대한 자세한 내용은 이 문서의 범위를 벗어납니다. 더 명확하게 설명하기 위해 JSON을 사용하겠습니다.

2. 커넥터 자체 구성

이제 소스에서 데이터를 읽는 커넥터 자체의 구성으로 직접 이동할 수 있습니다.

두 DBMS인 PostgreSQL과 MongoDB에 대한 커넥터의 예를 살펴보겠습니다. 여기에는 제가 경험이 있고 차이점이 있습니다(작지만 어떤 경우에는 중요합니다!).

구성은 JSON 표기법으로 설명되며 POST 요청을 사용하여 Kafka Connect에 업로드됩니다.

2.1. 포스트그레SQL

PostgreSQL용 커넥터 구성 예:

{
  "name": "pg-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "127.0.0.1",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "definitelynotpassword",
    "database.dbname" : "dbname",
    "database.server.name": "pg-dev",
    "table.include.list": "public.(.*)",
    "heartbeat.interval.ms": "5000",
    "slot.name": "dbname_debezium",
    "publication.name": "dbname_publication",
    "transforms": "AddPrefix",
    "transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
    "transforms.AddPrefix.regex": "pg-dev.public.(.*)",
    "transforms.AddPrefix.replacement": "data.cdc.dbname"
  }
}

이 설정 후 커넥터 작동 원리는 매우 간단합니다.

• 처음 실행 시 구성에 지정된 데이터베이스에 연결하여 모드로 시작됩니다. 초기 스냅샷, 조건을 사용하여 얻은 초기 데이터 세트를 Kafka로 보냅니다. SELECT * FROM table_name.
• 초기화가 완료되면 커넥터는 PostgreSQL WAL 파일의 변경 사항을 읽는 모드로 전환됩니다.

사용된 옵션 정보:

• name — 아래 설명된 구성이 사용되는 커넥터의 이름 앞으로 이 이름은 Kafka Connect REST API를 통해 커넥터 작업(즉, 상태 보기/다시 시작/구성 업데이트)에 사용됩니다.
• connector.class — 구성된 커넥터에서 사용할 DBMS 커넥터 클래스입니다.
• plugin.name — WAL 파일의 데이터를 논리적으로 디코딩하기 위한 플러그인의 이름입니다. 선택 가능 wal2json, decoderbuffs и pgoutput. 처음 두 개는 DBMS에 적절한 확장을 설치해야 합니다. pgoutput PostgreSQL 버전 10 이상의 경우 추가 조작이 필요하지 않습니다.
• database.* — 데이터베이스 연결 옵션. 여기서 database.server.name — Kafka 클러스터에서 주제 이름을 구성하는 데 사용되는 PostgreSQL 인스턴스 이름입니다.
• table.include.list — 변경 사항을 추적하려는 테이블 목록 형식으로 지정 schema.table_name; 와 함께 사용할 수 없습니다. table.exclude.list;
• heartbeat.interval.ms — 커넥터가 하트비트 메시지를 특수 주제로 보내는 간격(밀리초)입니다.
• heartbeat.action.query — 각 하트비트 메시지를 보낼 때 실행될 요청(옵션은 버전 1.1에 나타남)
• slot.name — 커넥터에서 사용할 복제 슬롯의 이름입니다.
• publication.name - 이름 출판 커넥터가 사용하는 PostgreSQL에서. 만약 존재하지 않는다면, Debezium은 그것을 생성하려고 시도할 것입니다. 연결된 사용자에게 이 작업에 대한 충분한 권한이 없으면 커넥터가 오류와 함께 종료됩니다.
• transforms 대상 주제의 이름을 변경하는 방법을 정확하게 결정합니다.
  • transforms.AddPrefix.type 정규식을 사용할 것임을 나타냅니다.
  • transforms.AddPrefix.regex — 대상 주제의 이름을 재정의하는 마스크
  • transforms.AddPrefix.replacement - 우리가 재정의하는 것을 직접적으로.

하트비트 및 변환에 대해 자세히 알아보기

기본적으로 커넥터는 커밋된 각 트랜잭션에 대해 Kafka에 데이터를 보내고 해당 LSN(로그 시퀀스 번호)이 서비스 항목에 기록됩니다. offset. 그러나 커넥터가 전체 데이터베이스가 아니라 해당 테이블의 일부(데이터 업데이트가 자주 발생하지 않음)만 읽도록 구성되면 어떻게 될까요?

• 커넥터는 WAL 파일을 읽고 모니터링 중인 테이블에 대한 트랜잭션 커밋을 감지하지 않습니다.
• 따라서 주제나 복제 슬롯에서 현재 위치를 업데이트하지 않습니다.
• 그러면 WAL 파일이 디스크에 보관되어 디스크 공간이 부족해질 수 있습니다.

그리고 이것이 바로 옵션이 구출되는 곳입니다. heartbeat.interval.ms и heartbeat.action.query. 이러한 옵션을 쌍으로 사용하면 하트비트 메시지가 전송될 때마다 별도의 테이블에서 데이터 변경 요청을 수행할 수 있습니다. 따라서 커넥터가 현재 위치한(복제 슬롯에 있는) LSN은 지속적으로 업데이트됩니다. 이를 통해 DBMS는 더 이상 필요하지 않은 WAL 파일을 제거할 수 있습니다. 옵션이 작동하는 방식에 대해 자세히 알아볼 수 있습니다. 선적 서류 비치.

주의 깊게 주목할 만한 또 다른 옵션은 다음과 같습니다. transforms. 편리함과 아름다움이 더 중요하지만...

기본적으로 Debezium은 다음 명명 정책을 사용하여 주제를 생성합니다. serverName.schemaName.tableName. 이는 항상 편리하지 않을 수 있습니다. 옵션 transforms 정규식을 사용하여 특정 이름을 가진 주제로 라우팅해야 하는 이벤트인 테이블 목록을 정의할 수 있습니다.

우리 구성에서는 감사합니다. transforms 다음과 같은 일이 발생합니다. 모니터링되는 데이터베이스의 모든 CDC 이벤트는 다음 이름의 주제로 이동합니다. data.cdc.dbname. 그렇지 않으면 (이 설정 없이) Debezium은 기본적으로 다음과 같은 각 테이블에 대한 주제를 생성합니다. pg-dev.public.<table_name>.

커넥터 제한

PostgreSQL용 커넥터 구성에 대한 설명을 마무리하려면 해당 작업의 다음 기능/제한 사항에 대해 이야기하는 것이 좋습니다.

1. PostgreSQL용 커넥터의 기능은 논리적 디코딩 개념에 의존합니다. 그러므로 그는 데이터베이스 구조 변경 요청을 추적하지 않습니다. (DDL) - 따라서 이 데이터는 주제에 포함되지 않습니다.
2. 복제 슬롯을 사용하므로 커넥터 연결이 가능합니다. 만 주요 DBMS 인스턴스에.
3. 커넥터가 데이터베이스에 연결되는 사용자에게 읽기 전용 권한이 있는 경우 처음 시작하기 전에 복제 슬롯을 수동으로 생성하고 데이터베이스에 게시해야 합니다.

구성 적용

이제 구성을 커넥터에 로드해 보겠습니다.

curl -i -X POST -H "Accept:application/json" 
  -H  "Content-Type:application/json"  http://localhost:8083/connectors/ 
  -d @pg-con.json

다운로드가 성공했고 커넥터가 시작되었는지 확인합니다.

$ curl -i http://localhost:8083/connectors/pg-connector/status 
HTTP/1.1 200 OK
Date: Thu, 17 Sep 2020 20:19:40 GMT
Content-Type: application/json
Content-Length: 175
Server: Jetty(9.4.20.v20190813)

{"name":"pg-connector","connector":{"state":"RUNNING","worker_id":"172.24.0.5:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"172.24.0.5:8083"}],"type":"source"}

좋습니다. 설정이 완료되었으며 사용할 준비가 되었습니다. 이제 소비자인 척하고 Kafka에 연결한 후 테이블에 항목을 추가하고 변경하겠습니다.

$ kafka/bin/kafka-console-consumer.sh 
  --bootstrap-server kafka:9092 
  --from-beginning 
  --property print.key=true 
  --topic data.cdc.dbname

postgres=# insert into customers (id, first_name, last_name, email) values (1005, 'foo', 'bar', '[email protected]');
INSERT 0 1
postgres=# update customers set first_name = 'egg' where id = 1005;
UPDATE 1

우리 주제에서는 다음과 같이 표시됩니다.

변경 사항이 포함된 매우 긴 JSON

{
"schema":{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
}
],
"optional":false,
"name":"data.cdc.dbname.Key"
},
"payload":{
"id":1005
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"before"
},
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"after"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"version"
},
{
"type":"string",
"optional":false,
"field":"connector"
},
{
"type":"string",
"optional":false,
"field":"name"
},
{
"type":"int64",
"optional":false,
"field":"ts_ms"
},
{
"type":"string",
"optional":true,
"name":"io.debezium.data.Enum",
"version":1,
"parameters":{
"allowed":"true,last,false"
},
"default":"false",
"field":"snapshot"
},
{
"type":"string",
"optional":false,
"field":"db"
},
{
"type":"string",
"optional":false,
"field":"schema"
},
{
"type":"string",
"optional":false,
"field":"table"
},
{
"type":"int64",
"optional":true,
"field":"txId"
},
{
"type":"int64",
"optional":true,
"field":"lsn"
},
{
"type":"int64",
"optional":true,
"field":"xmin"
}
],
"optional":false,
"name":"io.debezium.connector.postgresql.Source",
"field":"source"
},
{
"type":"string",
"optional":false,
"field":"op"
},
{
"type":"int64",
"optional":true,
"field":"ts_ms"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"id"
},
{
"type":"int64",
"optional":false,
"field":"total_order"
},
{
"type":"int64",
"optional":false,
"field":"data_collection_order"
}
],
"optional":true,
"field":"transaction"
}
],
"optional":false,
"name":"data.cdc.dbname.Envelope"
},
"payload":{
"before":null,
"after":{
"id":1005,
"first_name":"foo",
"last_name":"bar",
"email":"[email protected]"
},
"source":{
"version":"1.2.3.Final",
"connector":"postgresql",
"name":"dbserver1",
"ts_ms":1600374991648,
"snapshot":"false",
"db":"postgres",
"schema":"public",
"table":"customers",
"txId":602,
"lsn":34088472,
"xmin":null
},
"op":"c",
"ts_ms":1600374991762,
"transaction":null
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
}
],
"optional":false,
"name":"data.cdc.dbname.Key"
},
"payload":{
"id":1005
}
}{
"schema":{
"type":"struct",
"fields":[
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"before"
},
{
"type":"struct",
"fields":[
{
"type":"int32",
"optional":false,
"field":"id"
},
{
"type":"string",
"optional":false,
"field":"first_name"
},
{
"type":"string",
"optional":false,
"field":"last_name"
},
{
"type":"string",
"optional":false,
"field":"email"
}
],
"optional":true,
"name":"data.cdc.dbname.Value",
"field":"after"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"version"
},
{
"type":"string",
"optional":false,
"field":"connector"
},
{
"type":"string",
"optional":false,
"field":"name"
},
{
"type":"int64",
"optional":false,
"field":"ts_ms"
},
{
"type":"string",
"optional":true,
"name":"io.debezium.data.Enum",
"version":1,
"parameters":{
"allowed":"true,last,false"
},
"default":"false",
"field":"snapshot"
},
{
"type":"string",
"optional":false,
"field":"db"
},
{
"type":"string",
"optional":false,
"field":"schema"
},
{
"type":"string",
"optional":false,
"field":"table"
},
{
"type":"int64",
"optional":true,
"field":"txId"
},
{
"type":"int64",
"optional":true,
"field":"lsn"
},
{
"type":"int64",
"optional":true,
"field":"xmin"
}
],
"optional":false,
"name":"io.debezium.connector.postgresql.Source",
"field":"source"
},
{
"type":"string",
"optional":false,
"field":"op"
},
{
"type":"int64",
"optional":true,
"field":"ts_ms"
},
{
"type":"struct",
"fields":[
{
"type":"string",
"optional":false,
"field":"id"
},
{
"type":"int64",
"optional":false,
"field":"total_order"
},
{
"type":"int64",
"optional":false,
"field":"data_collection_order"
}
],
"optional":true,
"field":"transaction"
}
],
"optional":false,
"name":"data.cdc.dbname.Envelope"
},
"payload":{
"before":{
"id":1005,
"first_name":"foo",
"last_name":"bar",
"email":"[email protected]"
},
"after":{
"id":1005,
"first_name":"egg",
"last_name":"bar",
"email":"[email protected]"
},
"source":{
"version":"1.2.3.Final",
"connector":"postgresql",
"name":"dbserver1",
"ts_ms":1600375609365,
"snapshot":"false",
"db":"postgres",
"schema":"public",
"table":"customers",
"txId":603,
"lsn":34089688,
"xmin":null
},
"op":"u",
"ts_ms":1600375609778,
"transaction":null
}
}

두 경우 모두 레코드는 변경된 레코드의 키(PK)와 변경의 본질, 즉 레코드가 이전에 무엇이었는지, 이후에 무엇이 되었는지로 구성됩니다.

• 의 경우 INSERT: 이전 값(before) 같음 null, 그리고 그 이후 - 삽입된 라인.
• 의 경우 UPDATE:에서 payload.before 라인의 이전 상태가 표시되고, payload.after — 변화의 본질을 지닌 새로운 것.

2.2 몽고DB

이 커넥터는 표준 MongoDB 복제 메커니즘을 사용하여 기본 DBMS 노드의 oplog에서 정보를 읽습니다.

이미 설명한 PgSQL용 커넥터와 유사하게 여기에서도 처음 시작할 때 기본 데이터 스냅샷이 생성되고 그 후 커넥터가 oplog 읽기 모드로 전환됩니다.

구성 예 :

{
"name": "mp-k8s-mongo-connector",
"config": {
"connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
"tasks.max": "1",
"mongodb.hosts": "MainRepSet/mongo:27017",
"mongodb.name": "mongo",
"mongodb.user": "debezium",
"mongodb.password": "dbname",
"database.whitelist": "db_1,db_2",
"transforms": "AddPrefix",
"transforms.AddPrefix.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.AddPrefix.regex": "mongo.([a-zA-Z_0-9]*).([a-zA-Z_0-9]*)",
"transforms.AddPrefix.replacement": "data.cdc.mongo_$1"
}
}

보시다시피 이전 예와 비교하여 여기에는 새로운 옵션이 없지만 데이터베이스 연결을 담당하는 옵션 수와 접두사만 줄었습니다.

설정 transforms 이번에는 다음을 수행합니다. 스키마에서 대상 주제의 이름을 변환합니다. <server_name>.<db_name>.<collection_name> в data.cdc.mongo_<db_name>.

결함 허용

우리 시대의 내결함성 및 고가용성 문제는 그 어느 때보다 심각합니다. 특히 데이터와 트랜잭션에 관해 이야기할 때 데이터 변경 사항을 추적하는 것이 이 문제에서 제외되지 않습니다. 원칙적으로 무엇이 잘못될 수 있는지, 각 경우에 Debezium에 어떤 일이 일어날지 살펴보겠습니다.

옵트아웃 옵션에는 세 가지가 있습니다.

1. 카프카 연결 실패. Connect가 분산 모드에서 작동하도록 구성된 경우 동일한 group.id를 설정하려면 여러 작업자가 필요합니다. 그런 다음 그 중 하나가 실패하면 커넥터가 다른 작업자에서 다시 시작되고 Kafka 주제에서 마지막으로 커밋된 위치부터 계속 읽습니다.
2. Kafka 클러스터와의 연결 끊김. 커넥터는 단순히 Kafka로 전송에 실패한 위치에서 읽기를 중지하고 시도가 성공할 때까지 주기적으로 재전송을 시도합니다.
3. 데이터 소스를 사용할 수 없음. 커넥터는 구성된 대로 소스에 다시 연결을 시도합니다. 기본값은 다음을 사용하여 16번 시도하는 것입니다. 지수 백오프. 16번째 시도에 실패하면 해당 작업은 다음과 같이 표시됩니다. 실패한 Kafka Connect REST 인터페이스를 통해 수동으로 다시 시작해야 합니다.
   • 의 경우 PostgreSQL 데이터는 손실되지 않습니다. 왜냐하면 복제 슬롯을 사용하면 커넥터가 읽지 않는 WAL 파일을 삭제하는 것을 방지할 수 있습니다. 이 경우 코인의 단점도 있습니다. 커넥터와 DBMS 사이의 네트워크 연결이 장기간 중단되면 디스크 공간이 부족해질 가능성이 있고 이로 인해 장애가 발생할 수 있습니다. 전체 DBMS.
   • 의 경우 MySQL의 binlog 파일은 연결이 복원되기 전에 DBMS 자체에 의해 회전될 수 있습니다. 이로 인해 커넥터가 실패 상태로 전환되고, 정상 작동을 복원하려면 초기 스냅샷 모드에서 다시 시작하여 binlog에서 계속 읽어야 합니다.
   • 약 MongoDB의. 설명서에 따르면 로그/oplog 파일이 삭제되고 커넥터가 중단된 위치에서 계속 읽을 수 없는 경우 커넥터의 동작은 모든 DBMS에 대해 동일합니다. 이는 커넥터가 상태로 전환됨을 의미합니다. 실패한 모드에서 다시 시작해야 합니다. 초기 스냅샷.

     그러나 예외가 있습니다. 커넥터 연결이 오랫동안 끊어지고(또는 MongoDB 인스턴스에 도달할 수 없음) 이 시간 동안 oplog가 회전한 경우 연결이 복원되면 커넥터는 사용 가능한 첫 번째 위치에서 침착하게 계속 데이터를 읽습니다. 이것이 바로 Kafka의 일부 데이터가 아니 칠 것이다.

결론

Debezium은 CDC 시스템에 대한 나의 첫 경험이며 전반적으로 매우 긍정적입니다. 이 프로젝트는 주요 DBMS 지원, 구성 용이성, 클러스터링 지원 및 활발한 커뮤니티로 성공했습니다. 실무에 관심이 있는 분들은 가이드를 읽어보시길 권합니다. 카프카 커넥트 и 데베 지움.

Kafka Connect용 JDBC 커넥터와 비교할 때 Debezium의 가장 큰 장점은 DBMS 로그에서 변경 사항을 읽어 최소한의 대기 시간으로 데이터를 수신할 수 있다는 것입니다. Kafka Connect의 JDBC 커넥터는 고정된 간격으로 모니터링되는 테이블을 쿼리하고 (같은 이유로) 데이터가 삭제될 때 메시지를 생성하지 않습니다(존재하지 않는 데이터를 어떻게 쿼리할 수 있습니까?).

유사한 문제를 해결하려면 Debezium 외에 다음 솔루션에 주의를 기울일 수 있습니다.

• JDBC 커넥터 Kafka 연결;
• MySQL 전용의 여러 솔루션:
  • MySQL 스트리머;
  • 맥스웰;
  • SpinalTap;
• 오라클 골든게이트, 그러나 이것은 완전히 다른 "체중 범주"입니다.

PS

블로그에서도 읽어보세요.

• «Kubernetes에서 Kafka 클러스터에 적합한 크기 결정";
• «SRE 일상생활의 실용적인 이야기. 2 부";
• «Kubernetes용 PostgreSQL 문에 대한 간략한 개요, 우리의 선택 및 경험".

출처 : habr.com