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

Community Maintained

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

1. Соберите данные для подключения

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

• ХОСТ и ПОРТ: как правило, порт 8443 при использовании TLS или 8123 при отсутствии TLS.

• ИМЯ БАЗЫ ДАННЫХ: по умолчанию существует база данных с именем default, используйте имя базы данных, к которой вы хотите подключиться.

• ИМЯ ПОЛЬЗОВАТЕЛЯ и ПАРОЛЬ: по умолчанию имя пользователя default. Используйте имя пользователя, подходящее для вашего случая.

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

Выберите HTTPS, и данные будут доступны в примере команды curl.

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

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

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

3. Загрузите драйвер JDBC для ClickHouse

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

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

1. Чтобы настроить службу контроллера в Apache NiFi, посетите страницу конфигурации потока NiFi, нажав на кнопку "шестигранник"

   

2. Выберите вкладку Службы контроллера и добавьте новую службу контроллера, нажав на кнопку + в верхнем правом углу

   

3. Найдите DBCPConnectionPool и нажмите на кнопку "Добавить"

   

4. Новая добавленная служба DBCPConnectionPool по умолчанию будет находиться в состоянии Неверно. Нажмите на кнопку "шестигранник", чтобы начать конфигурацию

   

5. В разделе "Свойства" введите следующие значения

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

6. В разделе Настройки измените имя службы контроллера на "ClickHouse JDBC" для удобства

   

7. Активируйте службу контроллера DBCPConnectionPool, нажав на кнопку "молния", а затем кнопку "Включить"

   
   
   
   

8. Проверьте вкладку Службы контроллера и убедитесь, что служба контроллера включена

   

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

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

   

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

   Свойство	Значение	Примечание
   Служба пулов подключения к базе данных	ClickHouse JDBC	Выберите службу контроллера, настроенную для ClickHouse
   SQL запрос SELECT	SELECT * FROM system.metrics	Впишите ваш запрос сюда

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

   

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

   

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

   

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

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

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

   Свойство	Значение	Примечание
   Читатель записей	JSONTreeReader	Выберите подходящий читатель записей
   Писатель записей	JSONReadSetWriter	Выберите подходящий писатель записей
   Минимальное число записей	1000	Измените это на большее значение, чтобы минимальное количество строк объединялось в одну запись. По умолчанию - 1 строка
   Максимальное число записей	10000	Измените это на большее число, чем "Минимальное число записей". По умолчанию - 1 000 строк

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

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

   

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

   

4. В разделе "Свойства" процессора PutDatabaseRecord введите следующие значения

   Свойство	Значение	Примечание
   Читатель записей	JSONTreeReader	Выберите подходящий читатель записей
   Тип базы данных	Generic	Оставьте по умолчанию
   Тип оператора	INSERT
   Служба пулов подключения к базе данных	ClickHouse JDBC	Выберите службу контроллера ClickHouse
   Имя таблицы	tbl	Введите имя вашей таблицы здесь
   Перевод имен полей	false	Установите в "false", чтобы имена полей, вставленных, должны соответствовать именам колонок
   Максимальный размер пакета	1000	Максимальное количество строк на вставку. Это значение не должно быть ниже значения "Минимальное число записей" в процессоре MergeRecord

5. Чтобы подтвердить, что каждая вставка содержит несколько строк, проверьте, что количество строк в таблице увеличивается как минимум на значение "Минимальное число записей", определенное в MergeRecord.

   

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