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

Community Maintained

Apache NiFi

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

Соберите сведения о подключении

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

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

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

Выберите HTTPS. Параметры подключения отображаются в примере команды curl.

Если вы используете самостоятельное (self-managed) развертывание ClickHouse, параметры подключения задаются администратором ClickHouse.

Загрузите и запустите Apache NiFi

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

Загрузите драйвер ClickHouse JDBC

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

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

1. Чтобы настроить Controller Service в Apache NiFi, перейдите на страницу NiFi Flow Configuration, нажав кнопку с иконкой шестерёнки

   

2. Выберите вкладку Controller Services и добавьте новый Controller Service, нажав кнопку + в правом верхнем углу

   

3. Найдите DBCPConnectionPool и нажмите кнопку Add

   

4. Только что добавленная служба контроллера DBCPConnectionPool по умолчанию будет находиться в состоянии Invalid. Нажмите кнопку с иконкой шестерёнки, чтобы начать настройку

   

5. В разделе Properties введите следующие значения

Property	Value	Remark
Database Connection URL	jdbc:ch:https://HOSTNAME:8443/default?ssl=true	Замените HOSTNAME в URL подключения соответствующим значением
Database Driver Class Name	com.clickhouse.jdbc.ClickHouseDriver
Database Driver Location(s)	/etc/nifi/nifi-X.XX.X/lib/clickhouse-jdbc-0.X.X-patchXX-shaded.jar	Абсолютный путь к JAR-файлу JDBC-драйвера ClickHouse
Database User	default	Имя пользователя ClickHouse
Password	password	Пароль ClickHouse

6. В разделе Settings измените имя Controller Service на ClickHouse JDBC для удобства

   

7. Активируйте Controller Service DBCPConnectionPool, нажав кнопку с иконкой молнии, а затем кнопку Enable

   
   
   
   

8. Проверьте вкладку Controller Services и убедитесь, что Controller Service включён

   

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

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

   

2. В разделе "Properties" процессора ExecuteSQL задайте следующие значения

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

3. Запустите процессор ExecuteSQL

   

4. Чтобы убедиться, что запрос был успешно обработан, изучите один из FlowFile в выходной очереди

   

5. Переключите представление в режим "formatted", чтобы просмотреть результат выходного FlowFile

   

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

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

2. В разделе «Properties» процессора MergeRecord введите следующие значения

   Property	Value	Remark
   Record Reader	JSONTreeReader	Выберите соответствующий читатель записей
   Record Writer	JSONReadSetWriter	Выберите соответствующий писатель записей
   Minimum Number of Records	1000	Измените это значение на большее, чтобы минимальное количество строк объединялось в одну запись. По умолчанию — 1 строка
   Maximum Number of Records	10000	Измените это значение на большее, чем «Minimum Number of Records». По умолчанию — 1000 строк

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

   Входные данные

   

   Выходные данные

   

4. В разделе «Properties» процессора PutDatabaseRecord введите следующие значения

   Property	Value	Remark
   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

5. Чтобы убедиться, что каждая вставка содержит несколько строк, проверьте, что количество строк в таблице увеличивается как минимум на значение «Minimum Number of Records», определенное в MergeRecord.

   

6. Поздравляем — вы успешно загрузили данные в ClickHouse с помощью Apache NiFi!