Сообщество разработчиков Apache NiFi регулярно радует новыми выпусками. Не успели мы полностью освоить январский релиз 2022, в начале марта появилась еще более свежая версия этого потокового маршрутизатора. Самое главное в Apache NiFi 1.16.0 для дата-инженера и администратора кластера.
Главные новинки Apache NiFi 1.16.0
Apache NiFi 1.16.0 включает несколько десятков улучшений, новых фич и исправленных ошибок. Особенно значимыми изменениями можно назвать следующие:
- поддержка повторных попыток обработки потоков данных на уровне платформы с возможностями их настройки в пользовательском интерфейсе и API;
- отключенные узлы теперь могут снова присоединяться к кластеру с обновлением потока после локальной записи резервной копии потока отключенного узла;
- кластер NiFi позволяет вносить изменения в поток данных даже при наличии отключенных узлов;
- новые компоненты для чтения сообщений CEF, простого выполнения геохеширования записей и прослушивания ловушек SNMP, что значительно упрощает обогащение данных, особенно при работе с веб-сервисами;
- отображение дополнительной информации в веб-GUI – пользовательский интерфейс теперь показывает, когда свойства компонента имеют начальные или конечные пробелы, что полезно при копировании значения из другого места и вставке его в NiFi. Также в GUI теперь отображается, когда в очереди скопились данные и в постоянном перезапуске процессора нет смысла – процессор не будет запускаться миллионы раз в минуту, чтобы ничего не выполнить.
- MiNiFi Java теперь может загружать nars в горячем режиме, как и NiFi;
- множество исправлений ошибок и улучшений для процессоров, которые взаимодействуют с облачными сервисами на AWS, Azure и GCP, а также с SFTP, FTP, Kafka и пр.;
- улучшение Stateless благодаря улучшенной изоляции загрузчика классов и возможности запускать только локальные nars-пакеты;
- повышение безопасности библиотек Jetty, Netty, log4j, slf4j и пр. с удобным механизмом автоматического обновления старой базы данных.
Рассмотрим некоторые из отмеченных обновлений более подробно.
Улучшение аутентификации
Недавно мы писали про OIDC-аутентификацию пользователей Apache NiFi. В версии 1.16.0 исправлена ошибка с тем, что провайдер OIDC должен использовать хранилище ключей NiFi и хранилище доверенных сертификатов. Провайдер OIDC использует общие HTTPS-запросы к OIDC IdP, но не настраивает эти запросы для использования хранилища ключей или доверенных сертификатов NiFi. Скорее, он использует хранилище ключей и доверенных сертификатов JVM по умолчанию, что приводит к трудностям при отладке PKIX и других ошибок согласования TLS. Его следует переключить на использование хранилища ключей и доверенных сертификатов NiFi, как это делают другие службы инфраструктуры NiFi.
С этим же связано улучшение провайдера удостоверений OIDC, направленное на устранение следующих проблем:
- недостаток логов для отладки проблем с IdP;
- грамматические и синтаксические ошибки в логах и сообщениях об ошибках;
- некорректная обработка утверждений в ответе провайдера удостоверений.
Здесь же отметим связанное улучшение интеграции службы контроллера OAuth2 с InvokeHTTP. В NiFi есть служба контроллера для запроса токенов OAuth2, но на нее невозможно было сослаться ни в одном процессоре. Для этого было внесено изменение в процессор InvokeHTTP, чтобы он мог использовать эту службу контроллера. Служба контроллера использует предварительно определенный идентификатор клиента и секрет клиента, чтобы сделать запрос к серверу доступа OAuth для получения токена. Затем токен будет использоваться InvokeHTTP для выполнения запросов.
Завершая тему про аутентификацию, отметим про добавление свойства окна обновления к поставщику токена OAuth2 в новой версии Apache NiFi. Теперь служба контроллера StandardOauth2AccessTokenProvider поддерживает внутреннюю проверку срока действия токена на основе свойства expiresIn текущего AccessToken. Внедрение службы провайдера включало жестко закодированное число, вычитаемое из свойства expiresIn, чтобы запрашивать обновление токена до точного времени истечения срока действия.
Для поддержки как краткосрочных, так и долгосрочных сценариев истечения срока действия токена в StandardOauth2AccessTokenProvider необходимо добавить новое свойство службы контроллера. Свойство должно предоставлять настраиваемую продолжительность, в течение которой служба будет пытаться обновить токен. В предыдущих выпусках, до введения StandardOauth2AccessTokenProvider класс AccessToken реализовывал проверку isExpired на основе точного времени истечения срока действия. Поскольку этот класс является частью API службы OAuth2, это поведение восстановлено, а новое свойство окна обновления реализовано в самой службе контроллера.
Пара новых фич для дата-инженера
С точки зрения практической дата-инженерии из новых фич Apache NiFi 1.16.0 хочется отметить дополнение к существующему процессору DetectDuplicate. Разница в том, что новый процессор DetectDuplicateRecord работает на уровне записи, кэшируя записи из каждого входящего FlowFile. Он определяет, была ли уже просмотрена кэшированная запись. Имя определяемых пользователем свойств определяет значения RecordPath, используемые для определения уникальности записи. Если пользовательские свойства отсутствуют, вся запись используется в качестве входных данных для определения уникальности. Все повторяющиеся записи направляются в «дубликаты». Если запись не определена как дубликат, обработчик направляет запись как «не дублирующую».
Этот процессор делает доступными две различные структуры данных фильтрации в зависимости от уровня точности и количества записей, которые пользователь хочет обработать:
- фильтр HashSet гарантирует 100% обнаружение дубликатов за счет хранения одного хэша на запись;
- фильтр BloomFilter будет использовать эффективное/постоянное пространство благодаря вероятностным гарантиям. Это полезно при обработке очень большого количества записей, и некоторые ложные срабатывания допустимы, т.е. некоторые записи могут быть помечены как дубликаты, даже если они не были просмотрены ранее.
Еще интересным кажется новый процессор GeohashRecord. Geohash может использоваться в качестве уникальных идентификаторов для геопространственных данных и применяется при индексировании для эффективных пространственных запросов, таких как запросы диапазона и локальный поиск. Новый процессор позволяет пользователям NiFi кодировать значения широты/долготы в Geohash и декодировать геошэширование в координаты широты/долготы. Процессор GeohashRecord на основе записей добавлен вместе с функциями языка выражений и его можно применять к данным, ориентированным на записи, с предопределенной схемой. А функции языка выражений будут эффективными в сочетании с другими процессорами, такими как InvokeHTTP.
Наконец, добавлены новые процессоры для объединения данных, ориентированных на записи, с данными обогащения. Ранее существовавшие процессоры LookupRecord и GeoEnrichRecord не обеспечивали всех вариантов использования, например, обогащение данных по мере их прохождения через систему через обращение к внешней веб-службе без отправки в нее полезной нагрузки. В этом случае целесообразно преобразовать полезную нагрузку в запрос и отправить его внешнему веб-сервису, получить результат этого вызова и использовать его для обогащения исходных данных. До версии 1.16.0 NiFi не имел удобного механизма для этого. Поэтому разработчики добавили два дополнительных процессора: ForkEnrichment и JoinEnrichment.
ForkEnrichment используется для создания клона входящего FlowFile, присваивая соответствующие атрибуты оригиналу и клону, а затем отправляя каждый из них в разные отношения (оригинал и обогащение). Данные, отправленные в «обогащенное» соединение, затем могут быть преобразованы во все, что необходимо для отправки в качестве запроса веб-службе. Затем результат будет передан процессору JoinEnrichment, который должен получить входные данные от «исходного» соединения и соединения «обогащения» и объединить записи вместе.
Это позволит реализовать следующие кейсы для соединения записей:
- сопоставление записей по их индексу в FlowFile с помощью обертки, когда ее элемент инкапсулирует первую запись из исходного FlowFile и первую запись из обогащенного;
- сопоставление записей по их индексу в FlowFile и вставка обогащенной записи в исходную полезную нагрузку;
- выполнение SQL-запроса с предложением JOIN для объединения записей на основе некоторого поля в данных.
Другие полезные улучшения и новые фичи свежего выпуска Apache NiFi мы рассмотрим в следующий раз. Читайте в нашей новой статье, что такое NiFi Flow Design System, чем это полезно и как использовать.
А освоить все тонкости администрирования и эксплуатации Apache NiFi для эффективной аналитики больших данных вы сможете на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
