여러 주제 및 테이블 대상으로 kafka 싱크 연결을 배포하는 방법

내에서 이전 질문 , 내가 분산 카프카와 데이터베이스의 실시간 동기화에 대한 소비자 배포에 대한 자세한 동의로 결정했다. 같은 경우 PostgreSQL에서 SQL Server로 가져 오려는 테이블이 수백 개가 넘습니다. PostgreSQL에서 Kafka까지 저는 wal2json 플러그인과 함께 Debezium 커넥터를 사용했습니다. 그리고 Kafka에서 SQL Server까지 JDBC 커넥터를 사용합니다. 3 개의 동일한 설정 브로커 (다른 주소)가 있습니다.

broker.id=0
broker.rack=1
port=9093
listeners=PLAINTEXT://0.0.0.0:9093
advertised.listeners=PLAINTEXT://localhost:9093
log.dir=/home/admin/kafka/tmp/kafka_log1
offsets.topic.num.partition=1
offsets.topic.replication.factor=3
min.isnyc.replicas=2
default.replication.factor=3
zookeeper.connect=localhost:2181,localhost:2182,localhost:2183
zookeeper.connection.timeout.ms=7200000
delete.topic.enable=true
message.max.bytes=50497182 
replica.fetch.max.bytes=50497182
group.max.session.timeout.ms=7200000

다음과 같은 몇 가지 가능한 해결책을 시도했습니다.

1. 1 개의 파티션과 3 개의 복제본을 사용하도록 주제를 설정합니다. 내 테이블 _에 이름 이 있기 때문에 경고를받습니다.

kafka-topics.sh -create --bootstrap-server localhost:9093,localhost:9094,localhost:9095  --replication-factor 3 --partitions 1 --topic $topic_name --config retention.ms=5400000

2. 나는 서로 다른 작업자로 debezium과 jdbc 커넥터를 분리합니다. 다음과 같이 동일한 구성 (호스트 포트 제외, debezium의 경우 8085, 싱크의 경우 8084)을 가진 두 명의 작업자가 있습니다.

bootstrap.servers=localhost:9093,localhost:9094,localhost:9095
group.id=debezium-cluster
key.converter.schemas.enable=true
value.converter.schemas.enable=true
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
offset.storage.topic=connect-offsets-debezium
offset.storage.replication.factor=3
config.storage.topic=connect-configs-debezium
status.storage.topic=connect-status-debezium
producer.buffer.memory=29999999
producer.max.buffered.records=19999999
producer.max.request.size=51497182 
producer.retries=100
producer.max.in.flight.requests.per.connection=1
producer.request.timeout.ms=20000
producer.enable.idempotence=true
producer.retry.backoff.ms=500
producer.send.buffer.bytes=50497182
producer.receive.buffer.bytes=50497182
producer.ack=1
offset.flush.timeout.ms=300000
producer.buffer.memory=51497182
consumer.enable.auto.commit=true
consumer.retries=100
consumer.auto.commit.interval.ms=100000
consumer.max.partition.fetch.bytes=50497182
consumer.max.poll.records=10000
consumer.request.timeout.ms=20000
consumer.retry.backoff.ms=50000
consumer.session.timeout.ms=50000
consumer.auto.offset.reset=latest
consumer.isolation.level=read_committed
consumer.max.poll.interval.ms=5400000
fetch_max_bytes=50497182
rest.port=8085
plugin.path=/home/admin/kafka/connectors

3. 다음없이 하나씩 루프 싱크 커넥터 :

#!/bin/bash
CSV_LIST="/home/admin/kafka/main/config/tables/table_lists.csv"
DATA=${CSV_LIST}

while IFS=',' read table pk mode; do
topic_name=${table} 
curl -X POST http://localhost:8084/connectors -H 'Content-Type:application/json' -d '{"name" :"sqlservercon_'$topic_name'",
    "config":{"connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector",
            "topics":"'$table'",
            "connection.url":"jdbc:sqlserver://-:1433",
            "connection.user":"-",
            "connection.password":"-",
            "transforms":"unwrap",
            "transforms.unwrap.type":"io.debezium.transforms.ExtractNewRecordState",
            "transforms.unwrap.drop.tombstones":"false",
            "auto.create":"true",
            "insert.mode":"'$mode'",
            "pk.fields":" '$pk'",
            "pk.mode":"record_value",
            "destination.table.format":"db.dbo.'$table'"
}}' | jq
done < ${DATA}

배포 방법은 다음과 같습니다.

1. zookeeper 및 kafka 서버 시작
2. 주제 만들기
3. Debezium 소스에 대한 kafka 작업자 시작
4. debezium 커넥터 추가 (1db에는 하나의 커넥터 만 필요하므로)
5. 싱크대를 위해 kafka 작업자 시작
6. 루프로 JDBC 커넥터 추가

안타깝게도 몇 가지 교착 상태와 소비자의 인식이 없기 때문에 모든 데이터를 새 SQL Server 데이터베이스로 이동하는 것이 여전히 만족스럽지 않습니다. 최적의 소비자 배포를위한 좋은 제안이 있는지 알고 싶습니다. 각 커넥터에 대해 한 명의 작업자를 추가해야합니까? 아니면 각 주제간에 전환하는 것과 같은 작업을 수행해야합니까?