Как написать свой Controller Service в Apache NiFi

Что такое Controller Service в Apache NiFi и как дата-инженеру создать собственный набор настроек для совместного и повторного использования в потоковом конвейере обработки данных.

Apache NiFi реализует потоковую парадигму обработки информации, выполняя ETL-операции над FlowFile с помощью обработчиков, называемыми процессорами. Если какие-то процессоры работают по одинаковым правилам, например, для взаимодействия с внешними системами, с точки зрения разработки логично применить к этому принцип DRY (Don’t Repeat Yourself). Объединить общие настройки, чтобы использовать их для нескольких процессоров, в Apache NiFi позволяют службы контроллеров (Controller Service). Таким образом, ключевое назначение Controller Service в NiFi напоминает подключаемую библиотеку, плагин или настройки для соединения с внешними сервисами в Apache AirFlow.

Определив в Controller Service правила подключения к внешней системе, алгоритм шифрования или сериализации/десериализации данных, его можно использовать в рамках всего фреймворка или для набор процессоров, объединенных в группу (ProcessGroup). В официальной документации Apache NiFi отмечено, что Controller Service – это общая служба, которая может использоваться задачами отчетности, процессорами и другими службами для настройки или выполнения задач. Поскольку Controller Service является полноценным объектом выполнения задач, то все процессоры, которые к нему обращаются, разделяют его ресурсы. Это особенно важно, если ресурсы ограничены, например, как количество открытых соединений с базой данных. Если у Controller Service задан лимит таких соединений, то процессоры, использующие эту службу, могут конфликтовать при выполнении подключения с помощью сервиса, если их количество больше заданного предела.

Чтобы изменить Controller Service, надо остановить все ссылающиеся на него компоненты. В потоке данных, чтобы определить область действия Controller Service и ограничить его доступность, службу контроллера можно создать в любой группе процессов. Службы контроллера, созданные в группе процессов, будут доступны всем дочерним компонентам. Чтобы Controller Service был доступен для всех процессоров в потоке данных, надо определить его в конфигурации корневой группы процессов или групп подпроцессов, где они будут использоваться. Для защищенных экземпляров NiFi пользователи могут просматривать и добавлять службы контроллера в зависимости от своих привилегий. При отсутствии доступа к Controller Service они не будут отображаться в пользовательском интерфейсе. Права доступа могут назначаться на глобальном уровне или для конкретного Controller Service. Обычно в мультиарендной среде разрешения пользователей обычно привязываются к группе процессов. Если область действия службы контроллера определяется таким же образом, ее использование соответственно ограничивается. Внутри контроллера — инфраструктуры, которая выполняет поток данных, доступность Controller Service ограничена определенными задачами отчетности и другими службами. Эта область видимости работает совместно с областью действия, реализованной в контексте потока данных. Пользователи с разрешениями на задачи отчетности также будут иметь доступ к определенным здесь службам. Добавить Controller Service можно не только для процессора, но и для всего потока данных, сделав это в веб-GUI Apache NiFi.

После добавления службы контроллера ее можно настроить в веб-GUI NiFi.

Также дата-инженер может написать свою службу контроллера, чтобы потом ее использовать. Как это сделать, рассмотрим далее.

Для разработки своего Controller Service необходимо использовать интерфейс ControllerService. Он позволяет делиться функциональностью и состоянием в JVM. Этот интерфейс похож на интерфейс процессора, о котором мы писали здесь. Но он не имеет метода onTrigger(), поскольку для Controller Service не поддерживается периодический запуск и непосредственная интеграция в поток обработки данных. Поэтому службы контроллера не имеют связей (отношений).

Как и в случае с интерфейсом процессора, интерфейс ControllerService предоставляет методы для настройки, проверки и инициализации. Все эти методы идентичны методам интерфейса Processor, кроме того, что методу initialize() передается контекст ControllerServiceInitializationContext, а не ProcessorInitializationContext.

Controller Service долен состоять из интерфейса, расширяющего ControllerService, чтобы взаимодействовать с реализациями через их интерфейс. Например, процессору не предоставлена конкретная реализация ControllerService, а потому он должен ссылаться на сервис только через интерфейсы, расширяющие ControllerService. Это ограничение обусловлено тем, что процессор может существовать в одном NAR-архиве NiFi, а реализация службы контроллера, в которой находится процессор, находится в другом NAR. Чтобы фреймворк мог переключиться на соответствующий загрузчик классов ClassLoader и вызвать нужный метод в конкретной реализации, открытые интерфейсы реализуются динамически. Но при этом процессор и реализация службы контроллера должны использовать одно и то же определение интерфейса ControllerService. Таким образом, оба NAR должны зависеть от того NAR, в котором находится интерфейс службы контроллера.

Получить ControllerService можно процессором, другой службой контроллера, параметром-провайдером или задачей отчетности (ReportingTask) с помощью ControllerServiceLookup или с помощью метода identifiesControllerService() класса PropertyDescriptor Builder. ControllerServiceLookup может быть получен процессором из контекста инициализации ProcessorInitializationContext, который передается методу initialize(). Аналогично из контекста самой службы контроллера ControllerServiceInitializationContext, а также с помощью ParameterProvider из ParameterProviderInitializationContext и с помощью ReportingTask через объект ReportingConfiguration, переданный методу initialize(). В большинстве случаев использование метода identifiesControllerService() класса PropertyDescriptor Builder является более предпочтительным и простым способом. Чтобы использовать этот метод, необходимо создать объект описания свойств PropertyDescriptor, который ссылается на службу контроллера. Например, следующий Java-код создает SSLContextService – интерфейс, расширяющий ControllerService для выбора службы контекста SSL из меню в графическом интерфейсе Apache NiFi:

Для использования выбранной службы в процессоре надо реализовать следующий код:

Узнайте больше про администрирование и использование Apache NiFi для построения эффективных ETL-конвейеров потоковой аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:

Статья:

Курс: