Дамп таблицы из Kafka в MariaDB с KSQL

Я выполняю агрегирование с KSQL, и мне нужно сохранить выходную таблицу в MariaDB. Я уже настроил MariaDB и JdbcSinkConnector. К сожалению, раковина мне просто не подойдет.

Это структура таблицы в KSQL, которую я хотел бы сбросить в MariaDB:

Field     | Type                           
--------------------------------------------
 a        | VARCHAR(STRING)  (primary key) 
 b        | VARCHAR(STRING)  (primary key) 
 c        | VARCHAR(STRING)  (primary key) 
 d        | INTEGER                        
--------------------------------------------

Я группирую по столбцам a, b, c и выполняю некоторую агрегацию, которая является столбцом d.

Это разъем:

create sink connector test with (
'tasks.max' = 1,
'key.converter.schema.registry.url' = 'http://schema-registry:8081', 
'value.converter.schema.registry.url' = 'http://schema-registry:8081',
'connector.class'          = 'io.confluent.connect.jdbc.JdbcSinkConnector',
'key.converter'='org.apache.kafka.connect.storage.StringConverter',
'value.converter'='org.apache.kafka.connect.storage.StringConverter',
'key.converter.schemas.enable' = 'false',
'value.converter.schemas.enable' = 'true',
'config.action.reload' = 'restart',
'errors.log.enable' = 'true',
'errors.log.include.messages' = 'true',
'print.key' = 'true',
'errors.tolerance' = 'all',
'topics'                   = 'my-topic',
'connection.url'           = 'jdbc:mysql://mariadb-docker-container:3306/ksql?autoReconnect=true&useSSL=false',
'connection.user'          = 'root', 
'connection.password'      = 'strongest-password-you-have-ever-seen',
'pk.fields' = 'a, b, c',
'pk.mode' = 'record_value',
'delete.enabled' = 'false');

Запуск этого соединителя дает мне следующую ошибку:

kafka-connect               | [2021-03-08 12:13:51,339] ERROR [TEST|task-0] WorkerSinkTask{id=TEST-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:187)
kafka-connect               | org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
kafka-connect               |   at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:591)
kafka-connect               |   at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:326)
kafka-connect               |   at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:229)
kafka-connect               |   at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:201)
kafka-connect               |   at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:185)
kafka-connect               |   at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:235)
kafka-connect               |   at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
kafka-connect               |   at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
kafka-connect               |   at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
kafka-connect               |   at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
kafka-connect               |   at java.base/java.lang.Thread.run(Thread.java:834)
kafka-connect               | Caused by: org.apache.kafka.connect.errors.ConnectException: Sink connector 'TEST' is configured with 'delete.enabled=false' and 'pk.mode=record_value' and therefore requires records with a non-null Struct value and non-null Struct schema, but found record at (topic='my-topic',partition=1,offset=30,timestamp=1615205630513) with a String value and string value schema.
kafka-connect               |   at io.confluent.connect.jdbc.sink.RecordValidator.lambda$requiresValue$2(RecordValidator.java:86)
kafka-connect               |   at io.confluent.connect.jdbc.sink.BufferedRecords.add(BufferedRecords.java:82)
kafka-connect               |   at io.confluent.connect.jdbc.sink.JdbcDbWriter.write(JdbcDbWriter.java:73)
kafka-connect               |   at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:75)
kafka-connect               |   at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:563)
kafka-connect               |   ... 10 more
kafka-connect               | [2021-03-08 12:13:51,339] ERROR [TEST|task-0] WorkerSinkTask{id=TEST-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:188)
kafka-connect               | [2021-03-08 12:13:51,339] INFO [TEST|task-0] Stopping task (io.confluent.connect.jdbc.sink.JdbcSinkTask:119)
kafka-connect               | [2021-03-08 12:13:51,339] INFO [TEST|task-0] Closing connection #1 to MySql (io.confluent.connect.jdbc.util.CachedConnectionProvider:108)
kafka-connect               | [2021-03-08 12:13:51,340] INFO [TEST|task-0] [Consumer clientId=connector-consumer-TEST-0, groupId=connect-TEST] Revoke previously assigned partitions my-topic-1, my-topic-0 (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:307)
kafka-connect               | [2021-03-08 12:13:51,340] INFO [TEST|task-0] [Consumer clientId=connector-consumer-TEST-0, groupId=connect-TEST] Member connector-consumer-TEST-0 sending LeaveGroup request to coordinator broker:29092 (id: 2147483646 rack: null) due to the consumer is being closed (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:1010)

Я также пробовал использовать AvroConverter os JsonConverter, но это не помогло, они просто выдавали другие ошибки (ошибки синтаксического анализа), которые не очень помогли. Я знаю, что это как-то связано со структурой данных, согласно некоторым исследованиям. Но как мне предоставить структуру или преобразовать таблицу в Struct, если таблица является результатом группировки столбцов и агрегирования?

Любые идеи? Я также подумываю просто не использовать коннектор, а написать небольшую программу, которая извлекает таблицы / темы и записывает их в MariaDB, на случай, если коннектор не будет работать.


apache-kafka apache-kafka-connect ksqldb
person Richard Steinbrecht    schedule 08.03.2021    source источник
comment
Какую версию ksqlDB вы используете?   -  person Robin Moffatt    schedule 08.03.2021
comment
@RobinMoffatt 0.15   -  person Richard Steinbrecht    schedule 08.03.2021

Ответы (1)


arrow_upward
1
arrow_downward

Если вы используете JDBC Sink, вам необходимо использовать формат сериализации для ваших данных, который включает схему, например используя Avro, Protobuf или схему JSON.

В ksqlDB вы можете указать, что при создании вашего объекта:

CREATE TABLE MY_TABLE WITH (FORMAT='AVRO') AS 
  SELECT A,B,C,COUNT(*) AS D 
    FROM STREAM_FOO 
GROUP BY A,B,C;

Обратите внимание, что поддержка ключей Avro была добавлена ​​в ksqlDB 0.15.

Теперь, когда ваши данные находятся в Avro, вы можете создать свой соединитель приемника, используя соответствующие конвертеры.

create sink connector test with (
  'connector.class'                     = 'io.confluent.connect.jdbc.JdbcSinkConnector',
  'tasks.max'                           = 1,
  'key.converter.schema.registry.url'   = 'http://schema-registry:8081',
  'value.converter.schema.registry.url' = 'http://schema-registry:8081',
  'key.converter'                       = 'io.confluent.connect.avro.AvroConverter',
  'value.converter'                     = 'io.confluent.connect.avro.AvroConverter',
  'key.converter.schemas.enable'        = 'false',
  'value.converter.schemas.enable'      = 'true',
  'config.action.reload'                = 'restart',
  'errors.log.enable'                   = 'true',
  'errors.log.include.messages'         = 'true',
  'print.key'                           = 'true',
  'errors.tolerance'                    = 'all',
  'topics'                              = 'my-topic',
  'connection.url'                      = 'jdbc:mysql://mariadb-docker-container:3306/ksql?autoReconnect=true&useSSL=false',
  'connection.user'                     = 'root',
  'connection.password'                 = 'strongest-password-you-have-ever-seen',
  'pk.fields'                           = 'a, b, c',
  'pk.mode'                             = 'record_key',
  'delete.enabled'                      = 'false');

Две проблемы, с которыми вы столкнулись:

  1. Использование StringConverter означает отсутствие схемы и, следовательно, отчета в сообщении об ошибке String value and string value schema..
  2. Ключ (GROUP BY) таблицы записывается в ключ сообщения Kafka, поэтому именно отсюда (record_key) вы должны получить pk.fields.

Ссылка:

person Robin Moffatt    schedule 08.03.2021
comment
Вы сделали мой день. - person Richard Steinbrecht; 09.03.2021
comment
Не за что :-) - person Robin Moffatt; 09.03.2021