გაუმარჯოს ყველას!

ამოცანა შემდეგია - არის ნაკადი, წარმოდგენილია ზემოთ სურათზე, რომელიც უნდა გადაიტანოს N სერვერებზე Apache NiFi. ნაკადის ტესტი - მიმდინარეობს ფაილის გენერირება და გაგზავნა სხვა NiFi ინსტანციაზე. მონაცემთა გადაცემა ხდება NiFi Site to Site პროტოკოლის გამოყენებით.

NiFi Site to Site (S2S) არის უსაფრთხო, ადვილად კონფიგურირებადი გზა მონაცემთა გადასატანად NiFi ინსტანციებს შორის. როგორ მუშაობს S2S, იხილეთ დოკუმენტაცია და მნიშვნელოვანია არ დაგავიწყდეთ NiFi ინსტანციის კონფიგურაცია, რათა დაუშვას S2S, იხ აქ.

იმ შემთხვევებში, როდესაც ვსაუბრობთ მონაცემთა გადაცემაზე S2S-ის გამოყენებით, ერთ მაგალითს ეწოდება კლიენტი, მეორე სერვერი. კლიენტი აგზავნის მონაცემებს, სერვერი იღებს. მათ შორის მონაცემთა გადაცემის კონფიგურაციის ორი გზა:

1. დააყენებს. კლიენტის მაგალითიდან მონაცემები იგზავნება დისტანციური პროცესის ჯგუფის (RPG) გამოყენებით. სერვერის მაგალითზე მონაცემები მიიღება შეყვანის პორტის გამოყენებით
2. გაიგეთ. სერვერი იღებს მონაცემებს RPG-ის გამოყენებით, კლიენტი აგზავნის გამომავალი პორტის გამოყენებით.

გასაშლელი ნაკადი ინახება Apache Registry-ში.

Apache NiFi Registry არის Apache NiFi-ის ქვეპროექტი, რომელიც უზრუნველყოფს ინსტრუმენტს ნაკადის შენახვისა და ვერსიის კონტროლისთვის. ერთგვარი GIT. ინფორმაცია ინსტალაციის, კონფიგურაციისა და რეესტრთან მუშაობის შესახებ შეგიძლიათ იხილოთ აქ ოფიციალური დოკუმენტაცია. შესანახად ნაკადი გაერთიანებულია პროცესის ჯგუფში და ინახება ამ ფორმით რეესტრში. ამას მოგვიანებით სტატიაში დავუბრუნდებით.

დასაწყისში, როდესაც N არის მცირე რიცხვი, ნაკადი მიეწოდება და განახლდება ხელით მისაღებ დროში.

მაგრამ რაც N იზრდება, პრობლემები უფრო მრავალრიცხოვანი ხდება:

1. ნაკადის განახლებას მეტი დრო სჭირდება. თქვენ უნდა შეხვიდეთ ყველა სერვერზე
2. შაბლონის განახლების შეცდომები ხდება. აქ განაახლეს, მაგრამ აქ დაივიწყეს
3. ადამიანური შეცდომები დიდი რაოდენობით მსგავსი ოპერაციების შესრულებისას

ეს ყველაფერი მიგვიყვანს იქამდე, რომ პროცესის ავტომატიზაცია გვჭირდება. მე ვცადე შემდეგი გზები ამ პრობლემის გადასაჭრელად:

1. გამოიყენეთ MiNiFi ნაცვლად NiFi
2. NiFi CLI
3. NiPyAPI

MiNiFi-ის გამოყენებით

Apache MiNiFy - Apache NiFi-ს ქვეპროექტი. MiNiFy არის კომპაქტური აგენტი, რომელიც იყენებს იგივე პროცესორებს, როგორც NiFi, რაც საშუალებას გაძლევთ შექმნათ იგივე ნაკადები, როგორც NiFi-ში. აგენტის მსუბუქი ბუნება მიიღწევა, სხვა საკითხებთან ერთად, იმით, რომ MiNiFy-ს არ აქვს გრაფიკული ინტერფეისი ნაკადის კონფიგურაციისთვის. MiNiFy-ში გრაფიკული ინტერფეისის არარსებობა ნიშნავს, რომ აუცილებელია მინიფიზე ნაკადის მიწოდების პრობლემის გადაჭრა. ვინაიდან MiNiFy აქტიურად გამოიყენება IOT-ში, ბევრი კომპონენტია და საბოლოო მინი ინსტანციებზე ნაკადის მიწოდების პროცესი ავტომატიზირებულია. ნაცნობი დავალება, არა?

ამ პრობლემის მოგვარებაში დაგეხმარებათ კიდევ ერთი ქვეპროექტი - MiNiFi C2 სერვერი. ეს პროდუქტი განკუთვნილია იყოს ცენტრალური წერტილი კონფიგურაციის განლაგების არქიტექტურაში. როგორ დავაკონფიგურიროთ გარემო - აღწერილია ეს მუხლი ჰაბრეს შესახებ საკმარისი ინფორმაციაა პრობლემის მოსაგვარებლად. MiNiFi, C2 სერვერთან ერთად, ავტომატურად ანახლებს მის კონფიგურაციას. ამ მიდგომის ერთადერთი ნაკლი ის არის, რომ თქვენ უნდა შექმნათ შაბლონები C2 სერვერზე; რეესტრის მარტივი დაშვება საკმარისი არ არის.

ზემოთ მოცემულ სტატიაში აღწერილი ვარიანტი მუშაობს და არ არის რთული განსახორციელებელი, მაგრამ არ უნდა დაგვავიწყდეს შემდეგი:

1. Minifi-ს არ აქვს ყველა პროცესორი nifi-დან
2. მინიფი პროცესორის ვერსიები ჩამორჩება NiFi პროცესორის ვერსიებს.

წერის დროს NiFi-ის უახლესი ვერსიაა 1.9.2. MiniFi პროცესორის უახლესი ვერსია არის 1.7.0. პროცესორები შეიძლება დაემატოს MiNiFi-ს, მაგრამ NiFi და MiniFi პროცესორებს შორის ვერსიების შეუსაბამობის გამო, ეს შეიძლება არ იმუშაოს.

NiFi CLI

ვიმსჯელებთ აღწერა ინსტრუმენტი ოფიციალურ ვებსაიტზე, ეს არის ინსტრუმენტი NiFI-სა და NiFi რეესტრის შორის ურთიერთქმედების ავტომატიზაციისთვის ნაკადის მიწოდების ან პროცესის მართვის სფეროში. დასაწყებად, თქვენ უნდა ჩამოტვირთოთ ეს ინსტრუმენტი. აქედან გამომდინარე,.

გაუშვით კომუნალური პროგრამა

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

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

იმისათვის, რომ რეესტრიდან ჩავტვირთოთ საჭირო ნაკადი, უნდა ვიცოდეთ bucket-ის (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 მეთოდი სერვერზე მონაცემების გადასატანად. მონაცემთა გადაცემის ორგანიზებისთვის, თქვენ უნდა ჩართოთ მონაცემთა გადაცემა დამატებულ Remote Process Group-ზე (RPG), რომელიც უკვე შედის ჩვენს ნაკადში.

CLI-ში და სხვა წყაროებში არსებულ დოკუმენტაციაში მე ვერ ვიპოვე გზა მონაცემთა გადაცემის გასააქტიურებლად. თუ იცით როგორ გააკეთოთ ეს, დაწერეთ კომენტარებში.

რაკი ბაში გვაქვს და მზად ვართ ბოლომდე წავიდეთ, გამოსავალს ვიპოვით! ამ პრობლემის მოსაგვარებლად შეგიძლიათ გამოიყენოთ NiFi API. მოდით გამოვიყენოთ შემდეგი მეთოდი, ავიღოთ ID ზემოთ მოცემული მაგალითებიდან (ჩვენს შემთხვევაში ეს არის 7f522a13-016e-1000-e504-d5b15587f2f3). NiFi API მეთოდების აღწერა აქ.

სხეულში თქვენ უნდა გაიაროთ JSON, ასე:

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

პარამეტრები, რომლებიც უნდა იყოს შევსებული, რომ ის იმუშაოს:
იყო - მონაცემთა გადაცემის სტატუსი. ხელმისაწვდომია: TRANSMITTING მონაცემთა გადაცემის ჩასართავად, STOPPED გასათიშად
ვერსია - პროცესორის ვერსია

ვერსიის შექმნისას ნაგულისხმევი იქნება 0, მაგრამ ამ პარამეტრების მიღება შესაძლებელია მეთოდის გამოყენებით

bash სკრიპტების მოყვარულთათვის ეს მეთოდი შეიძლება ჩანდეს შესაფერისი, მაგრამ ჩემთვის ცოტა რთულია - bash სკრიპტები არ არის ჩემი საყვარელი. შემდეგი მეთოდი უფრო საინტერესო და მოსახერხებელია ჩემი აზრით.

NiPyAPI

NiPyAPI არის პითონის ბიბლიოთეკა NiFi ინსტანციებთან ურთიერთობისთვის. დოკუმენტაციის გვერდი შეიცავს ბიბლიოთეკასთან მუშაობისთვის საჭირო ინფორმაციას. სწრაფი დაწყება აღწერილია პროექტი github-ზე.

ჩვენი სკრიპტი კონფიგურაციის გასახსნელად არის პროგრამა Python-ში. მოდით გადავიდეთ კოდირებაზე.
ჩვენ დავაყენეთ კონფიგურაციები შემდგომი მუშაობისთვის. დაგვჭირდება შემდეგი პარამეტრები:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #путь до nifi-api инстанса, на котором разворачиваем process group
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #путь до nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeutifulRegistry' #название registry, как будет называться в инстансе nifi
nipyapi.config.bucket_name = 'BucketName' #название bucket, из которого подтягиваем flow
nipyapi.config.flow_name = 'FlowName' #название flow, которое подтягиваем

შემდეგ მე ჩავსვამ ამ ბიბლიოთეკის მეთოდების სახელებს, რომლებიც აღწერილია აქ.

დააკავშირეთ რეესტრი nifi ინსტანციაზე გამოყენებით

nipyapi.versioning.create_registry_client

ამ ეტაპზე, თქვენ ასევე შეგიძლიათ დაამატოთ შემოწმება, რომ რეესტრი უკვე დამატებულია მაგალითზე; ამისათვის შეგიძლიათ გამოიყენოთ მეთოდი

nipyapi.versioning.list_registry_clients

ჩვენ ვპოულობთ თაიგულს კალათაში ნაკადის შემდგომი საძიებლად

nipyapi.versioning.get_registry_bucket

ნაპოვნი თაიგულის გამოყენებით, ჩვენ ვეძებთ ნაკადს

nipyapi.versioning.get_flow_in_bucket

შემდეგი, მნიშვნელოვანია გვესმოდეს, უკვე დაემატა თუ არა ეს პროცესი ჯგუფი. პროცესის ჯგუფი მოთავსებულია კოორდინატების მიხედვით და შეიძლება წარმოიშვას სიტუაცია, როდესაც მეორე კომპონენტი ზემოდან ერთს. მე შევამოწმე, ეს შეიძლება მოხდეს :) ყველა დამატებული პროცესის ჯგუფის მისაღებად ჩვენ ვიყენებთ მეთოდს

nipyapi.canvas.list_all_process_groups

ჩვენ შეგვიძლია შემდგომი ძებნა, მაგალითად, სახელით.

მე არ აღვწერ შაბლონის განახლების პროცესს, მხოლოდ ვიტყვი, რომ თუ პროცესორები დაემატება შაბლონის ახალ ვერსიას, მაშინ რიგებში შეტყობინებების არსებობის პრობლემა არ არის. მაგრამ თუ პროცესორები ამოღებულია, მაშინ შეიძლება წარმოიშვას პრობლემები (nifi არ გაძლევთ საშუალებას ამოიღოთ პროცესორი, თუ მის წინ შეტყობინებების რიგი დაგროვდა). თუ გაინტერესებთ როგორ მოვაგვარე ეს პრობლემა, მომწერეთ და განვიხილავთ ამ საკითხს. კონტაქტები სტატიის ბოლოს. მოდით გადავიდეთ პროცესის ჯგუფის დამატების საფეხურზე.

სკრიპტის გამართვისას, მე შევხვდი თავისებურებას, რომ ნაკადის უახლესი ვერსია ყოველთვის არ არის ამოღებული, ამიტომ გირჩევთ ჯერ შეამოწმოთ ეს ვერსია:

nipyapi.versioning.get_latest_flow_ver

პროცესის ჯგუფის დაყენება:

nipyapi.versioning.deploy_flow_version

ჩვენ ვიწყებთ პროცესორებს:

nipyapi.canvas.schedule_process_group

CLI-ის შესახებ ბლოკში ეწერა რომ მონაცემთა გადაცემა ავტომატურად არ არის ჩართული დისტანციური პროცესის ჯგუფში? სკრიპტის განხორციელებისას მეც შემექმნა ეს პრობლემა. ამ დროს მე ვერ შევძელი მონაცემთა გადაცემის დაწყება API-ის გამოყენებით და გადავწყვიტე მივწერო NiPyAPI ბიბლიოთეკის დეველოპერს და ვთხოვო რჩევა/დახმარება. დეველოპერმა გამომეხმაურა, განვიხილეთ პრობლემა და მან დაწერა, რომ დრო სჭირდებოდა „რაღაცის შესამოწმებლად“. და შემდეგ, ორიოდე დღის შემდეგ, მოდის წერილი, რომელშიც პითონში წერია ფუნქცია, რომელიც აგვარებს ჩემს გაშვების პრობლემას!!! იმ დროს NiPyAPI ვერსია იყო 0.13.3 და, რა თქმა უნდა, მსგავსი არაფერი იყო. მაგრამ 0.14.0 ვერსიაში, რომელიც ცოტა ხნის წინ გამოვიდა, ეს ფუნქცია უკვე იყო ბიბლიოთეკაში. Შეხვედრა,

nipyapi.canvas.set_remote_process_group_transmission

ასე რომ, NiPyAPI ბიბლიოთეკის გამოყენებით, ჩვენ დავაკავშირეთ რეესტრი, გავაფართოვეთ ნაკადი და დავიწყეთ პროცესორები და მონაცემთა გადაცემაც. შემდეგ თქვენ შეგიძლიათ დაამარცხოთ კოდი, დაამატოთ ყველა სახის ჩეკი, ჩაწერა და ეს ყველაფერია. მაგრამ ეს სრულიად განსხვავებული ამბავია.

ავტომატიზაციის ვარიანტებიდან, რაც მე განვიხილეთ, ბოლო მეჩვენებოდა ყველაზე ეფექტური. ჯერ ერთი, ეს ჯერ კიდევ პითონის კოდია, რომელშიც შეგიძლიათ ჩასვათ დამხმარე პროგრამის კოდი და ისარგებლოთ პროგრამირების ენის ყველა უპირატესობით. მეორეც, NiPyAPI პროექტი აქტიურად ვითარდება და პრობლემების შემთხვევაში შეგიძლიათ მიწეროთ დეველოპერს. მესამე, NiPyAPI კვლავ უფრო მოქნილი ინსტრუმენტია NiFi-სთან ურთიერთობისთვის რთული პრობლემების გადასაჭრელად. მაგალითად, იმის დასადგენად, არის თუ არა შეტყობინებების რიგები ცარიელი ნაკადში და შესაძლებელია თუ არა პროცესის ჯგუფის განახლება.

Სულ ეს არის. მე აღვწერე 3 მიდგომა ნაკადის მიწოდების ავტომატიზაციისთვის NiFi-ში, ხარვეზები, რომლებიც შეიძლება დეველოპერმა წააწყდეს და მივაწოდე სამუშაო კოდი მიწოდების ავტომატიზაციისთვის. თუ თქვენც ისევე გაინტერესებთ ეს თემა, როგორც მე - დაწერე!

წყარო: www.habr.com