Поскольку Apache NiFi позволяет не только использовать готовые процессоры, но и разработать свой собственный, дата-инженеру полезно знать лучшие практики проектирования таких обработчиков Flow File.
Принцип единой ответственности при проектировании процессора Apache NiFi
В Apache NiFi есть более 300 готовых процессоров, которые выполняют определенные действия с потоковыми файлами в рамках конвейера обработки данных. Также можно написать свой собственный обработчик самостоятельно, что мы рассматривали здесь и здесь. При самостоятельном проектировании процессора необходимо учитывать следующие аспекты:
- Шаблон проектирования, например, прием, вывод ил обогащение данных;
- Принцип единой ответственности, согласно которому каждый процессор выполняет атомарную единицу работы, т.е. только одну конкретную точечную операцию, например, прослушивание HTTP-запросов к конечной точке, или преобразование данных, а не все сразу.
- Простота использования, включая понятное именование самого процессора и его свойств. Сюда же следует отнести вопрос о настройке значений этих свойств, чтобы не заставлять пользователей тратить много времени на настройку параметров. Тем не менее, рекомендуется избегать жестко запрограммированных значений, по возможности используя параметры или атрибуты потокового файла, чтобы использовать их повторно как шаблоны. Повторное использование также становится проще при аннотировании поведения процессора, что мы рассмотрим далее.
- Сканирование содержимого вместо буферизации данных всего потокового файла, чтобы избежать ошибки нехватки памяти.
Сперва хочется подчеркнуть необходимость декомпозиции всего конвейера обработки данных до набора элементарных операций, чтобы каждый процессор делал что-то одно и делал это хорошо. Вместо объединения нескольких функций в один процессор, например, прием, преобразование данных и их отправка во внешний сервис, следует реализовать 3 отдельных процессора. Это упростит работу как над отдельно взятым обработчиком, так и над всем конвейером в целом. В частности, если многофункциональный сложный процессор в рассмотренном примере не может связаться с удаленным сервисом, чтобы отправить туда принятые и преобразованные данные, он направит данные в отношение failure и будет нести ответственность за повторную трансляцию данных, которая может привести к зависанию системы.
Аналогичный пример касается промежуточных вычислений с данными, в частности, преобразование формата и структуры, которые могут пригодиться нескольким разным обработчикам. Таким образом, принцип единой ответственности при проектировании пользовательского процессора позволяет сделать эти обработчики пригодными для повторного использования, а также упростить их отладку и повысить пропускную способность всей системы потоковой передачи на базе Apache NiFi.
Правила именования и аннотирования
Стандартизованные названия и употребление общеизвестных терминов улучшают пользовательский опыт. Поэтому, если предоставить пользователям единообразный внешний вид именования процессоров, согласно стандартных соглашений об именах, вопросов о том, как применять на практике тот или ной обработчик, станет меньше.
В частности, процессоры, которые извлекают данные из удаленной системы, называются Get<Service> или Get<Protocol>, в зависимости от того, запрашивают ли они данные из произвольных источников по известному протоколу, такому как HTTP или FTP (GetHTTP или GetFTP) или извлекают данные из внешнего сервиса, например, GetKafka. Процессоры, передающие данные в удаленную систему, обычно называются Put<Service> или Put<Protocol>.
Имена отношений пишутся строчными буквами и используют пробелы для разграничения слов. А в названиях свойств рекомендуется использовать стиль CamelCase, где значимые слова пишутся с заглавной буквы. По умолчанию атрибуты Flow File отображаются в алфавитном порядке. Если необходимо отображать пользовательские атрибуты вместе, можно использовать специальный префикс, например, cust_attr_<<name>>.
При создании процессора разработчик может дать фреймворку подсказки о том, как использовать этот обработчик наиболее эффективно, применив аннотации к классу процессора. Эти аннотации существуют в трех подпакетах org.apache.nifi.annotation:
- подпакет documentation используется для предоставления документации пользователю;
- подпакет lifecycle указывают фреймворку, какие методы следует вызывать в процессоре, чтобы реагировать на соответствующие события жизненного цикла;
- пакет behavior помогают фреймворку понять, как взаимодействовать с процессором с точки зрения планирования и общего поведения.
Следующие аннотации из пакета org.apache.nifi.annotation.behavior можно использовать для изменения того, как Apache NiFi будет обрабатывать процессор:
- EventDriven — указывает платформе, что процессор может быть запланирован с использованием стратегии планирования, управляемой событиями. Пока эта стратегия все еще является экспериментальной;
- SideEffectFree – указывает на отсутствие в процессоре побочных эффектов, внешних по отношению к NiFi.Платформа может многократно вызывать процессор с одними и теми же входными данными, получая один и тот же результат, что соответствует идемпотентному поведению. Это полезно при выполнении таких действий, как передача ProcessSession с одного процессора на другой, чтобы в случае возникновения проблемы можно было откатить многие действия процессоров и выполнить их снова.
- SupportsBatching – указывает на то, что платформа может объединять несколько фиксаций ProcessSession в одну. Если эта аннотация присутствует, пользователь сможет выбрать, предпочитает ли он высокую пропускную способность или более низкую задержку процессора при его планировании. Эту аннотацию рекомендуется применять к большинству процессоров, но с безопасным сохранением данных в репозиториях NiFi (Content, FlowFile и Provenance). Это не подходит для тех процессоров, которые получают данные из внешнего источника, фиксируют сеанс, а затем удаляют удаленные данные или подтверждают транзакцию с удаленным ресурсом.
- TriggerSerially — при наличии этой аннотации платформа не позволит пользователю запланировать более одного параллельного потока для одновременного выполнения метода onTrigger(). Вместо этого количество потоков всегда будет установлено на Однако, это не отменяет необходимость потокобезопасности процессора, поскольку выполняемый методом onTrigger() поток может меняться между вызовами.
- PrimaryNodeOnly – указывает на режим выполнения процессора на основном узле. Хотя запуск на всех узлах кластера NiFi обеспечивает лучший параллелизм, некоторые процессоры вызывают ошибки при запуске на нескольких узлах, например, чтение файлов из удаленных файловых систем на всех узлах кластера приведет к ненужному дублированию данных.Такие обработчики должны выполняться только на одном узле, что и позволяет указать аннотация @PrimaryNodeOnly.
- TriggerWhenAnyDestinationAvailable – указывает на то, что NiFi будет планировать запуск процессора, если какая-либо из его исходящих очередей заполнена, т.е. какое-то отношение стало доступным. Отношение считается доступным, если ни одно из соединений, использующих это отношение, не заполнено. Например, процессор DistributeLoad использует эту аннотацию. Если используется стратегия планирования «циклический перебор», процессор не будет работать, когда какая-то исходящая очередь заполнена. Изменить это поведение позволяет аннотация @TriggerWhenAnyDestinationAvailable.
- TriggerWhenEmpty – указывает на игнорирование размера входных очередей и запуск процессора независимо от того, есть ли какие-либо данные во входной очереди. Это полезно, если процессор должен периодически запускаться для тайм-аута сетевого подключения. Без этой аннотации поведение NiFi по умолчанию заключается в запуске процессора только в том случае, если в его входной очереди есть хотя бы один FlowFile или если у процессора нет входных очередей.
- InputRequirement – изменяет поведение по умолчанию, когда процессор может быть запланирован для запуска без входящего соединения. Аннотации @InputRequirement можно присвоить значение INPUT_REQUIRED, INPUT_ALLOWED, или INPUT_FORBIDDEN, указывая NiFi, когда процессор должен быть стать невалидным. Например, если процессор помечен @InputRequirement(Requirement.INPUT_FORBIDDEN), пользователь даже не сможет создать соединение с этим процессором в качестве пункта назначения.
Буферизация данных
Напомним, NiFi хранит в одном физическом файле множество данных о содержимом разных FlowFile. Содержимое каждого FlowFile хранится на диске и считывается в память JVM только при необходимости. Это позволяет NiFi обрабатывать объекты различного масштаба, не требуя, чтобы процессоры продюсеров и потребителей данных хранили их целиком в памяти. Хотя разделение, агрегирование и преобразование очень больших объектов можно выполнять без вреда для памяти, иногда это может привести к некоторым проблемам. В частности, из-за универсального характера NiFi, позволяющего работать с данными любых форматов, разработчик Data Flow может попытаться буферизовать весь контент FlowFile в памяти. Этого следует избегать при неизвестном формате обрабатываемых данных. Например, процессору, ответственному за выполнение XPath для XML-документа, потребуется загрузить весь контент данных в память. Но XML-документ не должен быть очень большим.
Вместо буферизации данных в памяти рекомендуется оценивать их по мере потоковой передачи из репозитория содержимого, т. е. сканировать контент из того, что InputStream предоставляется обратному вызову с помощью метода ProcessSession.read(). Чтобы не читать из репозитория контента каждый байт, следует использовать BufferedInputStream или другим образом буферизовать небольшой объем данных.
В заключение перечислим несколько рекомендаций по эксплуатации типовых и пользовательских процессоров NiFi при проектировании конвейера обработки данных:
- вместо сохранения потоков на корневом уровне лучше использовать группы процессов, чтобы структурировать конвейер обработки данных;
- метки (ярлыки), включая цветовую маркировку процессоров и стрелок для отображения соединений, помогут лучше описать свой поток и сэкономить временя при отладке;
- указание имени для всех групп процессов, процессоров, очередей, портов ввода и вывода поможет при отладке и улучшит понимание логики потока данных.
Читайте в нашей новой статье про изолированные процессоры в кластере NiFi: что это такое, зачем их использовать и как это сделать.
Освойте все возможности Apache NiFi для построения эффективных ETL-конвейеров потоковой аналитики больших данных на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
