Как сделать начальный снимок загрузки с дебезиум mysql разъемом кафка? - PullRequest
Новая
       79

Как сделать начальный снимок загрузки с дебезиум mysql разъемом кафка?

1 голос
/ 08 апреля 2020

Я использую Kafka Cluster и Debezium MySql, чтобы получать сообщения от обновлений базы данных Elasticsearch. Некоторое время я проводил некоторые тесты и в итоге получил смешанный ES, поэтому я хотел выполнить полную загрузку Elasticsearch fre sh.

Итак, я хотел остановить разъем дебезия, чтобы прекратить подачу ES и удалите все индексы, чтобы при повторном запуске разъема он мог выполнить полную загрузку. Насколько я знаю, коннектор работает только с действиями pause/resume, а те не выполняют первоначальную загрузку.

В этот момент я вручную удалю коннектор и созданные темы и воссоздаю их так, чтобы он Можно загрузить как это было изначально, но есть идеи, как выполнить этот процесс очистки в правильные шаги?

1 Ответ

0 голосов
/ 14 апреля 2020

Я не знаю, есть ли лучший способ, но это решение сработало для меня:

  1. Остановлены службы для демона /etc/kafka/connect-distributed.properties. 
sudo systemctl stop confluent-connect-distributed.service
Удалить разъем Дебезиум 
curl -X DELETE http://localhost:8083/connectors/<connector name>
удалите все темы, связанные с моей работой, в данном случае этот кластер kafka был разработчиком, поэтому я удалил все темы, даже те, которые по умолчанию. 
__consumer_offsets
_confluent-command
_schemas
connect-configs
connect-offsets
connect-status

kafka-topics --bootstrap-server <kafka bootstrap> --delete --topic <topic name>
Создайте заново темы по умолчанию 
kafka-topics --create --bootstrap-server <boostrap kafka> --topic connect-configs --replication-factor 3 --partitions 1 --config cleanup.policy=compact
kafka-topics --create --bootstrap-server <boostrap kafka> --topic connect-offsets --replication-factor 3 --partitions 50 --config cleanup.policy=compact
kafka-topics --create --bootstrap-server <boostrap kafka> --topic connect-status --replication-factor 3 --partitions 10 --config cleanup.policy=compact
Запустить сервис снова 
sudo systemctl start confluent-connect-distributed.service
Снова создайте соединитель 
curl -s -i -X PUT -H  "Content-Type:application/json" \
    http://localhost:8083/connectors/<connector name>/config \
    -d '<json>'

Таким образом, вы получите исходный снимок вasticsearch.

...