Зачем нужен Controller Service в Apache NiFi и как его написать

Что такое 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.

Добавление службы контроллера в Apache NiFi

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

Настройка службы контроллера в Apache NiFi

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

Эксплуатация Apache NIFI

Код курса

NIFI3

Ближайшая дата курса

13 мая, 2024

Продолжительность

24 ак.часов

Стоимость обучения

72 000 руб.

Разработка своей службы контроллера

Для разработки своего 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:

public static final PropertyDescriptor SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder()
  .name("SSL Context Service")
  .description("Specified the SSL Context Service that can be used to create secure connections")
  .required(true)
  .identifiesControllerService(SSLContextService.class)
  .build();

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

final SSLContextService sslContextService = context.getProperty(SSL_CONTEXT_SERVICE)
  .asControllerService(SSLContextService.class);