Hallo allemaal!

De taak is als volgt - er is een flow, zoals weergegeven in de afbeelding hierboven, die uitgerold moet worden naar N servers met Apache NiFi. De flow is testgericht - er vindt bestandsgeneratie plaats en verzending naar een andere NiFi-instantie. Gegevensoverdracht gebeurt via het NiFi Site to Site-protocol.

NiFi Site to Site (S2S) is een veilige, eenvoudig te configureren manier om gegevens tussen NiFi-instanties over te dragen. Hoe S2S werkt, zie je in de documentatie en het is belangrijk om de NiFi-instantie in te stellen om S2S toe te staan, zie here.

In gevallen waarin gegevensoverdracht via S2S plaatsvindt, wordt één instantie de client genoemd en de andere de server. De client verzendt gegevens, de server ontvangt ze. Er zijn twee manieren om gegevensoverdracht tussen hen in te stellen:

1. Push. Van de client-instantie worden gegevens verzonden met behulp van een Remote Process Group (RPG). Op de server-instantie worden gegevens ontvangen via de Input Port
2. Pull. De server ontvangt gegevens via RPG, de cliënt verzendt via de Output port.

De flow voor uitrol wordt opgeslagen in de Apache Registry.

Apache NiFi Registry is een subproject van Apache NiFi en biedt een hulpmiddel voor het opslaan van flows en versiebeheer. Een soort GIT. Informatie over installatie, configuratie en gebruik van de registry kan gevonden worden in de officiële documentatie. De flow voor opslag wordt samengevoegd in een process group en wordt op deze manier in de registry opgeslagen. Later in het artikel zullen we hier nog op terugkomen.

Aan het begin, wanneer N een klein aantal is, wordt de flow handmatig geleverd en geactualiseerd binnen een aanvaardbare tijd.

Maar naarmate N toeneemt, worden er meer problemen:

1. het kost meer tijd om de flow te actualiseren. We moeten naar alle servers gaan
2. er ontstaan fouten bij het actualiseren van de sjablonen. Hier is het bijgewerkt, maar hier is het vergeten
3. menselijke fouten bij het uitvoeren van een groot aantal homogeen operaties

Dit leidt ons tot de conclusie dat we het proces moeten automatiseren. Ik heb de volgende manieren geprobeerd om dit probleem op te lossen:

1. Gebruik MiNiFi in plaats van NiFi
2. NiFi CLI
3. NiPyAPI

Gebruik MiNiFi

Apache MiNiFi — subproject Apache NiFi. MiNiFy is a compact agent that uses the same processors as NiFi, allowing for the creation of the same flows as in NiFi. The lightweight nature of the agent is achieved in part by the fact that MiNiFy does not have a graphical interface for configuring flows. The absence of a graphical interface in MiNiFy means that the problem of delivering flows to MiNiFy instances must be addressed. Since MiNiFy is actively used in IoT, there are many components, and the delivery of flows to the final MiNiFy instances needs to be automated. A familiar task, right?

Another subproject that helps solve this problem is the MiNiFi C2 Server. This product is designed to be the central point in the configuration rollout architecture. How to configure the environment is described in dit artikel on Habr, and there is enough information to solve the specified task. MiNiFi, in conjunction with the C2 server, automatically updates its configuration. The only downside to this approach is that templates must be created on the C2 Server; a simple commit to the registry is not sufficient.

The option described in the article above is functional and not difficult to implement, but one must not forget the following:

1. Not all processors from NiFi are available in MiNiFi.
2. The versions of processors in MiNiFi lag behind the versions of processors in NiFi.

At the time of writing, the latest version of NiFi is 1.9.2. The processor version of the latest MiNiFi version is 1.7.0. Processors can be added to MiNiFi, but due to version discrepancies between NiFi and MiNiFi processors, this may not work.

NiFi CLI

Volgens description tool on the official website, it is a tool for automating interactions between NiFi and NiFi Registry in the area of flow delivery or process management. To get started, this tool needs to be downloaded hier.

We launch the utility

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

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

To load the required flow from the registry, we need to know the bucket identifier and the flow identifier. This data can be obtained either via CLI or in the web interface of the NiFi registry. In the web interface, it looks like this:

Using CLI, it is done like this:

#> 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

We start importing the process group from the registry:

#> 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

An important point is that any NiFi instance can be specified as the host to which we deploy the process group.

Process group is toegevoegd met stopped processors, deze moeten worden gestart.

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

Geweldig, de processors zijn gestart. We moeten echter, volgens de eisen, ervoor zorgen dat de NiFi-instanties gegevens naar andere instanties sturen. Laten we aannemen dat we de Push-methode voor gegevensoverdracht naar de server hebben gekozen. Om de gegevensoverdracht te organiseren, moet je de gegevensoverdracht op de toegevoegde Remote Process Group (RPG), die al in onze flow is ingeschakeld, inschakelen (Enable transmitting).

In de documentatie in de CLI en andere bronnen heb ik geen manier gevonden om de gegevensoverdracht in te schakelen. Als je weet hoe dit moet, laat het me dan alsjeblieft weten in de opmerkingen.

Aangezien we bash hebben en bereid zijn door te zetten, laten we een oplossing vinden! We kunnen de NiFi API gebruiken om dit probleem op te lossen. Laten we de volgende methode gebruiken, waarbij we de ID uit de bovenstaande voorbeelden nemen (in ons geval is dat 7f522a13-016e-1000-e504-d5b15587f2f3). Beschrijving van de NiFi API-methoden. here.

In de body moet je JSON doorgeven van de volgende vorm:

{
    "revision": {
	    "clientId": "value",
	    "version": 0,
	    "lastModifier": "value"
	},
    "state": "value",
    "disconnectedNodeAcknowledged": true
}

De parameters die je moet invullen om het te laten werken:
state — status van gegevensoverdracht. Beschikbaar TRANSMITTING voor het inschakelen van gegevensoverdracht, STOPPED voor het uitschakelen.
version — versie van de processor.

de versie is standaard 0 bij het aanmaken, maar deze parameters kunnen worden verkregen met behulp van de methode.

Voor liefhebbers van bash-scripts lijkt deze methode misschien handig, maar het valt me zwaar — bash-scripts zijn niet mijn favoriet. De volgende manier lijkt mij interessanter en handiger.

NiPyAPI

NiPyAPI — een bibliotheek voor de programmeertaal Python voor interactie met NiFi-instanties. De pagina met documentatie bevat de noodzakelijke informatie voor het werken met de bibliotheek. De quick start is beschreven in project op github.

Ons script voor het uitrollen van de configuratie is een programma geschreven in Python. Laten we beginnen met coderen.
We configureren de instellingen voor verdere werkzaamheden. We hebben de volgende parameters nodig:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #pad naar de nifi-api instantie waarop we de process group uitrollen
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #pad naar de nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeutifulRegistry' #naam van de registry, zoals deze in de niFi-instantie zal worden genoemd
nipyapi.config.bucket_name = 'BucketName' #naam van de bucket waarvan we de flow ophalen
nipyapi.config.flow_name = 'FlowName' #naam van de flow die we ophalen.

Vervolgens zal ik de namen van de methoden van deze bibliotheek invoegen, die zijn beschreven here.

We verbinden de registry met de niFi-instantie met behulp van

nipyapi.versioning.create_registry_client

Op dit punt kunt u ook controleren of de registry al aan de instantie is toegevoegd, daarvoor kunt u de methode gebruiken

nipyapi.versioning.list_registry_clients

We vinden de bucket voor de verdere zoektocht naar flow in de mand

nipyapi.versioning.get_registry_bucket

Op basis van de gevonden bucket zoeken we flow

nipyapi.versioning.get_flow_in_bucket

Vervolgens is het belangrijk te begrijpen of deze process group al is toegevoegd. De process group wordt op coördinaten geplaatst en het kan zich voordoen dat er een tweede component bovenop de eerste komt. Ik heb dit gecontroleerd, dit kan gebeuren 🙂 Om alle toegevoegde process groups te verkrijgen gebruiken we de methode

nipyapi.canvas.list_all_process_groups

en vervolgens kunnen we bijvoorbeeld zoeken op naam.

Ik zal het proces van het bijwerken van de sjabloon niet beschrijven, ik zal alleen zeggen dat als in de nieuwe versie van de sjabloon processors worden toegevoegd, er geen problemen zijn met de aanwezigheid van berichten in de wachtrijen. Maar als processors worden verwijderd, kunnen er problemen ontstaan (nifi laat geen processor verwijderen als er een wachtrij van berichten voor ligt). Als u benieuwd bent hoe ik dit probleem heb opgelost, laat het me dan weten, dan bespreken we dit. Contactgegevens aan het einde van het artikel. Laten we verder gaan met het toevoegen van de process group.

Tijdens het debuggen van het script stuitte ik op een eigenaardigheid dat niet altijd de laatste versie van de flow wordt opgehaald, daarom raad ik aan eerst deze versie te verifiëren:

nipyapi.versioning.get_latest_flow_ver

We implementeren de process group:

nipyapi.versioning.deploy_flow_version

We starten de processors:

nipyapi.canvas.schedule_process_group

In het gedeelte over de CLI stond geschreven dat de gegevensoverdracht in de remote process group automatisch niet is ingeschakeld? Bij het implementeren van het script stuitte ik ook op dit probleem. Op dat moment kon ik de gegevensoverdracht via API niet starten en besloot ik de ontwikkelaar van de NiPyAPI-bibliotheek te schrijven en om advies/hulp te vragen. De ontwikkelaar antwoordde me, we bespraken het probleem en hij schreef dat hij tijd nodig had om "iets te controleren". En kijk, enkele dagen later komt er een brief waarin een functie in Python is beschreven die mijn probleem oplost!!! Op dat moment was de versie van NiPyAPI 0.13.3 en daarin was natuurlijk niets dergelijks. Maar in versie 0.14.0, die onlangs is uitgebracht, is deze functie al opgenomen in de bibliotheek. Maak kennis met,

nipyapi.canvas.set_remote_process_group_transmission

Dus, met behulp van de NiPyAPI-bibliotheek hebben we de registry verbonden, de flow geïmplementeerd en zelfs de processoren en gegevensoverdracht gestart. Nu kunnen we de code stroomlijnen, verschillende controles en logging toevoegen, en dat soort dingen. Maar dat is alweer een heel ander verhaal.

Van de opties voor automatisering die ik heb overwogen, leek de laatste mij de meest bruikbare. Ten eerste, het is nog steeds python-code, waarin je ondersteunende programmatuur kunt integreren en kunt profiteren van alle voordelen van de programmeertaal. Ten tweede, het NiPyAPI-project ontwikkelt zich actief en in geval van problemen kun je de ontwikkelaar benaderen. Ten derde is NiPyAPI een flexibeler hulpmiddel voor interactie met NiFi bij het oplossen van complexe vraagstukken. Bijvoorbeeld, om te bepalen of de wachtrijen voor berichten in de flow momenteel leeg zijn en of de processgroep kan worden bijgewerkt.

Dat is alles. Ik heb 3 benaderingen beschreven voor de automatisering van de levering van flow in NiFi, de valkuilen waarmee een ontwikkelaar kan worden geconfronteerd, en ik heb werkende code voor de automatisering van de levering gegeven. Als dit onderwerp je ook interesseert, net als mij — laat het me weten!

Bron: habr.com