Обзор Apache NiFi: управление потоками данных

Обзор Apache NiFi: управление потоками данных начинается с одной задачи — провести объект через цепочку обработчиков, сохранив его содержимое, атрибуты и историю. NiFi строится на FlowFile, processors, connections и репозиториях, которые хранят состояние очередей. Ниже — разбор этих понятий, учебная схема обработки файла с записями и проверка типовых ошибок.

Задача NiFi

Apache NiFi решает задачу управляемого перемещения данных между системами. Документация проекта на nifi.apache.org описывает продукт как среду, где каждый объект, движущийся через систему, представлен FlowFile, а обработку выполняют processors. Читатель приходит к NiFi, когда нужно соединить источник файлов, базу данных, HTTP-приёмник или очередь сообщений и получить наблюдаемый маршрут с понятным поведением при сбоях.

Задача формулируется как поток работ. Инженер рисует граф: узел забирает данные, следующий узел преобразует, третий пишет результат. Между узлами стоят очереди, которые сглаживают разную скорость обработки. Документация nifi.apache.org прямо говорит, что connections служат связями между processors и работают как очереди, позволяя процессам взаимодействовать с разной скоростью. Это ядро модели: граф описывает маршрут, а очереди отделяют быстрый источник от медленного приёмника.

Вторая часть задачи — воспроизводимость и контроль. NiFi хранит состояние в репозиториях, поэтому перезапуск узла не обязан терять данные, уже принятые в поток. Документация проекта отмечает, что processors выполняют работу над нулём или более FlowFile в единице работы и либо фиксируют результат, либо откатывают его. Такая семантика важна для сценариев, где частичная запись хуже повторной попытки.

Третья часть — наблюдаемость. Когда поток обрабатывает тысячи объектов, инженеру нужен ответ на вопрос, где именно задержался конкретный файл. NiFi ведёт provenance-данные: документация nifi.apache.org указывает, что система автоматически записывает, индексирует и предоставляет данные о происхождении объектов по мере их движения, включая ветвления и преобразования. Это превращает граф из чёрного ящика в проверяемый маршрут.

Практический выбор в пользу NiFi возникает там, где нужен визуально описанный поток с маршрутизацией ошибок. Продукт подходит для сбора файлов, разбора записей, разделения валидных и проблемных данных, отправки результата во внешнюю систему. Он не заменяет собой брокер сообщений и не снимает необходимость проектировать схему данных. Задача NiFi — дать управляемый конвейер с состоянием, очередями и историей.

В NiFi полезно различать три уровня. Первый — объект (FlowFile с содержимым и атрибутами). Второй — операция (processor, который читает, преобразует или пишет). Третий — маршрут (connections, очереди, backpressure, provenance). Понимание этих уровней снимает большую часть путаницы при первом знакомстве с интерфейсом.

Content и attributes FlowFile

FlowFile — единица данных в NiFi. Документация nifi.apache.org определяет его так: FlowFile представляет каждый объект, движущийся через систему, и для каждого объекта NiFi хранит карту пар ключ-значение строковых атрибутов и связанное содержимое размером от нуля байт. Это разделение принципиально: метаданные живут отдельно от полезной нагрузки.

Содержимое (content) — это байты: CSV, JSON, изображение, архив. Атрибуты (attributes) — строковые метаданные: имя файла, источник, путь, размер, пользовательские поля. Атрибуты дешевы для чтения и маршрутизации, содержимое дороже. Инженер проектирует поток так, чтобы решения о маршруте принимались по атрибутам там, где это возможно, а содержимое читалось только на этапе реального преобразования.

Учебный пример из принятого артефакта задаёт FlowFile с содержимым orders.csv, в котором две CSV-записи: id1,amount100 и id2,amount−5. Атрибуты FlowFile: filename=orders.csv, source=training. Атрибут source предложен для учебной схемы; его создание нужно предусмотреть явно. Такое различие важно: инженер сам решает, какие атрибуты добавить на входе, и сам отвечает за их согласованность.

Атрибуты переживают преобразования и участвуют в маршрутизации. Если processor добавляет атрибут с результатом проверки, следующий узел может направить FlowFile по этому признаку. Содержимое при этом может оставаться неизменным. Документация проекта подчёркивает, что NiFi отслеживает карту атрибутов для каждого объекта, поэтому маршрутизация по метаданным становится штатным приёмом.

Практический шаг для читателя: перед построением потока выписать список атрибутов, которые нужны на каждом этапе. Для учебной схемы это filename и source на входе, признак результата проверки на выходе. Если атрибут нужен позже, его добавляют раньше. Если атрибут не нужен, его не добавляют, чтобы не увеличивать объём метаданных.

Отдельно стоит помнить про размер содержимого. FlowFile может нести ноль байт, и это допустимое состояние. Нулевое содержимое встречается при передаче сигналов или маркеров, когда важны только атрибуты. Инженер, который ожидает непустой файл, обязан проверить это условие явно, иначе пустой объект пройдёт по маршруту и вызовет ошибку на этапе разбора.

Processors и connections

Processors выполняют работу. Документация nifi.apache.org описывает их как компоненты, которые оперируют нулём или более FlowFile в единице работы и либо фиксируют работу, либо откатывают её. Каждый processor имеет свойства, relationships и побочные эффекты. Свойства настраивают поведение, relationships определяют, куда уйдёт FlowFile после обработки.

Connections соединяют processors и работают как очереди. Документация проекта указывает, что connections обеспечивают фактическую связь между processors и позволяют процессам взаимодействовать с разной скоростью. Очередь хранит FlowFile до тех пор, пока принимающий processor не заберёт их. Это разделение во времени сглаживает пики и защищает медленный узел от перегрузки.

Учебная схема строится из трёх шагов. Первый — GetFile: документация компонента на nifi.apache.org говорит, что он читает файлы из каталога и передаёт их в содержимое новых FlowFile. Второй — ValidateRecord: документация компонента сообщает, что он проверяет записи входящего FlowFile по заданной схеме, направляя соответствующие записи в relationship valid, а несоответствующие — в invalid. Третий — маршрутизация результата: валидные записи идут дальше, проблемные — в карантин.

Таблица ниже сопоставляет элементы потока и их роль в учебной схеме.

Важная деталь про relationships. Документация ValidateRecord на nifi.apache.org перечисляет relationships valid, invalid и failure. Записи, соответствующие схеме, уходят в valid; несоответствующие — в invalid; при невозможности прочитать, проверить или записать данные исходный FlowFile уходит в failure. В учебной схеме мы используем собственные обозначения результатов: valid содержит запись id1, quarantine — запись id2 с причиной negative_amount. Слово quarantine обозначает логический результат учебной схемы. Имя relationship выбирают из документации конкретного processor.

Ещё одна деталь — разделение записей. Документация ValidateRecord указывает, что один входящий FlowFile может быть разделён на два отдельных FlowFile, если часть записей соответствует схеме, а часть нет. Это объясняет, почему баланс на уровне записей и баланс на уровне FlowFile — разные вещи. В учебном примере вход содержит две записи, и баланс записей равен 2 = 1 + 1. Количество выходных FlowFile эта арифметика не определяет.

Практический порядок сборки потока: сначала выбрать источник и приёмник, затем определить схему данных, затем настроить relationships и очереди, и только после этого добавлять преобразования. Такой порядок снижает риск переделки графа. Каждый новый processor добавляется с явным ответом на вопрос, что произойдёт с FlowFile при успехе и при ошибке.

Backpressure

Backpressure защищает систему от переполнения. Документация nifi.apache.org сообщает, что NiFi поддерживает буферизацию всех данных в очередях, а также возможность создавать обратное давление при достижении заданных лимитов или удалять данные по достижении заданного возраста. Это два разных механизма: порог останавливает приём, срок хранения удаляет устаревшее.

В учебном примере используется порог backpressure по числу объектов в очереди. В учебном примере выбран порог 100, и он относится к FlowFile-объектам в очереди. Две CSV-строки находятся внутри одного такого объекта. Значение 100 выбрано для примера и не является значением по умолчанию. Читатель настраивает порог под свой поток, исходя из объёма объектов и скорости обработки.

Срок хранения (expiration) работает независимо. Если данные в очереди стареют быстрее, чем обрабатываются, они могут быть удалены по возрасту. Это защищает диск от бесконечного накопления, но означает потерю объектов, которые не успели пройти дальше. Инженер выбирает срок хранения осознанно, соотнося его с требованиями к полноте данных.

Таблица ниже показывает, как два механизма влияют на поведение потока.

Практический шаг: настроить порог backpressure на connection перед медленным processor. Если источник быстрее приёмника, очередь растёт, и порог останавливает источник. Это переносит давление назад по потоку и даёт время разобрать накопленное. Без порога очередь растёт до исчерпания диска.

Отдельная ошибка — путать порог очереди с ограничением на размер файла. Этот порог считает FlowFile-объекты. Число строк и объём содержимого учитывают отдельно. Файл с миллионом записей — это один FlowFile, и он занимает одну позицию в очереди. Если инженер ожидает, что порог 100 ограничит число записей, ожидание не оправдается.

Ещё один практический момент — мониторинг очередей. Растущая очередь указывает на дисбаланс скоростей. Инженер смотрит на размер очереди, число объектов и возраст самого старого объекта. Эти три показателя вместе дают картину: очередь растёт, объекты стареют, приёмник не успевает. Решение — ускорить приёмник, добавить параллелизм или изменить маршрут.

Provenance

Provenance — история обработки объектов. Документация nifi.apache.org указывает, что NiFi автоматически записывает, индексирует и предоставляет данные о происхождении объектов по мере их движения через систему, включая ветвления, слияния и преобразования. Это означает, что для каждого FlowFile можно посмотреть, какие processors его обрабатывали и когда.

Provenance отвечает на вопросы расследования. Если файл не дошёл до приёмника, инженер открывает историю и видит, на каком шаге объект остановился. Если файл обработан дважды, история показывает два события. Если запись ушла в карантин, история показывает, какой processor принял это решение.

Важное ограничение: provenance прослеживает обработку, но не доказывает бизнес-правильность. Он покажет, что processor выполнился и направил объект в relationship. Он не покажет, что правило amount>=0 было применено корректно, если само правило реализовано с ошибкой. Проверка бизнес-логики остаётся задачей инженера и его тестов.

В учебной схеме provenance полезен для проверки маршрута. В истории нужно проверить захват файла GetFile, результат проверки схемы ValidateRecord и события отдельного шага бизнес-проверки amount>=0. Маршруты valid/quarantine из бизнес-проверки сопоставляют с исходным файлом. История показывает обработку каждого шага; правильность правила проверяют отдельно.

Provenance также связан с replay. Документация проекта упоминает возможность повторного воспроизведения данных. Доступность replay зависит от настроек хранения: если события и содержимое сохраняются, объект можно воспроизвести; если срок хранения истёк, воспроизведение недоступно. Инженер учитывает это при проектировании.

Практический шаг: перед запуском потока определить, какие события нужно хранить и как долго. Для учебной схемы достаточно хранить события обработки файла orders.csv. Для продуктивной системы срок хранения выбирается по требованиям аудита и расследований. Provenance — инструмент наблюдаемости, и его настройки влияют на объём хранимых данных.

Учебный поток и исключения

Учебный поток строится на бумаге и описывает логическую схему. Вход: один FlowFile с содержимым orders.csv, две CSV-записи (id1,amount100) и (id2,amount−5). Атрибуты: filename=orders.csv, source=training. Бизнес-правило: amount>=0. Это правило задано нами и требует отдельной реализации. ValidateRecord проверяет записи по схеме, а бизнес-проверку диапазона выполняет отдельный подходящий шаг.

Порядок шагов выглядит так. GetFile читает файл из каталога и создаёт FlowFile. ValidateRecord проверяет записи по схеме и разделяет их: соответствующие уходят в valid, несоответствующие — в invalid. Бизнес-проверка amount>=0 выполняется отдельным шагом и направляет запись id2 в карантин с причиной negative_amount. Валидная запись id1 идёт дальше по потоку.

Логические результаты учебной схемы: valid содержит запись id1, quarantine — запись id2 с причиной negative_amount. Это обозначения результатов учебной схемы; имена relationships проверяют у конкретного processor. Баланс на уровне записей: вход 2 = valid 1 + quarantine 1. Количество выходных FlowFile эта арифметика не определяет.

Учебный текст входного файла можно сохранить так:

id,amount
1,100
2,-5

Проверка исключений включает несколько сценариев. Первый — пустой входящий FlowFile. Документация ValidateRecord на nifi.apache.org сообщает, что если входящий FlowFile не содержит записей, выходной FlowFile не создаётся ни для одного relationship, а входящий FlowFile удаляется. Инженер, ожидающий выходной объект, обязан учесть это поведение.

Второй сценарий — ошибка чтения или записи. Документация ValidateRecord указывает, что при невозможности прочитать, проверить или записать данные исходный FlowFile направляется в relationship failure. Это отдельный путь для технических ошибок, отличный от маршрута несоответствующих схеме записей.

Третий сценарий — повторное поступление файла. Документация GetFile на nifi.apache.org сообщает, что при значении KeepSourceFile=true файл не удаляется после копирования в Content Repository, что приводит к его постоянному повторному захвату и полезно для тестирования. Повторное поступление может дать дубликаты. Устранение дубликатов в учебной схеме не заявляется.

Четвёртый сценарий — права на каталог. Документация GetFile указывает, что при отказе от сохранения исходного файла NiFi нужны права на запись в каталоге, где лежит файл, иначе файл игнорируется. Инженер проверяет права до запуска потока, иначе источник молча не отдаёт данные.

Пятый сценарий — проверка результата. После обработки инженер сверяет баланс записей и смотрит provenance. Если баланс не сходится, ошибка в маршрутизации или в шаге проверки. Если баланс сходится, но данные неверны, ошибка в бизнес-правиле. Эти две проверки разделены, и это упрощает диагностику.

Ограничения

Учебная схема — спецификация на бумаге. Редакция NiFi не запускала. Имена processors и relationships зависят от версии, и их сверяют по текущим component docs. Ни flow, ни порт не открываются без авторизации и TLS. Пример не заявляет установленную скорость.

Ограничение первое — бизнес-правила. ValidateRecord проверяет записи по схеме. Проверка произвольного диапазона значений, такого как amount>=0, требует отдельного шага. Инженер, который ожидает от ValidateRecord проверки диапазона, получит неверный результат: запись с отрицательным значением может соответствовать схеме и уйти в valid.

Ограничение второе — баланс записей и объектов. Баланс на уровне записей не определяет количество выходных FlowFile. Один входящий FlowFile может быть разделён на несколько выходных, и наоборот. Инженер отдельно планирует число объектов и объём их содержимого.

Ограничение третье — дубликаты. Повторное поступление файла при KeepSourceFile=true может дать дубликаты. Дедупликация и гарантия однократной доставки не заявляются. Если поток пишет во внешнюю систему, повторная запись возможна, и её обработка — отдельная задача.

Ограничение четвёртое — provenance. Provenance прослеживает события обработки. Он не доказывает бизнес-правильность и не гарантирует бессрочное хранение каждого исторического байта содержимого. Доступность replay зависит от настроек хранения.

Ограничение пятое — окружение. Пример не описывает установку и не заявляет производительность. Числа в учебной схеме — условные значения: порог 100 выбран для примера, две записи заданы для наглядности. Инженер подставляет свои значения и проверяет их на своём окружении.

Таблица ниже сводит ограничения и практические выводы.

Частые вопросы

Что такое Apache NiFi простыми словами?
Apache NiFi — система управления потоками данных, где каждый объект представлен FlowFile с содержимым и атрибутами, обработку выполняют processors, а связи между ними работают как очереди. Документация проекта на nifi.apache.org описывает эти элементы как основу модели. Инженер собирает граф из processors и connections и получает наблюдаемый маршрут обработки.

Чем FlowFile отличается от обычного файла?
FlowFile содержит содержимое и карту строковых атрибутов. Документация nifi.apache.org указывает, что NiFi хранит для каждого объекта карту пар ключ-значение и связанное содержимое размером от нуля байт. Содержимое может быть пустым, а атрибуты при этом несут метаданные для маршрутизации.

Как работает backpressure в NiFi?
Backpressure останавливает приём в очередь при достижении заданного порога объектов. Документация проекта сообщает, что NiFi поддерживает буферизацию данных и создание обратного давления при достижении лимитов. В учебном примере выбран порог по числу FlowFile-объектов; строки внутри файла учитывают отдельно.

Что показывает provenance?
Provenance показывает историю обработки объектов: какие processors их обрабатывали и когда. Документация nifi.apache.org указывает, что система автоматически записывает и индексирует данные о происхождении объектов при их движении. Provenance помогает расследовать маршрут, а проверку бизнес-правил инженер выполняет отдельно.

Можно ли проверить диапазон значений через ValidateRecord?
ValidateRecord проверяет записи по схеме и направляет соответствующие в valid, несоответствующие — в invalid. Документация компонента на nifi.apache.org описывает именно проверку по схеме. Бизнес-проверку диапазона, например amount>=0, реализуют отдельным подходящим шагом.