Подключение Apache NiFi к ClickHouse

Apache NiFi — это ПО с открытым исходным кодом для управления рабочими процессами, предназначенное для автоматизации потоков данных между программными системами. Оно позволяет создавать ETL-конвейеры данных и включает более 300 процессоров данных. В этом пошаговом руководстве показано, как подключить Apache NiFi к ClickHouse в качестве источника и пункта назначения, а также загрузить примерный набор данных.

Подготовьте сведения о подключении

Чтобы подключиться к ClickHouse по HTTP(S), вам понадобится следующая информация:

Параметр(ы)	Описание
`HOST` and `PORT`	Обычно используется порт 8443 при использовании TLS и 8123 без TLS.
`DATABASE NAME`	По умолчанию есть база данных `default`; используйте имя базы данных, к которой хотите подключиться.
`USERNAME` and `PASSWORD`	По умолчанию имя пользователя — `default`. Используйте имя пользователя, подходящее для вашего сценария использования.

Сведения о подключении для вашего сервиса ClickHouse Cloud доступны в консоли ClickHouse Cloud. Выберите сервис и нажмите Connect:

Выберите HTTPS. Сведения о подключении будут показаны в примере команды curl.

Сведения о подключении к ClickHouse Cloud по HTTPS

Если вы используете самоуправляемый ClickHouse, сведения о подключении задаёт ваш администратор ClickHouse.

Скачайте и запустите Apache NiFi

Для новой установки скачайте бинарный файл с https://nifi.apache.org/download.html и начните с выполнения команды ./bin/nifi.sh start

Скачайте JDBC-драйвер ClickHouse

Перейдите на страницу релизов JDBC-драйвера ClickHouse на GitHub и найдите последнюю версию JDBC-драйвера
В выбранном релизе нажмите “Show all xx assets” и найдите JAR-файл, содержащий ключевое слово “shaded” или “all”, например clickhouse-jdbc-0.5.0-all.jar
Поместите JAR-файл в папку, доступную Apache NiFi, и запишите его абсолютный путь

Добавьте службу контроллера DBCPConnectionPool и настройте её свойства

Чтобы настроить службу контроллера в Apache NiFi, откройте страницу конфигурации потока NiFi, нажав кнопку с шестерёнкой
Перейдите на вкладку Controller Services и добавьте новую службу контроллера, нажав кнопку + в правом верхнем углу
Найдите DBCPConnectionPool и нажмите кнопку “Add”
После добавления DBCPConnectionPool по умолчанию будет находиться в состоянии Invalid. Нажмите кнопку с шестерёнкой, чтобы перейти к настройке
В разделе “Properties” введите следующие значения

Property	Value	Remark
URL подключения к базе данных	jdbchttps://HOSTNAME:8443/default?ssl=truetrue	Замените HOSTNAME в URL подключения на нужное значение
Имя класса драйвера базы данных	com.clickhouse.jdbc.ClickHouseDriver
Расположение драйвера базы данных	/etc/nifi/nifi-X.XX.X/lib/clickhouse-jdbc-0.X.X-patchXX-shaded.jar	Абсолютный путь к JAR-файлу драйвера ClickHouse JDBC
Пользователь базы данных	default	Имя пользователя ClickHouse
Пароль	password	Пароль ClickHouse

В разделе Settings измените имя службы контроллера на “ClickHouse JDBC” для удобства дальнейшей работы
Активируйте службу контроллера DBCPConnectionPool, нажав кнопку с молнией, а затем кнопку “Enable”
Проверьте вкладку Controller Services и убедитесь, что служба контроллера включена

Чтение из таблицы с помощью процессора ExecuteSQL

Добавьте процессор ExecuteSQL вместе с соответствующими предшествующими и последующими процессорами

В разделе “Properties” процессора ExecuteSQL введите следующие значения

Property	Value	Remark
Database Connection Pooling Service	ClickHouse JDBC	Выберите Controller Service, настроенный для ClickHouse
SQL select query	SELECT * FROM system.metrics	Введите здесь свой запрос

Запустите процессор ExecuteSQL
Чтобы убедиться, что запрос был успешно обработан, проверьте один из FlowFile в выходной очереди
Переключитесь в режим “formatted”, чтобы просмотреть результат в выходном FlowFile

Запись в таблицу с помощью MergeRecord и PutDatabaseRecord процессоров

Чтобы выполнить вставку нескольких строк за одну операцию, сначала нужно объединить несколько записей в одну. Это можно сделать с помощью процессора MergeRecord

В разделе “Properties” процессора MergeRecord введите следующие значения

Свойство	Значение	Примечание
Record Reader	`JSONTreeReader`	Выберите подходящее средство чтения записей
Record Writer	`JSONReadSetWriter`	Выберите подходящее средство записи записей
Minimum Number of Records	1000	Увеличьте это значение, чтобы в одну запись объединялось минимально необходимое число строк. По умолчанию — 1 строка
Maximum Number of Records	10000	Укажите значение больше, чем “Minimum Number of Records”. По умолчанию — 1 000 строк

Чтобы убедиться, что несколько записей объединяются в одну, проверьте вход и выход процессора MergeRecord. Обратите внимание, что на выходе получается массив из нескольких входных записей Вход Выход

В разделе “Properties” процессора PutDatabaseRecord введите следующие значения

Свойство	Значение	Примечание
Record Reader	`JSONTreeReader`	Выберите подходящее средство чтения записей
Database Type	Generic	Оставьте значение по умолчанию
Statement Type	INSERT
Database Connection Pooling Service	ClickHouse JDBC	Выберите сервис-контроллер ClickHouse
Table Name	tbl	Укажите здесь имя таблицы
Translate Field Names	false	Установите значение “false”, чтобы имена полей при вставке совпадали с именами столбцов
Maximum Batch Size	1000	Максимальное количество строк на одну вставку. Это значение не должно быть меньше значения “Minimum Number of Records” в процессоре `MergeRecord`

Чтобы убедиться, что каждая вставка содержит несколько строк, проверьте, что число строк в таблице увеличивается как минимум на значение “Minimum Number of Records”, заданное в MergeRecord.
Поздравляем — вы успешно загрузили данные в ClickHouse с помощью Apache NiFi!