我正在使用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
所以基本上查询与我在连接器配置加WHERE s.id > -1 ORDER BY s.id ASC中query提供的查询相同，因为此联接中的查询生成的结果集为 21 百万行 MySQL 长时间发送数据。当我再次检查show full processlist;时，我现在看到 4 个这样的查询，然后是 8 个，然后是 16 个，依此类推。
问题是：
- 为什么 Kafka 连接在添加时试图一次获取所有行：
s.id > -1 ORDER BY s.id ASC. - 是否可以将连接器配置为不执行此操作，而是获取较小的数量？
"batch.max.rows": "100"是否仅在初始轮询后控制批量大小？
更新：
这个问题有一个开放的话题。我认为这个问题可以结束。
query.suffix 是在 5.5 中添加的。 我用它来添加一个限制语句，它工作得很好，它只是将限制附加到查询的末尾。
查看问题
JDBC 源连接器，带有incrementingmode并通过query， 使用以下 where 子句执行该查询：WHERE incrementingColumnName > lastIncrementedValue ORDER BY incrementingColumnName ASC。 (如果使用增量模式和查询，则无法在那里传递where子句)。
第一次轮询lastIncrementedValue为 -1，因此它会尝试查询所有记录。提取每条记录后，lastIncrementedValue 会增加，因此下一次查询将仅轮询新数据。batch.max.rows是指SourceTask::poll(...)将返回到 Kafka Connect 框架的记录数量。 它是将一次发送到 Kafka 的批处理的最大大小。
我认为，当您从单个表中获取数据时，它的工作速度更快，因为查询执行速度更快(不太复杂)。 如果使用其他 SQL 工具执行这些查询，它将执行类似的操作。