В этой статье для дата-инженеров разберемся, что такое NiFi Scripted Components и как они заполняют пробел между скриптами и пользовательскими компонентами: процессорами, контроллерами, сообщениями и средствами их чтения/записи. Рассмотрим примеры скиптовых процессоров и сервисов, а также определим реальные достоинства и недостатки этих компонентов.
Почему просто скриптовых процессоров Apache NiFi недостаточно?
С практической точки зрения главным плюсом Apache NiFi для дата-инженера является большое количество готовых к использованию компонентов, особенно обработчиков потоковых файлов, т.е. процессоров. Фреймворк предоставляет более 300 готовых процессоров, но дата-инженер их может написать и самостоятельно, о чем мы уже упоминали здесь. Такой собственноручно написанный процессор может объединять несколько готовых, а также выполнять уникальные операции с нетиповыми источниками, реализуя оригинальную бизнес-логику обработки потокового файла. Для таких случаев в NiFi есть процессор ExecuteScript, о чем мы недавно писали. Он выполняет пользовательский сценарий, поддерживающий движки для разных языков программирования (Clojure, ECMAScript, Groovy, lua, Python, Ruby) с учетом FlowFile и сеанса процесса.
Более простой альтернативой для пары bash-команд пригодится процессор ExecuteStreamCommand, который выполняет внешнюю команду для содержимого потокового файла и создает новый FlowFile с результатами выполнения команды. Для более сложного поведения нужны процессоры ExecuteGroovyScript или ExecuteScript, но для них характерны следующие недостатки:
- только два исходящих отношения;
- проблема обновления зависимостей;
- только динамические свойства.
Все эти минусы процессоров ExecuteGroovyScript и ExecuteScript связаны с этапом разработки потокового конвейера. Если нужно больше отношений, следует установить какой-либо атрибут со статусом, а затем направить файл потока с процессором RouteOnAttribute. Однако, рекомендуется свести размер потока к минимуму и обновить JAR-файлы зависимостей, сделав недействительными процессоры, которые их используют, чтобы избежать неактуальности в логике обработки данных из-за кеширования устаревших версий JAR-файлов. Динамические свойства делают невозможным использование конфиденциальных свойств. Наконец, процессоры работают как обработчики и не реализуют возможности контроллеров.
С точки зрения DevOps-инженера, обслуживание скриптов тоже имеет определенные сложности. В частности, нет фреймворков для автоматизированного тестирования, отсутствуют стимулы использования неявных аргументов и нет возможности описать свойства. NiFi полон неявных значений, которые передаются по всему потоку в виде атрибутов, которые должны быть видимыми для их применения в пользовательских компонентах. Для этого придется долго изучать программный код и документацию на него, если она имеется, что в реальности бывает не всегда.
Поэтому пригодятся пользовательские компоненты Apache NiFi, которые можно создать с помощью предоставляемых API. Что это такое и чем они отличаются от скриптов, мы рассмотрим далее.
Скриптовые компоненты: за и против
Популярность скриптов вызвана возможностью их быстро создать и проверить логику работы. Однако, по мере развития ETL-конвейера скрипты усложняются и возникают вышеописанные трудности с их разработкой и обслуживанием, включая миграцию на новые версии. Однако, готовые компоненты NiFi не всегда реализуют нужные дата-инженеру возможности. Чтобы заполнить этот пробел между скриптами и готовыми компонентами NiFi объединяет эти 2 инструмента, предоставляя скриптовые компоненты:
- InvokeScriptedProcessor – вызывает скриптовый движок для процессора, определенного в скрипте. Скрипт определяет допустимый класс, который реализует интерфейс Processor, и устанавливает соответствующую переменную processor в экземпляре класса. Методы процессора, такие как onTrigger(), делегируются экземпляру процессора со скриптом. В диалоговое окно конфигурации добавлены любые отношения или дескрипторы свойств, определенные обработчиком скриптов. Пока этот скриптовый процессор находится в статусе «экспериментальный», поскольку его поведение в длительных вариантах использования еще не достаточно изучено.
- ScriptedLookupService – позволяет пользователю предоставить экземпляр LookupService со скриптом для обогащения записей из входящего файла потока. Из-за ошибки в Jython, которая еще не устранена, пока невозможно использовать Jython для написания Python-скрипта для этого сервиса.
- SimpleScriptedLookupService – позволяет пользователю предоставить экземпляр LookupService со скриптом для обогащения записей из входящего потокового файла. Скрипт вернет необязательное строковое значение, а не произвольный объект типа записи – в этом случае нужно реализовать метод getValueType(), даже если он будет проигнорирован. Аналогично предыдущему сервису, из-за ошибки в Jython, которая еще не устранена, невозможно использовать Jython для написания Python-скрипта для этого сервиса.
- ScriptedActionHandler — позволяет пользователю предоставить скрипт ActionHandler для пользовательского запуска правил в зависимости от предоставленных фактов. Скрипт должен установить переменную actionHandler для реализации ActionHandler.
- ScriptedReader – позволяет пользователю предоставить экземпляр RecordReaderFactory со сценарием для чтения/анализа/создания записей из входящего файла потока.
- ExecuteScript – выполняет скрипт с учетом файла потока и сеанса процесса. Скрипт отвечает за обработку входящего файла потока, например, передача в SUCCESS или удаление, а также любые файлы потока, созданные скриптом. Если обработка неполная или неправильная, сессия будет отменена. Пока этот скриптовый процессор находится в статусе «экспериментальный», поскольку его поведение в длительных вариантах использования еще не достаточно изучено.
- ScriptedFilterRecord – этот процессор позволяет фильтровать записи из FlowFiles с помощью предоставленного пользователем скрипта, который оценивает каждую запись и возвращает логическое значение. Записи с результатом true перенаправляются на совпадающие отношения в пакете, а другие записи будут отфильтрованы.
- ScriptedTransformRecord – предоставляет возможность оценить простой скрипт для каждой записи во входящем FlowFile. Скрипт может преобразовывать, фильтровать или разделять записи.
- ScriptedPartitionRecord – получает данные, ориентированные на записи, т.е. которые могут быть прочитаны сконфигурированным средством чтения записей и оценивает пользовательский скрипт для каждой записи во входящем файле потока. Затем каждая запись группируется с другими записями, использующими тот же раздел, и для каждой группы записей создается FlowFile. Две записи совместно используют один и тот же раздел, если оценка скрипта приводит к одинаковому возвращаемому значению для обеих и они будут считаться частью одного раздела.
- ScriptedValidateRecord – обеспечивает возможность проверки записей в FlowFiles с помощью предоставленного пользователем скрипта, который имеет запись в качестве входного аргумента и возвращает логическое значение. На основе этого результата процессор классифицирует записи как «валидные» или «невалидные» и направляет их в соответствующие отношения в пакетном режиме. Кроме того, исходный FlowFile будет направлен в исходное отношение (original relationship) или в failed при неудачной обработке.
- ScriptedReader – позволяет пользователю предоставить экземпляр RecordReaderFactory со скриптом для чтения/анализа/создания записей из входящего файла потока.
- ScriptedRecordSetWriter – позволяет пользователю предоставить экземпляр RecordSetWriterFactory со скриптом для записи записей в файл исходящего потока.
- ScriptedRecordSink – позволяет пользователю предоставить экземпляр RecordSinkService со скриптом для передачи записей в желаемую цель. Скрипт должен установить переменную ‘recordSink’ для реализации RecordSinkService.
- ScriptedRulesEngine – позволяет пользователю предоставить скрипт RulesEngineService для пользовательского запуска правил в зависимости от предоставленных фактов. Скрипт должен установить переменную ‘rulesEngine’ для реализации сервиса движка правил RulesEngineService.
- ScriptedReportingTask – предоставляет отчеты и информацию о состоянии для скрипта. Объекты ReportingContext, ComponentLog и VirtualMachineMetrics становятся доступными в виде переменных (context, log и vmMetrics соответственно) для дальнейшей обработки скриптом. Контекст делает доступной информацию про события, происхождение, бюллетени, службы контроллера, группы процессов, метрики виртуальной машины Java и т. д.
Все эти компоненты работают непосредственно, преобразуя предоставленную разработчиком пользовательскую реализацию в теле компонента. К примеру, можно написать свой код процессора и вставить его в свойство Script Body InvokeScriptedProcessor. Свойства, определенные в коде, будут видны в процессоре, но они не являются динамическими и имеют документацию в пользовательском интерфейсе NiFi. Также можно видеть отношения, определенные в коде и документацию по ним. Это очень удобно при разработке и поддержке потоковых ETL-конвейеров. Кроме того, скриптовые процессоры как промежуточные компоненты между скриптами и пользовательскими решениями позволяют автоматизировать юнит-тестирование написанного кода. С ними не нужно создавать новые конвейеры развертывания, можно вставить код в тело скрипта или использовать любой JAR-файл в рабочем процессе, как в случае со скриптами.
Также снижаются накладные расходы на разработку: не надо самостоятельно реализовать класс компонента, создавая maven-проект с настроенными зависимостями и развертыванием в NiFi. По сравнению с использованием скриптов, скриптовые компоненты позволяют настроить валидаторы для значений свойств, определить обязательность свойств и все, что связано с созданием пользовательских компонентов. Нет привязки только к процессорам, а применение лучших практик становится проще.
Обратной стороной этих достоинств является проблема с автоматическим обновлением зависимостей, которые все равно присутствуют. Кроме того, нет конфиденциальных значений, что потенциально может стать небезопасным. Пользовательские компоненты по-прежнему намного лучше подходят для модульного тестирования.
Узнайте больше про администрирование и эксплуатацию Apache NiFi для построения эффективных ETL-конвейеров потоковой аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
