Блог

Я создаю конвейер данных, используя соединитель источника и приемника Kafka. Соединитель источника использует базу данных SQL и публикует в теме, а соединитель приемника подписывается на тему и помещает в другую базу данных SQL. В таблице 16 ГБ данных. Теперь проблема в том, что данные не передаются из одной базы данных в другую. Однако, если размер таблицы невелик, например 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 для передачи больших объемов данных?

Мне удалось преодолеть эту проблему, ограничив количество записей, возвращаемых в одном запросе к базе данных, например, 5000 за раз.

Решение будет зависеть от базы данных и диалекта SQL. Приведенные ниже примеры будут работать и корректно управлять смещениями для одной таблицы. Увеличивающийся столбец ID и временная метка должны быть установлены в соответствии с инструкциями, указанными здесь: https://docs.confluent.io/kafka-connect-jdbc/current/source-connector/index.html#incremental-query-modes

Пример таблицы myTable содержит следующие столбцы:

• id увеличивается при каждом добавлении новой записи
• lastUpdatedTimestamp — обновляется каждый раз, когда обновляется запись
• некоторые другие атрибуты

id и lastUpdatedTimestamp должен однозначно идентифицировать запись в наборе данных.

Соединитель формирует запрос следующим образом:

config.query Kafka Connect WHERE clause for a selected mode 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

 "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
  

Oracle

 "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
  

Этот подход не будет работать с bulk режимом. Он работает с timestamp incrementing режимом и может работать с timestamp или incrementing режимами в зависимости от характеристик таблицы.

Объединение многих таблиц — идея, которую я не тестировал!

Это усложняется, если запрос выполняет объединения во многих таблицах. Для этого потребуется следующее:

• укажите идентификатор, который однозначно идентифицировал бы строку, например, объединенную tableA.id tableB.id
• укажите временную метку последней обновленной записи таблицы, какой бы ни была последняя
• упорядочивайте записи запроса соответствующим образом

Существует ограничение на то, какой длины может быть идентификатор из-за длинного типа данных Java, используемого Kafka Connect, т.Е. 9,223,372,036,854,775,807