Подключите 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. Посетите страницу релиза JDBC драйвера ClickHouse на 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 * 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!