Соединение Apache NiFi с ClickHouse

Community Maintained

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

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

To connect to ClickHouse with HTTP(S) you need this information:

• The HOST and PORT: typically, the port is 8443 when using TLS or 8123 when not using TLS.

• The DATABASE NAME: out of the box, there is a database named default, use the name of the database that you want to connect to.

• The USERNAME and PASSWORD: out of the box, the username is default. Use the username appropriate for your use case.

The details for your ClickHouse Cloud service are available in the ClickHouse Cloud console. Select the service that you will connect to and click Connect:

Choose HTTPS, and the details are available in an example curl command.

If you are using self-managed ClickHouse, the connection details are set by your ClickHouse administrator.

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

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

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

1. Перейдите на страницу релизов драйвера ClickHouse JDBC на GitHub и найдите последнюю версию релиза JDBC
2. В релизной версии нажмите на "Показать все xx assets" и найдите файл 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-файлу драйвера ClickHouse JDBC
Пользователь базы данных	default	Имя пользователя ClickHouse
Пароль	password	Пароль ClickHouse

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

   

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

   
   

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

   

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

1. Добавьте процессор ​​ExecuteSQL, а также соответствующие upstream и downstream процессоры

   

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!