Бронислав Житников — NiFi. Пишем код для codeless-системы

Сложность: 2/3 (Если NiFi не
трогали - пропускайте данный доклад. Это отличное подспорье для углубленного знакомства с NiFi)

📺: https://youtu.be/Lnta1yPIMrY?si=cuGJLbmET3cQvadf
📺 :VK

🔍 Условия:

1️⃣ В NiFi может не хватать встроенных процессоров
2️⃣ Иногда нужно избежать приземления данных на диск и для этого несколько процессоров собрать в один
3️⃣ Встроенный процессор иногда стоит доработать, либо исправить в нем ошибки

⚙️ Путь к необходимости писать код:

1️⃣ ExecuteStreamCode — позволяет запустить внешний скрипт
2️⃣ ExecuteScript — дает доступ к внутренним фишкам NiFi
3️⃣ ScriptedService — позволяет изменить часть логики
4️⃣ ScriptedProcessor — позволяет прописать всю логику процессора

🛠 Устройство NiFi:
Для написания кода в большинстве случаев придется работать с NiFi API (Processors&Service)

1️⃣ ScriptEngine: докладчик рекомендует писать скрипты сразу на Groovy, так как Jython не даст свободы использования Python и при этом даст ограничения в работе с API
2️⃣ Можно писать свои модули сразу на Java
3️⃣ В версии 2.0 ожидается написание модулей на Python

⚙️ Шаги создания Processor:

1️⃣ Чтение Developers Guide
2️⃣ Применение Maven-Archetype — быстрый старт разработки, есть только часть процессоров и сервисов
3️⃣ Заполняем аннотации — могут определять поведение процессора, наполнять документацию, определять ключевые методы
4️⃣ Заполняем OnScheduled
5️⃣ Заполняем OnTrigger

⚙️ Аннотации поведения:

1️⃣ @InputRequirement — требует ли входящей очереди (обязательна, допустима, недопустима)
2️⃣ @TriggerWhenEmpty — требуется ли FlowFile. Позволяет запускаться процессору без файлов во входящей очереди. Для процессоров без входящих очередей не имеет смысла
3️⃣ @PrimaryNodeOnly — запуск только на ведущей ноде. Пригодится для процессоров без входящих очередей
4️⃣ @SupportBatching — возможность запускать длительный Run, снимая нагрузку с планировщика

⚙️ Аннотации документирования:

1️⃣ @CapabilityDescription — описание процессора
2️⃣ @UseCase — редко встречается, но очень помогают пользователям
3️⃣ @ReadAtrribute/@WriteAttribute — описывает, какие атрибуты приходят на вход и оказываются на выходе из процессора
4️⃣ @SeeAlso — справка о дополнительных источниках

📝 Расширенное документирование:
1️⃣ в каталоге resoursces создать docs, там создать каталог с полным именем процессора
2️⃣ Создать файл additionalDetails.html

🪛 Важные элементы разработки:

1️⃣ Properties — важно определить, как параметры обрабатываются; работают ли они с сервисами и с какими; определить наличие dynamic properties; настроить валидаторы и обязательность указания параметров
2️⃣ Relationships — заполнить имя, описание, обязательно указать настройку "выключен по умолчанию"
3️⃣ Lifecycle — c @OnAdded начинается создание экземпляра процессора;
@Onscheduled — запуск процессора, при перезапуске NiFi так же вызывается
@OnUnscheduled — остановка процессора, если его запуск больше не запланирован. Часть тредов могут быть активны после вызова метода с этой аннотацией. Лучше избегать применения в логике для выключения процессора
@OnStopped — следует применять эту аннотацию к методу, срабатывающему при полной остановке процессора
@OnRemoved — удаление экземпляра процессора
@OnShutdown — избегать использования в логике, так как NiFi чаще останавливается нештатно

⚙️⚙️ OnTrigger — метод построения логики работы процессора

1️⃣ Проверить наличие файла
2️⃣ Помнить об экземплярах вызова, изменяемые данные должны быть внутри OnTrigger
3️⃣ Расчет только из атрибутов
4️⃣ Не изменять атрибуты во время изменения контента
5️⃣ Не забывать отправлять файлы в следующие отношения или удалять
6️⃣ session.penalize — позволяет определить период, когда файл не подлежит обработке, используется для обработки ошибок
7️⃣ context.yield — в случае отсутствия отклика БД процессор может не запускаться какое-то время, даже если расписание на него выставлено
8️⃣ ProcessSession — управляет обработкой файла
9️⃣ ProcessContext — определяет доступ к информации из процессора и среды