29 января 2024 года вышла очередная веха 2-ой версии Apache NiFi, которая включает ряд новых функций и существенных обновлений зависимостей, а также несколько критических изменений. Рассмотрим самые интересные из них.
Новые процессоры Apache NiFi 2.0.0-M2
С точки зрения управления версиями, веха рассматривается как некоторое значимое обновление, контрольная точка, меняющая дальнейшее развитие проекта. Как правило, изменения из нескольких вех объединятся в одной версии. Первая веха 2-го релиза Apache NiFi вышла в ноябре 2023 года, о чем мы писали здесь. Спустя 2 месяца вышла 2-я веха с 4-мя новыми службами контроллера и 7-ю процессорами, а также 12 обновлениями зависимостей и критическими изменениями.
В части зависимостей обновлены Spring Framework до 6 версии, Jetty до 12, Jakarta Servlet API до 6, Jakarta XML Binding до 4, аннотации Swagger до 2 и спецификации OpenAPI до 3.0. Среди критических обновлений удалены модули сервера MiNiFi C2 и конфигурация образа Docker, процессоры InfluxDB и службы уведомлений Bootstrap, процессоры JSON-преобразований JoltTransformJSON и JoltTransformRecord перемещены из пакета nifi—standard—nar в nifi—jolt—nar. Вместо класса SimpleDateFormat для анализа и форматирования даты и времени теперь используется DateTimeFormatter. Из-за перехода с Servlet API 3 на версию 6 потребуются обновления и перекомпиляция пользовательских расширений GUI.
С точки зрения дата-инженера наиболее интересными решениями стали внедрения новых компонентов. В частности, добавлен процессор PutMongoBulk, который обеспечивает массовую запись данных в документо-оринетированную СУБД MongoDB. Ранее существовавший в NiFi процессор PutMongo мог обновлять (или добавлять) только одну запись за раз, что очень неэффективно при массовых операциях. Другое процессор, PutMongoRecord использует настроенное средство чтения записей и схему для чтения входящего набора записей из тела FlowFile, а затем вставляет пакеты этих записей в настроенную коллекцию MongoDB. Это предполагает подробное знание структуры вставляемых JSON-документов. Поэтому новый процессор PutMongoBulk использует API самой MongoDB — BulkWrite, объединяя несколько операций в одну без каких-либо предположений о структуре самих вставляемых/изменяемых документов. Таким образом, MongoDB обрабатывает все за одну операцию, что снижает нагрузку и повышает эффективность ETL-процессов.
Для повышения общей эффективности обработки очень больших Parquet-файлов добавлено 2 новых процессора: CalculateParquetOffsets и CalculateParquetRowGroupOffsets. Первый генерирует N файлов потока из входных данных и добавляет атрибуты со смещениями, необходимыми для чтения группы строк в содержимом FlowFile. Процессор CalculateParquetRowGroupOffsets генерирует один FlowFile из каждой группы строк входных данных и добавляет атрибуты со смещениями, необходимыми для чтения группы строк в содержимом FlowFile. Также для работы с Parquet-файлами добавлен процессор SplitParquet, который принимает в качестве параметра количество записей для разделения FlowFile. На выходе процессор генерирует несколько FlowFile с неизмененным содержимым и добавляет атрибуты со смещениями, необходимыми для чтения группы строк в содержимом.
С разделением связан еще один новый процессор Apache NiFi – FilterAttribute, который определяет, какие атрибуты FlowFile должны существовать в выбранных точках потока и уменьшает их набор. Это похоже на работу процессора процессора UpdateAttribute, который определяет набор удаляемых атрибутов только по заданному регулярному выражению. А FilterAttribute предполагает явное указание набора атрибутов, которые необходимо сохранить, что проще задания регулярного выражения. Также в Apache NiFi 2.0.0-M2 добавлены процессоры для взаимодействия с SaaS-решением обработки обращений Zendesk и публикации сообщений в канал Slack.
Новые службы контроллера
Напомним, службы контроллеров (Controller Service) в Apache NiFi используются для сохранения настроек подключения к внешним системам, чтобы не повторять это в каждом отдельном процессоре, выполняющем обработку принимаемой или передаваемой информации. Именно Controller Service содержит правила подключения к внешней системе, алгоритм шифрования или сериализации/десериализации данных. Поскольку это общая служба, которая может использоваться задачами отчетности, процессорами и другими службами для настройки или выполнения задач, все процессоры, которые к нему обращаются, разделяют его ресурсы. Это важно помнить, когда эти ресурсы ограничены, например, как количество открытых соединений с базой данных. Поэтому даже при установке лимита в Controller Service на подобные ресурсы, процессоры, использующие эту службу, могут конфликтовать при выполнении подключения с помощью сервиса, если их количество больше заданного предела.
Из новых служб контроллера в Apache NiFi 2.0.0-M2 хочется особенно выделить StandardJsonSchemaRegistry, которая позволяет использовать один процессор для проверки всей схемы JSON. Ранее существовавший процессор ValidateJSON проверяет содержимое FlowFile на соответствие настраиваемой схеме JSON. Он не поддерживает входные данные, содержащие несколько объектов JSON, например JSON с разделителями новой строки. Если входной FlowFile содержит JSON с разделителями новой строки, будет проверена только первая строка. Поэтому использование ValidateJSON требует выделенного экземпляра процессора для каждой схемы JSON-документа, если их несколько. Это довольно трудоемко в строительстве и обслуживании, а также требует больших ресурсов. А поскольку использование реестра схем считается хорошей практикой, недостатки ValidateJSON устранены с помощью StandardJsonSchemaRegistry.
Также со реестр схем связана еще одна новая служба контроллера – DatabaseTableSchemaRegistryService. Она использует общедоступный Java-интерфейс DBCPService
и имя таблицы, извлекает метаданные таблицы из базы данных и создает из нее RecordSchema, чтобы получить определение таблицы из базы данных в виде схемы записи NiFi. Это позволяет гарантировать, что данные в потоке соответствуют определению таблицы, а не полагаются на преобразования типов и прочие особенности конкретного решения.
Наконец, для интерфейса службы FileResourceService, которая предоставляет файловый ресурс для других компонентов при условии доступа и наличия разрешения у NiFi на чтение, добавлена реализация на основе Google Cloud Storage, чтобы получать данные непосредственно из облачного хранилища.
Таким образом, с каждым новым выпуском Apache NiFi становится все удобнее и имеет больше функциональных возможностей для проектирования и реализации потоковых ETL-конвейеров. О новинках очередного выпуска этого мажорного релиза читайте в новой статье.
Узнайте больше про администрирование и использование Apache NiFi для построения эффективных ETL-конвейеров потоковой аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
