Здравейте на всички!

Задачата се състои в следното — има flow, представен на изображението по-горе, който трябва да бъде разпределен на N сървъра с Apache NiFi. Flow е тестов — става дума за генериране на файл и изпращането му в друг инстанс NiFi. Передачата на данни се извършва с помощта на протокола NiFi Site to Site.

NiFi Site to Site (S2S) — безопасен, лесно конфигурируем метод за предаване на данни между инстансите на NiFi. Как работи S2S, вижте в документацията и важно е да не забравите да конфигурирате NiFi инстанса, за да разрешите S2S, вижте тук.

В случаите, когато става въпрос за предаване на данни с S2S — един инстанс се нарича клиентски, а вторият — сървърен. Клиентският изпраща данни, а сървърният ги приема. Има два начина за конфигуриране на предаването на данни между тях:

1. Push. От клиентския инстанс данните се изпращат с помощта на Remote Process Group (RPG). На сървърния инстанс данните се приемат с помощта на Input Port
2. Pull. Сървърът приема данни с помощта на RPG, клиентът изпраща с помощта на Output port.

Flow за разпределение съхраняваме в Apache Registry.

Apache NiFi Registry — подпрограма на Apache NiFi, представляваща инструмент за съхранение на flow и управление на версиите. Нещо като GIT. Информация за инсталацията, конфигурацията и работата с registry можете да намерите в официалната документация. Flow за съхранение се обединява в process group и в такъв вид се съхранява в registry. По-нататък в статията ще се върнем към това.

В началото, когато N е малко число, flow се доставя и актуализира ръчно за приемливо време.

Но с нарастването на N, проблемите стават повече:

1. за актуализация на flow отнема повече време. Трябва да влезете на всички сървъри
2. възникват грешки при актуализиране на шаблони. Тук е обновено, а тук е забравено
3. грешки на човека при изпълнение на голямо количество еднотипни операции

Всичко това ни подтиква да автоматизираме процеса. Опитах следните подходи за решаване на тази задача:

1. Да използвам MiNiFi вместо NiFi
2. NiFi CLI
3. NiPyAPI

Използване на MiNiFi

Apache MiNiFy — под проект Apache NiFi. MiNiFy — компактен агент, който използва същите процесори като NiFi, позволяващ създаването на същите потоци, както в NiFi. Леката конструкция на агента се постига и благодарение на отсъствието на графичен интерфейс за конфигуриране на потока. Липсата на графичен интерфейс в MiNiFy означава, че е необходимо да се реши проблемът с доставката на потока до minifi. Тъй като MiNiFy активно се използва в IoT, компонентите са много и процесът на доставка на потока до крайни инстанции на minifi трябва да бъде автоматизиран. Позната задача, нали?

Решението на такъв проблем ще предостави още един под проект — MiNiFi C2 Server. Този продукт е предназначен да бъде централната точка в архитектурата за разпространение на конфигурации. Как да конфигурирате средата е описано в тази статия на Хабра и информацията е достатъчна за решаване на поставената задача. MiNiFi свързан с C2 сървър автоматично обновява конфигурацията си. Единственият недостатък на този подход е необходимостта от създаване на шаблони на C2 сервера; обикновеното комитване в регистъра не е достатъчно.

Вариантът, описан в статията по-горе, е работещ и не сложен за реализация, но не бива да забравяме следното:

1. В minifi няма всички процесори от nifi
2. Версиите на процесорите в Minifi изостават от версиите на процесорите в NiFi.

Към момента на написването на публикацията последната версия на NiFi е 1.9.2. Версията на процесорите на последната версия на MiNiFi е 1.7.0. Процесорите могат да се добавят в MiNiFi, но поради несъответствия във версиите между процесорите NiFi и MiNiFi това може да не проработи.

NiFi CLI

Съдейки по описанието инструмента на официалния сайт, това е инструмент за автоматизация на взаимодействието между NiFI и NiFi Registry в областта на доставката на потоци или управление на процеси. За да започнете работа, е необходимо да изтеглите този инструмент оттук.

Стартираме утилитата

.\/bin\/cli.sh
           _     ___  _
 Apache   (_)  .' ..](_)   ,
 _ .--.   __  _| |_  __    )
[ `.-. | [  |'-| |-'[  |  \/  
|  | | |  | |  | |   | | '    '
[___||__][___][___] [___]',  ,'
                           `'
          CLI v1.9.2

Type 'help' to see a list of available commands, use tab to auto-complete.

За да подгрузим необходимия поток от регистъра, трябва да знаем идентификаторите на коша (bucket identifier) и самия поток (flow identifier). Тези данни могат да бъдат получени или чрез cli, или в уеб интерфейса на NiFi регистъра. В уеб интерфейса изглежда така:

С помощта на CLI се прави така:

#> registry list-buckets -u http://nifi-registry:18080

#   Name             Id                                     Description
-   --------------   ------------------------------------   -----------
1   test_bucket   709d387a-9ce9-4535-8546-3621efe38e96   (empty)

#> registry list-flows -b 709d387a-9ce9-4535-8546-3621efe38e96 -u http://nifi-registry:18080

#   Name           Id                                     Description
-   ------------   ------------------------------------   -----------
1   test_flow   d27af00a-5b47-4910-89cd-9c664cd91e85

Стартираме импорта на група от процеси от регистъра:

#> nifi pg-import -b 709d387a-9ce9-4535-8546-3621efe38e96 -f d27af00a-5b47-4910-89cd-9c664cd91e85 -fv 1 -u http://nifi:8080

7f522a13-016e-1000-e504-d5b15587f2f3

Важно е — като хост, на който налагаме група от процеси, може да бъде посочен всеки инстанс на nifi.

Групата за процеси е добавена с деактивирани процесори, те трябва да бъдат стартирани.

#> nifi pg-start -pgid 7f522a13-016e-1000-e504-d5b15587f2f3 -u http://nifi:8080

Отлично, процесорите стартираха. Въпреки това, според условията на задачата, инстанциите NiFi трябва да изпращат данни на други инстанции. Предположим, че за предаване на данни на сървъра е избран методът Push. За да организираме предаването на данни, трябва да активираме предаването на данни на добавената Група за отдалечен процес (RPG), която вече е включена в нашия поток.

В документацията в CLI и други източници не намерих начин да активирам предаването на данни. Ако знаете как да го направите, моля, пишете в коментарите.

Отивайки в bash и готови да стигнем до края, ще намерим решение! Можем да използваме NiFi API, за да решим този проблем. Ще използваме следния метод, като вземем ID от примера по-горе (в нашия случай това е 7f522a13-016e-1000-e504-d5b15587f2f3). Описание на методите на NiFi API тук.

В тялото трябва да предадем JSON в следния формат:

{
    "revision": {
	    "clientId": "стоност",
	    "version": 0,
	    "lastModifier": "стойност"
	},
    "state": "стойност",
    "disconnectedNodeAcknowledged": true
}

Параметрите, които трябва да попълним, за да "заработи":
state — статус на предаването на данни. Достъпен е TRANSMITTING за включване на предаването на данни, STOPPED за спиране
version — версия на процесора

версият по подразбиране ще бъде 0 при създаването, но тези параметри могат да бъдат получени, използвайки метода

За любителите на bash скриптове, този метод може да изглежда приложим, но за мен е трудно — bash скриптовете не са моят любим жанр. Следващият метод е по-интересен и удобен според мен.

NiPyAPI

NiPyAPI — библиотека за Python, за взаимодействие с инстанции NiFi. Страницата с документация съдържа необходимата информация за работа с библиотеката. Бързото стартиране е описано в проект на github.

Нашият скрипт за разгръщане на конфигурацията е програма на Python. Преминаваме към кодирането.
Настройваме конфигурирации за по-нататъшна работа. Нуждаем се от следните параметри:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #път до nifi-api инстанса, на който разгръщаме група от процеси
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #път до nifi-registry-api регистъра
nipyapi.config.registry_name = 'MyBeutifulRegistry' #име на регистъра, как ще се нарича в инстанса niFi
nipyapi.config.bucket_name = 'BucketName' #име на кофата, от която извличаме потока
nipyapi.config.flow_name = 'FlowName' #име на потока, който извличаме

След това ще включа имена на методи от тази библиотека, които са описани тук.

Свързваме регистъра с инстанса niFi с помощта на

nipyapi.versioning.create_registry_client

На този етап можете също така да добавите проверка дали registry вече е добавен към инстанцията, за което можете да използвате метода

nipyapi.versioning.list_registry_clients

Намираме bucket за по-нататъшно търсене на flow в кошницата

nipyapi.versioning.get_registry_bucket

По намерения bucket търсим flow

nipyapi.versioning.get_flow_in_bucket

Следващо важно е да разберем дали този process group вече е добавен. Process group се разполага по координати и може да се получи ситуация, в която втори компонент ще се наложи върху първия. Проверявах, може да се случи 🙂 За да получим всички добавени process group, използваме метода

nipyapi.canvas.list_all_process_groups

и след това можем да потърсим, например, по име.

Няма да описвам процеса на обновяване на шаблона, ще кажа само, че ако в новата версия на шаблона се добавят процесори, няма проблеми с наличието на съобщения в опашките. А ако се изтриват процесори, могат да възникнат проблеми (nifi не позволява изтриването на процесор, ако пред него се е натрупала опашка от съобщения). Ако ви интересува как реших този проблем — пишете ми, моля, ще обсъдим този момент. Контакти в края на статията. Да преминем към стъпката по добавяне на process group.

При отстраняване на грешки в скрипта се сблъсках с особеност, че не винаги се взема последната версия на flow, затова препоръчвам първо да уточните тази версия:

nipyapi.versioning.get_latest_flow_ver

Деплойваме process group:

nipyapi.versioning.deploy_flow_version

Стартираме процесорите:

nipyapi.canvas.schedule_process_group

В блока за CLI беше написано, че в remote process group автоматично не се включва предаването на данни? При реализирането на скрипта се сблъсках с този проблем също. В онзи момент, не успях да стартирам предаването на данни чрез API и реших да напиша на разработчика на библиотеката NiPyAPI и да поискам съвет/помощ. Разработчикът ми отговори, обсъдихме проблема и той каза, че му трябва време “да провери нещо”. И ето, след няколко дни получавам писмо, в което е написана функция на Python, решаваща моят проблем с пускането!!! В онзи момент версията на NiPyAPI беше 0.13.3 и в нея, разбира се, нищо такова нямаше. А в версия 0.14.0, която излезе съвсем наскоро, тази функция вече е включена в библиотеката. Запознайте се с

nipyapi.canvas.set_remote_process_group_transmission

И така, с библиотеката NiPyAPI свързахме registry, внедрихме flow и дори стартирахме процесорите и прехвърлянето на данни. След това можем да преработим кода, да добавим различни проверки, логиране и всичко останало. Но това е съвсем друга история.

От разгледаните от мен варианти за автоматизация, последният ми се стори най-работоспособен. Първо, все пак става въпрос за код на Python, в който можем да вграждаме помощен код и да се възползваме от всички предимства на програмния език. На второ място, проектът NiPyAPI активно се развива и при проблеми можем да се свържем с разработчика. На трето място, NiPyAPI е по-гъвкав инструмент за взаимодействие с NiFi при решаване на сложни задачи. Например, за определяне дали опашките за съобщения в момента са празни във flow и дали можем да актуализираме process group.

С това приключваме. Описах 3 подхода за автоматизация на доставката на flow в NiFi, препятствията, с които може да се сблъска разработчикът, и предоставих работен код за автоматизация на доставката. Ако и вас, както и мен, ви интересува тази тема — пишете!

Източник: habr.com