Kafka源和接收器连接器不适用于大数据

Question

我正在使用Kafka源和接收器连接器创建一个数据管道。源连接器从SQL数据库消费并发布到主题，而接收器连接器订阅主题并放入其他SQL数据库。表有16 GB的数据。现在的问题是，数据不能从一个数据库传输到另一个数据库。但是，如果表的大小很小，比如1000行，那么数据可以成功传输。

源连接器配置:

"config": {
       "connector.class": 
"io.confluent.connect.jdbc.JdbcSourceConnector",
       "tasks.max": "1",
       "connection.url": "",
       "mode": "incrementing",
       "incrementing.column.name": "ID",
       "topic.prefix": "migration_",
       "name": "jdbc-source",
       "validate.non.null": false,
       "batch.max.rows":5
     }

源连接器日志：

INFO WorkerSourceTask{id=cmc-migration-source-0} flushing 0 outstanding messages for offset commit 
[2019-03-08 16:48:45,402] INFO WorkerSourceTask{id=cmc-migration-source-0} Committing offsets
[2019-03-08 16:48:45,402] INFO WorkerSourceTask{id=cmc-migration-source-0} flushing 0 outstanding messages for offset commit
[2019-03-08 16:48:55,403] INFO WorkerSourceTask{id=cmc-migration-source-0} Committing offsets(org.apache.kafka.connect.runtime.WorkerSourceTask:397)

有人能指导我如何调整Kafka源连接器以传输大数据吗？

Answer 1

我通过限制单个查询中返回到数据库的记录数量（例如一次5000条），成功地解决了这个问题。

解决方案将取决于数据库和SQL方言。下面的例子将为单个表正确地工作和管理偏移量。增量ID列和时间戳必须按照此处指定的说明进行设置:https://docs . confluent . io/Kafka-connect-JDBC/current/source-connector/index . html # incremental-query-modes

示例表< code>myTable包含以下各列:

< li> id在每次添加新记录时递增 < Li > < code > lastUpdatedTimestamp -每次记录更新时更新 < li >其他一些属性

id 和 lastUpdateTimestamp 必须唯一标识数据集中的记录。

连接器构造查询如下：

config.query选定模式的Kafka Connect WHere子句config.query.suffix

PostgreSQL / MySQL

"config": {
    ...
    "poll.interval.ms" : 10000,
    "mode":"timestamp+incrementing",
    "incrementing.column.name": "id",
    "timestamp.column.name": "lastUpdatedTimestamp",
    "table.whitelist": "myTable",
    "query.suffix": "LIMIT 5000"
    ...
    }

将导致:

SELECT *
FROM "myTable"
WHERE "myTable"."lastUpdatedTimestamp" < ?
    AND (
        ("myTable"."lastUpdatedTimestamp" = ? AND "myTable"."id" > ?)
        OR
        "myTable"."lastUpdatedTimestamp" > ?
        )
ORDER BY
    "myTable"."lastUpdatedTimestamp",
    "myTable"."id" ASC
LIMIT 5000

如果要在WHERE子句中添加附加条件，请执行以下操作。

"config": {
    ...
    "poll.interval.ms" : 10000,
    "mode":"timestamp+incrementing",
    "incrementing.column.name": "id",
    "timestamp.column.name": "lastUpdatedTimestamp",
    "query": "SELECT * FROM ( SELECT id, lastUpdatedTimestamp, name, age FROM myTable WHERE Age > 18) myQuery",
    "query.suffix": "LIMIT 5000"
    ...
    }

将导致:

SELECT *
FROM (
    SELECT id, lastUpdatedTimestamp, name, age
    FROM myTable
    WHERE Age > 18
    ) myQuery
WHERE "myTable"."lastUpdatedTimestamp" < ?
    AND (
        ("myTable"."lastUpdatedTimestamp" = ? AND "myTable"."id" > ?)
        OR
        "myTable"."lastUpdatedTimestamp" > ?
        )
ORDER BY
    "myTable"."lastUpdatedTimestamp",
    "myTable"."id" ASC
LIMIT 5000

SQL Server

json prettyprint-override">"config": {
    ...
    "poll.interval.ms" : 10000,
    "mode":"timestamp+incrementing",
    "incrementing.column.name": "id",
    "timestamp.column.name": "lastUpdatedTimestamp",
    "query": "SELECT TOP 5000 * FROM (SELECT id, lastUpdatedTimestamp, name, age FROM myTable) myQuery",
    ...
    }

将导致:

SELECT TOP 5000 *
FROM (
    SELECT id, lastUpdatedTimestamp, name, age
    FROM myTable
    WHERE Age > 18
    ) myQuery
WHERE "myTable"."lastUpdatedTimestamp" < ?
    AND (
        ("myTable"."lastUpdatedTimestamp" = ? AND "myTable"."id" > ?)
        OR
        "myTable"."lastUpdatedTimestamp" > ?
        )
ORDER BY
    "myTable"."lastUpdatedTimestamp",
    "myTable"."id" ASC

神谕

"config": {
    ...
    "poll.interval.ms" : 10000,
    "mode":"timestamp+incrementing",
    "incrementing.column.name": "id",
    "timestamp.column.name": "lastUpdatedTimestamp",
    "query": "SELECT * FROM (SELECT id, lastUpdatedTimestamp, name, age FROM myTable WHERE ROWNUM <= 5000) myQuery",
    ...
    }

将导致:

SELECT *
FROM (
    SELECT id, lastUpdatedTimestamp, name, age
    FROM myTable
    WHERE ROWNUM <= 5000
    ) myQuery
WHERE "myTable"."lastUpdatedTimestamp" < ?
    AND (
        ("myTable"."lastUpdatedTimestamp" = ? AND "myTable"."id" > ?)
        OR
        "myTable"."lastUpdatedTimestamp" > ?
        )
ORDER BY
    "myTable"."lastUpdatedTimestamp",
    "myTable"."id" ASC

此方法不适用于批量模式。它适用于时间戳递增模式，也可能适用于时间戳或递增模式，具体取决于表特征。

加入许多表格-我还没有测试过的想法！

如果查询跨多个表执行连接，情况会变得更加复杂。这将需要以下条件:

• 提供唯一标识行的id，例如连接tableA.idtableB.id
• 提供最近更新的表记录的时间戳，以最新的为准
• 适当订购查询记录

由于 Kafka Connect 使用的 Java long 数据类型，ID 的长度有限制，即 9，223，372，036，854，775，807