Բարեւ բոլորին!

Պետք է լուծել հետևյալ խնդիրը՝ կա 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-ին։ RG-ի տեղադրման, կարգավորմամբ և աշխատմամբ կապված տեղեկությունները կարելի է գտնել պաշտոնական փաստաթղթերի։ Flow-ը, որը կպահանջվի պահպանել, միավորվում է process group-ի մեջ և այդ ձևաչափով պահվում է registry-ում։ Կա նաև այլ շրջադարձ, որին հետագայում կվերադառնանք։

Սկզբում, երբ N-n փոքր թիվ է, flow-ը ձեռքով է տրամադրվում և արդիականացվում ընդունելի ժամանակահատվածում։

Բայց N-ի աճին伴, խնդիրները շատանում են՝

1. flow-ի արդիականացման համար ավելի շատ ժամանակ է պահանջվում։ Պետք է այցելել բոլոր սերվերները։
2. առաջանում են շablոնների արդիականացման սխալներ։ Այստեղ թարմացրել ենք, բայց այստեղ մոռացել ենք։
3. մարդու սխալներ, երբ իրականացվում են շատ однотипный գործողություններ։

Այս ամենը մեզ հասցնում է այն մտքին, որ պետք է ավտոմատացնել գործընթացը։ Ես փորձել եմ հետևյալ լուծման մեթոդները՝

1. Օգտագործել MiNiFi փոխարեն NiFi
2. NiFi CLI
3. NiPyAPI

MiNiFi օգտագործել։

Apache MiNiFy — Apache NiFi-ի ենթախնդիր է։ MiNiFy՝ կոմպակտ գործակալ, որը օգտագործում է նույն պրոցեսորները, ինչ NiFi-ը, թույլ է տալիս ստեղծել այն նույն flow-երը, որոնք օգտագործվում են NiFi-ում։ Գործակալի թեթևությունը հասնում է նաև այս պատճառով, որ MiNiFy-ով չկա flow-ի կարգավորման գրաֆիկական ոլորտ։ MiNiFy-ի գրաֆիկական միջոցի բացակայությունը նշանակում է, որ պետք է լուծել flow-ի փորձադիմումը minifi-ի մեջ։ Որպեսզի, MiNiFy ակտիվորեն օգտագործվում է IOT-ում, բաղադրիչները շատ են, և flow-ի տրամադրման գործընթացը վերջնական minifi ստորակետերին ավտոմատացնել է պետք։ Familiar task, right?

Բազմաթիվ խնդիրներ լուծելու համար ևս մեկ ենթախնդիր՝ 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.

Որպեսզի գրանցմամբ անհրաժեշտ flow-ը ներբեռնենք, պետք է իմանանք բաքետի (bucket identifier) և ինքնին flow-ի (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), որը արդեն ներառված է մեր flow-ում, հանել տվյալների փոխանցումը (Enable transmitting)։

Շռնդահարը CLI և այլ աղբյուրներում, ես չգտա տվյալները ներառելու եղանակը։ Եթե դուք գիտեք, թե ինչպես դա անել՝ խնդրում եմ գրեք մեկնաբանություններում։

Միճապես, քանի որ մենք bash-ով ու հաստատակամ ենք գնալու մինչև վերջ՝ գտնենք ելքը։ Կարելի է օգտվել NiFi API-ից այս խնդիրը լուծելու համար։ Օգտվենք հետևյալ մեթոդով՝ ID-ն վերցնում ենք վերևի օրինակներից (մեր դեպքում դա 7f522a13-016e-1000-e504-d5b15587f2f3)։ NiFi API մեթոդների նկարագիր՝ այստեղ.

Body-ում պետք է փոխանցել JSON, հետևյալ տեսքով՝

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

Լրացնելու պարամետրերը, որպեսզի «սպանել» փոխարենը՝
state — տվյալների փոխանցման կարգավիճակը։ Ավելացվել է TRANSMITTING՝ տվյալների փոխանցումը անջատելու համար STOPPED
version — պրոցեսորի տարբերակը

ստանդարտ տարբերակը 0 կլինի ստեղծման ժամանակ, բայց այս պարամետրերը հնարավոր է ձեռք բերել օգտագործելով մեթոդը

Բաշ սցենարների սիրահարների համար այս մեթոդը կարող է համարվել օգտակար, բայց ինձ է դժվար թվում՝ բաշ սցենարները իմ սիրելիները չեն: Հաջորդ եղանակը հետաքրքիր և ավելի հարմար է իմ կարծիքով:

NiPyAPI

NiPyAPI — Python լեզվի գրադարան, որը նախատեսված է 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 գրանցարանի հասցեն
nipyapi.config.registry_name = 'MyBeutifulRegistry' #ի՞նչ անուն կունենա գրանցարանը 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

Նորենք bucket-ը, որպեսզի հետագայում փնտրենք flow-ը

nipyapi.versioning.get_registry_bucket

Ն gevonden bucket-ի հիման վրա փնտրենք flow-ը

nipyapi.versioning.get_flow_in_bucket

Հաջորդը կարևոր է հասկանալ, թե արդյոք այս process group-ը արդեն ավելացված է: Process group-ը տեղադրվում է համակարգի վրա և կարող է լինել իրավիճակ, որտեղ մի komponentի վրա երկրորդը կընկնի: Ես ստուգել եմ, դա կարող է լինել 🙂 Ինչպես ստանալ բոլոր հավելված process group-երը, օգտագործում ենք մեթոդը

nipyapi.canvas.list_all_process_groups

և հետո կարող ենք փնտրել օրինակ անունով:

Ես չեմ նկարագրելու սանվածքի թարմացման գործընթացը, միայն կասեմ, որ եթե նոր տարբերակում processors ավելանում են, ապա խնդիրներ չեն առաջանում ընդհանուր հերթերի առկայության հետ: Բայց եթե processors ջնջվում են, ապա խնդիրներ կարող են առաջանալ (nifi չի թույլատրում ջնջել processor, եթե նրա դիմաց հավաքվել է հաղորդումների հերթ): Եթե ձեզ հետաքրքրում է, թե ինչպես ես լուծել այս խնդիրը, խնդրում եմ գրեք ինձ, խոսենք այդ մասին: Կոնտակտները この記事の最後にあります。

Սցենարը debug անելու ժամանակ ես հանդիպեցի այնպիսի հատուկություններին, որ միշտ չէ, որ վերջին տարբերակը նետի, ուստի խորհուրդ եմ տալիս նախ ճշտել այս տարբերակը:

nipyapi.versioning.get_latest_flow_ver

Deploying process group:

nipyapi.versioning.deploy_flow_version

Վթարեք processors:

nipyapi.canvas.schedule_process_group

CLI блокում գրված էր, որ հեռավոր գործընթացների խմբում ավտոմատ կերպով տվյալների փոխանցումը չի ներառվում: Երբ ես իրականացնում էի սկրիպտը, նույնպես այս խնդրի առաջ կանգնեցի: Այդ ժամանակ 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-ում, և կարելի՞ է թարմացնել գործընթացների խումբը:

Ահա, սա ամենը: Ես նկարագրեցի NiFi-ում flow-ի ավտոմատացման 3 մոտեցումները, ծրագրողի առջև ծառացած թաքնված խնդիրները և տվեցի աշխատող կոդ ավտոմատացման համար: Եթե դու նաև, ինչպես ես, հետաքրքրված ես այս թեմայով — գրիր!

Ընտանիք: habr.com