Обзор

В этой статье подробно описана семантика «один и только один раз» проекта clickhouse-kafka-connect. реализация и практический опыт итог. Проект основан на Кафке. Платформа подключения и новая функция ClickHouse KeeperMap (хранилище состояний) реализуют функцию синхронизации данных Kafka с ClickHouse в реальном времени на основе семантики «точно один раз». Этот проект основан на Java API официального сайта ClickHouse для поддержки всех типов данных ( включая сложные типы данных: Map/Tuple/Json и т. д.); этот проект соответствует Apache2.0. License。

План реализации

Так называемая семантика «точно один раз»: то есть все данные Kafka синхронизируются с ClickHouse без дублирования и потери. Звучит просто, но реализация этой семантики действительно является большой проблемой. Обычный процесс обработки: прочитать Kafka, написать ClickHouse и записать Offset. Среди них решающее значение имеет порядок записи ClickHouse и записи Offset.

1. Семантика «хотя бы один раз»: сначала напишите ClickHouse、Record Offset снова успешен;、Последний терпит неудачу、Перезапуск.
2. Семантика «самый один раз»: сначала запишите смещение.、Напишите ClickHouse еще раз; первый вариант успешен;、Последний терпит неудачу、Перезапуск.

Итак, как это решение обеспечивает семантику «точно один раз»?

Приведенный выше пример соответствует одному работнику и одному потоку, а один поток соответствует одной теме/разделу. Для каждого пакета в каждой теме/разделе информация о записи хранилища состояния следующая:

1. MinOffset и maxOffset текущего пакета.
2. Перед записью текущего пакета в ClickHouse,Установите флаг BEFORE после записи в ClickHouse;,Установите флаг ПОСЛЕ.

При каждой обработке пакетных данных сначала запрашивается вышеуказанная информация, записанная в текущей теме/разделе. Если информация о записи имеет флаг AFTER, это означает, что последняя запись в ClickHouse была успешной; если информация о записи имеет флаг BEFORE, это означает, что последняя запись в ClickHouse не удалась; Если записанной информации нет, это означает, что тема/раздел обрабатывается впервые. Это самый простой сценарий:

Как показано на рисунке выше, флаг BEFORE устанавливается перед пакетом Insert, а флаг AFTER устанавливается после. Для последующих вставок состояние сохраняет информацию о записи каждой темы/раздела. При сравнении предыдущего состояния с новым пакетом возможны 4 результата и 2 действия в зависимости от того, установлен ли флаг ДО или ПОСЛЕ:

1. То же – предыдущее минимальное и максимальное смещение такое же, как и в текущей партии. Если флаг ДО, неактивная партия, установите ПОСЛЕ, если флаг ПОСЛЕ, отбросьте текущую партию;

2. Перекрытие – предыдущее минимальное смещение такое же, как и у текущей партии, но предыдущее максимальное смещение меньше, чем у текущей партии. В этом случае текущий пакет будет разделен на два фрагмента. Если флаг ДО, то вставить первый фрагмент, установить ПОСЛЕ, установить ДО, вставить второй фрагмент, установить ПОСЛЕ, если флаг ПОСЛЕ, установить ДО, вставить второй фрагмент, установить ПОСЛЕ;

3.Содержит — предыдущие диапазоны минимального и максимального смещения включают текущую партию. Если установлен флаг BEFORE, добавьте его в очередь недоставленных сообщений DLQ, если установлен флаг AFTER, отбросьте его.

4.Новый – самая распространенная ситуация. Предыдущее максимальное смещение + 1 было равно текущему минимальному смещению пакета. Если флаг установлен ДО, вставьте пакет, затем установите ПОСЛЕ, если флаг установлен ПОСЛЕ, установите ДО, вставьте пакет, а затем установите ПОСЛЕ;

Практические случаи

1. Установите ClickHouse RPM.

rpm -ivh clickhouse-common-static-23.7.5.30.x86_64.rpm
rpm -ivh clickhouse-server-23.7.5.30.x86_64.rpm
rpm -ivh clickhouse-client-23.7.5.30.x86_64.rpm

Примечание. Пароль по умолчанию, установленный при установке сервера, требуется, когда клиент ClickHouse подключается к серверу (здесь я установил его на felixzh).

2. Измените сервисный порт.

Порт по умолчанию: 9000, при необходимости его можно изменить.

3. Измените файл конфигурации config.xml.

chmod 744 /etc/clickhouse-server/config.xml
vim /etc/clickhouse-server/config.xml

4. Запустите сервис ClickHouse.

systemctl start clickhouse-server
systemctl status clickhouse-server

Если запуск не удался, вы можете проверить журнал для устранения неполадок:

/var/log/clickhouse-server/ clickhouse-server.err.log

5. Вход клиента

clickhouse-client -u default --password felixzh --port 9000  

На этом этапе развертывание, настройка и запуск ClickHouse завершены. Создайте последующую тестовую таблицу kafka2ck:

create table kafka2ck(`id` UInt8, `name` String) ENGINE=MergeTree() ORDER BY id;

6.преобразование clickhouse-kafka-connect

https://github.com/ClickHouse/clickhouse-kafka-connect

Проект разработан на основе Java11. Если вам нужно использовать Java8, просто измените его следующим образом:

Компиляция исходного кода: чистая сборка Gradle

Скопируйте clickhouse-kafka-connect-v0.0.18-confluent.jar в каталог установки kafka libs/.

7. Узел Kafka, подготовка файла конфигурации.

vim config/connect-standalone.properties

vim config/connect-ck-sink.properties

8. Запустите кластер kafka-connect.

bin/connect-standalone.sh config/connect-standalone.properties config/connect-ck-sink.properties

9.Кафка пишет номер теста

{"id":1,"name":"felixzh"}

10.Данные запроса ClickHouse

select * from kafka2ck;

Подвести итог

На этом этапе понятна семантика проекта clickhouse-kafka-connect «один и только один раз». реализация и внедрение кейс-практики завершены!