Kafka Connect JDBC 커넥터 쿼리 + 초기 폴링에서 대규모 데이터 세트로 증가 모드 초크

JDBC 커넥터를 사용하여 MySQL에서 Kafka로 데이터를 이동하고 있습니다. 내가 관심있는 데이터는 3 개의 테이블을 조인하는 선택에서 가져온 것이므로 mode:incrementing및 query다음을 사용 하여 커넥터를 구성했습니다 .

{
    "name": "stats",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "key.converter": "io.confluent.connect.avro.AvroConverter",
        "key.converter.schema.registry.url": "http://schema-registry.kafka-broker:8081",
        "value.converter": "io.confluent.connect.avro.AvroConverter",
        "value.converter.schema.registry.url": "http://schema-registry.kafka-broker:8081",
        "connection.url": "jdbc:mysql://DB_HOST:3306/SCHEMA?user=USER&password=PASSWORD&zeroDateTimeBehavior=CONVERT_TO_NULL&useSSL=false",
        "mode": "incrementing",
        "validate.non.null": "false",
        "topic.prefix": "t",
        "incrementing.column.name": "s.id",
        "transforms": "createKey,extractString",
        "transforms.createKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
        "transforms.createKey.fields": "uuid",
        "transforms.extractString.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
        "transforms.extractString.field": "uuid",
        "quote.sql.identifiers":"never",
        "query": "select s.id, concat(database(), '_', s.id) as uuid, c.email, s.type as type, s.created_at as log_date, a.type as a_type from stats s join concact c on c.id = s.recipient_id join address a on a.id = s.address_id",
        "errors.tolerance": "all",
        "errors.log.enable": "true",
        "errors.log.include.messages": "true",
        "batch.max.rows": "100",
        "poll.interval.ms": "60000"
    }
}

커넥터 상태를 확인할 때 실행 중입니다.

curl http://conncet:8083/connectors/stats/status

{
    "name": "stats",
    "connector": {
        "state": "RUNNING",
        "worker_id": "connect-3:38083"
    },
    "tasks": [
        {
            "id": 0,
            "state": "RUNNING",
            "worker_id": "connect-1:18083"
        }
    ],
    "type": "source"
}

그러나 한 시간 후에도 여전히 생성 된 주제를 볼 수 없습니다. 어떤 쿼리가 실행 중인지 MySQL에서 확인했으며 show full processlist;다음과 같은 두 가지 쿼리가 표시됩니다.

select s.id, concat(database(), '_', s.id) as uuid, c.email, s.type as type, s.created_at as log_date, a.type as a_type from stats s join concact c on c.id = s.recipient_id join address a on a.id = s.address_id WHERE s.id > -1 ORDER BY s.id ASC

따라서 기본적으로 쿼리는 query커넥터 구성에 제공 한 쿼리와 동일 WHERE s.id > -1 ORDER BY s.id ASC합니다.이 조인의 쿼리는 MySQL이 오랫동안 데이터를 보내는 2 천 1 백만 행의 결과 집합을 생성하기 때문입니다. 다시 확인하면 show full processlist;다음과 같은 4 개의 쿼리가 표시되고 8 개, 16 개 등이 표시됩니다.

질문은 다음과 같습니다.

1. Kafka connect 가 다음을 추가 할 때 한 번에 모든 행 을 가져 오려고하는 이유 : s.id > -1 ORDER BY s.id ASC.
2. 이 작업을 수행하지 않고 대신 더 적은 양을 가져 오도록 커넥터를 구성 할 수 있습니까?
3. "batch.max.rows": "100"초기 폴링 후에 배치 크기 만 제어하고 있습니까 ??

최신 정보:

이 문제에 대한 공개 주제가 있습니다. 이 질문은 끝날 수 있다고 생각합니다.