В моем блоге уже было множество заметок на тему преобразования и передачи данных с использованием Apache NiFi, однако все наши предыдущие проекты разрабатывались на разных подверсиях NiFi 1.0. И вот, спустя несколько лет после официального выхода,  NiFi 2.0 наконец добрался и до нас, в связи с чем возникла потребность кратко описать самые существенные (с точки зрения моего личного опыта и специфики наших проектов) нововведения в обновленном и переработанном NiFi.

1. NiFi Registry устарел, появилась удобная интеграция с Git

Раньше для версионирования потоков и совместной работы нужно было поднимать NiFi Registry как отдельный сервис, в котором хранятся JSON'ы наших потоков данных. В NiFi 2.0 управление версиями встроено прямо «из коробки» через Registry Clients. Теперь появилась возможность напрямую подключить свои Process Groups к GitHub, GitLab или Bitbucket. Компонент GitFlowRegistryClient был и раньше, но его настройка была сложна, а взаимодействие с репозиторием все равно проксировалось через Registry. Теперь все будет делаться напрямую и с меньшим количеством лишних действий.

2. Упростилась настройка SSL контекста

Настройка TLS/SSL в NiFi 1.x с использованием Java Keystore (JKS) и Truststore требовала множество лишних действий и сложных конвертаций с использованием keytool. Теперь же появился Controller Service под названием PEMEncodedSSLContextProvider, представляющий собой новую версию интерфейса SSLContextProvider специально для прямой работы с PEM-файлами. Он позволяет сразу же использовать тот набор файлов, который вам с наибольшей вероятностью пришлет системный администратор (.pem, .crt, .key) без необходимости создавать .jks файлы.

3. Добавили параметризацию Controller Services

В версиях 1.x существовало жесткое архитектурное ограничение: в настройках процессора (например, в поле Database Connection Pool для ExecuteSQL) приходилось жестко выбирать конкретный сервис из выпадающего списка кликом мыши. В Apache NiFi 2.x разработчики реализовали принцип «Parameterize All The Things», добавив возможность передавать Controller Services через параметры. Теперь в свойство процессора можно напрямую вписать текстовый параметр вида #{database_service_id}. Это изменение избавляет от необходимости вручную менять множество процессоров при миграции потоков между контурами. Теперь достаточно в один клик изменить значение самого параметра на ID нужного контроллера в Parameter Context, и NiFi автоматически остановит зависимые компоненты, подменит реализацию сервиса и запустит пайплайн обратно.

4. Новый встроенный Stateless-движок

С NiFi 2.x экспериментальный движок Stateless превратился в полноценный встроенный режим работы групп процессов. Если обычно NiFi протоколирует и сохраняет в свой репозиторий практически любые действия, то теперь переключателем в UI можно перевести выбранный поток данных в режим обработки строго в оперативной памяти, полностью исключив ресурсоемкую запись файлов потока на жесткий диск. Кроме колоссального прироста производительности и ультра-низкой задержки на потоках из тысяч мелких сообщений, Stateless-режим гарантирует строгую транзакционность. При сбое NiFi автоматически откатит всю цепочку шагов назад. Это может быть особенно полезно для контейнерных сред и Kubernetes (поддержка которого также расширена в NiFi 2.0), позволяя создавать легковесные поды в кластере, которым не нужны дисковые хранилища. Раньше возможность stateless-запуска была неудобной, так как требовала использование отдельной CLI-утилиты.

5. Появилась поддержка процессоров на Python

И одно из самых интересных. Множество статей в этом блоге было написано на тему того, как писать на Groovy сложные скрипты для конвертации данных. Теперь же при желании можно использовать полноценный Python. Если старый процессор ExecuteScript опирался на движок Jython (Python внутри JVM), в котором недоступны многие современные библиотеки, то в NiFi 2.x интеграцию с Python переписали с нуля, сделав его языком первого класса (First-Class Support). Благодаря нативной поддержке полноценного CPython, разработчики теперь могут писать кастомные процессоры на чистом Python и бесшовно подключать к потокам данных. NiFi берет на себя всю рутину по управлению инфраструктурой - платформа сама изолирует процессы, автоматически создает виртуальные окружения (venv) для каждого процессора и скачивает нужные зависимости. То есть теперь необязательно добавлять процессор ExecuteScript, внутри которого пишется код, а можно написать собственный Python-скрипт и положить его в специальную папку на сервере NiFi (обычно python/extensions). После перезапуска NiFi скрипт превращается в настоящий процессор в общем списке. Groovy, таким образом, можно оставить только для более простых и быстрых трансформаций. Подробнее см. NiFi Python Developer’s Guide

Вот пример процессора, который принимает входящий FlowFile, парсит JSON и добавляет новое поле с помощью библиотеки numpy (NiFi сам скачает через pip, если указать в зависимостях):

import json
import numpy as np
from nifiapi.flowfiletransform import FlowFileTransform, FlowFileTransformResult

class CalculateStats(FlowFileTransform):
    class Java:
        implements = ['org.apache.nifi.python.processor.FlowFileTransform']

    class ProcessorDetails:
        version = '2.0.0'
        description = 'Пример процессора NiFi 2.0'
        # Указываем внешние библиотеки, NiFi установит их автоматически через pip
        dependencies = ['numpy==1.26.4'] 

    def __init__(self, **kwargs):
        pass

    def transform(self, context, flowFile):
        content = flowFile.read().decode('utf-8')
        
        try:
            data = json.loads(content)
        except Exception:
            return FlowFileTransformResult(relationship="failure")

        data['generated_id'] = int(np.random.randint(1, 9))
        
        output_content = json.dumps(data)

        # Возвращаем обновленный контент в связь success
        return FlowFileTransformResult(
            relationship="success",
            contents=output_content.encode('utf-8')
        )

Итак, сегодня мы рассмотрели наиболее значительные (на мой взгляд) нововведения NiFi 2.0 с точки зрения удобства и возможностей разработки. Кроме них существует еще множество изменений, среди которых более приятный визуал веб-интерфейса, различные оптимизации, добавление новых процессоров и удаление устаревших, и многое другое. Например, отдельно стоит выделить переход на более производительную Java 21 и отказ от поддержки устаревших версий систем (например, kafka 2.0). Таких внешне менее заметных изменений еще много, и для каждого конкретного случая найдется свой процессор, контроллер сервисов и стратегии построения потоков данных. Обо всем этом мы еще поговорим в будущем.