На связи Никита Скирдин, программист 1С компании «Белый код». В прошлой статье мы уже говорили о загрузке данных для системы BI-аналитики. В этой же статье разберем решение задачи с использованием Apache NiFi — системы для автоматизации потоков данных. Хотя NiFi позиционируется как ETL-инструмент (extract transform load), позволяющий внутри себя осуществить необходимые преобразования над поступающими данными, ничто не мешает нам использовать его также для ELT-процесса (extract load transform).
Разработка в NiFi
Разработка процесса получения, обработки и отправки данных в NiFi осуществляется с помощью расстановки и соединения между собой независимых обработчиков (Processor) на полотне (NiFi Flow или Canvas). Другими словами, для большинства задач построение процессов происходит в парадигме Low-Code, без написания какого-либо кода на полноценном языке программирования. Это позволяет использовать NiFi даже в том случае, если вы не знаете Java, на котором написан сам NiFi, или Python, на котором описываются процессы в Apache Airflow.
Для гибкой работы с данными файлов и атрибутами в NiFi применяется специальный язык выражений: NiFi Expression Language. С его помощью возможно изменять строки, выполнять математические и логические операции и многое другое. На практике это означает, что с помощью обработчиков UpdateAttribute и ReplaceText можно произвольным образом менять содержимое атрибутов и самого файла соответственно.
NiFi берет на себя вопросы, связанные с обеспечением целостности и восстанавливаемости данных в случае, например, сбоя системы, потери питания сервера и т. п. Гарантирование доставки данных, загруженных в NiFi, достигается за счет применения концепции файлов потока (Flow File), которые объединяют в себе содержимое (данные, загруженные в NiFi из внешней системы) и атрибуты (метаданные, которые можно добавлять к файлам в самом NiFi) и которые сохраняются на жесткий диск, чтобы не допустить их потери в случае отключения питания.
Кроме того, NiFi предоставляет возможность просмотреть файлы, которые он обрабатывает, что крайне полезно для отладки системы в процессе разработки, поскольку позволяет на каждом этапе увидеть, как именно изменяются содержимое и атрибуты обрабатываемых данных.
На рисунке выше представлен интерфейс NiFi. Сверху мы видим панель действий, с помощью которой можно добавить на схему новые элементы (процессоры, группы, комментарии и т. п.). Слева сверху расположена панель навигации и действий над текущими выделенными объектами. По центру мы видим полотно с находящимися на нем обработчиками. Данная группа процессов позволяет преобразовать один Excel-файл с множеством листов в несколько CSV-файлов, по одному CSV-файлу на каждый лист.
Помимо возможности просмотреть содержимое и атрибуты файла на каждом этапе его обработки, NiFi позволяет увидеть и происхождение, и весь путь каждого файла по конвейеру обработки. Для этого необходимо зайти в раздел Data Provenance, где будут отображаться все события обработки файлов.
У любого файла можно также просмотреть его путь от загрузки до окончания обработки, для этого необходимо нажать на три точки у любого события и выбрать пункт Show Lineage:
Описанное в данной статье — это далеко не все возможности NiFi. Вместе с возможностью горизонтального масштабирования за счет кластеризации NiFi становится полезным инструментом в обработке данных любой сложности и любых объемов. При этом начать пользоваться NiFi довольно легко, поскольку для старта будет достаточно использования уже готовых компонент, а при необходимости можно дописать свои компоненты на Java.
Интеграция с ClickHouse
Чтобы интегрировать NiFi с ClickHouse, необходимо скачать официальный JDBC драйвер с открытым исходным кодом. Так как мы запускали NiFi через docker, то установка драйвера в контейнер выполнялась с помощью небольшого Dockerfile:
```
FROM apache/nifi:latest
USER root
RUN mkdir -p /opt/jar/
RUN curl -Lo '/opt/jar/clickhouse-jdbc-0.8.0-shaded-all.jar' 'https://github.com/ClickHouse/clickhouse-java/releases/download/v0.8.0/clickhouse-jdbc-0.8.0-shaded-all.jar'
USER nifi
```Затем в самом NiFi необходимо было настроить сервис для соединения с базой данных. Параметры сервиса DBCPConnectionPool выглядит следующим образом:
Основное отличие от настроек для СУБД, драйвера для которых встроены в NiFi, — это необходимость указать имя класса драйвера (Database Driver Class Name), а также путь к jar-файлу этого драйвера. Более подробно про подключение можно прочитать в официальной документации ClickHouse.
В общем процесс загрузки содержимого CSV-файлов в ClickHouse выглядит следующим образом:
После сохранения содержимого CSV в файл, который может читать ClickHouse (директория /var/lib/clickhouse/user_files), и установки необходимых для формирования запросов атрибутов (определения таблицы и партиции из имени файла) выполняются последовательно запросы для создания таблицы, удаления старых данных и вставки новых. Процесс завершается удалением файла из директории пользовательских файлов ClickHouse.
Сравнение с Airflow
Основное, что следует отметить при сравнении NiFi и Airflow — это разные ключевые концепции этих программных продуктов. Если в NiFi все выстроено вокруг концепции файлов потока, которые обрабатываются процессорами, то в Airflow ключевыми элементами являются ациклические графы (DAG) и задачи на этих графах. Это различие отражается и в возможностях мониторинга выполнения обработки: в NiFi мы можем увидеть историю каждого обработанного файла, в Airflow же мы наблюдаем за выполнением определенных задач из графа, при этом каждая задача может обрабатывать сразу все данные, как в случае запуска SQL-скриптов, которые выполняют действия над таблицами в целом.
Еще одно важное различие заключается в том, что в отличие от Airflow, NiFi является low-code платформой, для использования которой нет необходимости в сотрудниках, знающих Java или Python.
Замеры времени
Для проверки времени обработки CSV-файлов в NiFi и Airflow были отправлены файлы общим размером 2.61 ГБ. Обе системы отправляли файлы в пустую базу данных Clickhouse. CSV-файлы читались самим Clickhouse с помощью FROM file(‘filepath’), на стороне NiFi и Airflow выполнялась предварительная обработка: получение архива с файлами с FTP, распаковка архива, перемещение файлов в директорию /var/lib/clickhouse/user_files и создание и очистка таблиц. Все сервисы запускались внутри docker-контейнеров.
Airflow справился с обработкой файлов за 48 сек., а NiFi — за 40 сек. По результатам можно отметить, что обе системы справляются с обработкой за сопоставимое время.
Выводы
Хотя и NiFi, и Airflow прекрасно справляются с задачами ETL и ELT, выбор инструмента зависит от конкретного проекта. Если нужна потоковая обработка данных с гарантией того, что один раз поступившие в систему данные точно дойдут до места назначения, либо в случае ошибки будут сохранены в ней, то ваш выбор — Apache NiFi. Если же в приоритете управление различными процессами, которые запускаются по расписанию или в случае внешнего события, особенно когда необходимо обрабатывать данные порциями в разрезе временных интервалов, то лучше использовать Apache Airflow.
Кстати, мы развиваем сообщество в Телеграме «Шины не для машины». Это чат, где говорим про отечественные ESB, их преимущества, сложности внедрения, кейсы. В сообществе можно пообщаться напрямую с вендорами, например, с представителями DATAREON, Factor-ESB, USEBUS и другими.
