हॅलो प्रत्येकजण!

कार्य खालीलप्रमाणे आहे - वरील चित्रात एक प्रवाह आहे, जो एन सर्व्हरवर आणला जाणे आवश्यक आहे अपाचे निफाय. फ्लो टेस्ट - फाईल तयार केली जात आहे आणि दुसर्‍या NiFi उदाहरणावर पाठवली जात आहे. डेटा हस्तांतरण NiFi साइट टू साइट प्रोटोकॉल वापरून होते.

NiFi साइट टू साइट (S2S) हा NiFi उदाहरणांमध्ये डेटा हस्तांतरित करण्याचा एक सुरक्षित, सहज कॉन्फिगर करण्यायोग्य मार्ग आहे. S2S कसे कार्य करते, पहा दस्तऐवजीकरण आणि S2S ला अनुमती देण्यासाठी NiFi उदाहरण कॉन्फिगर करणे विसरू नका, हे महत्वाचे आहे येथे.

ज्या प्रकरणांमध्ये आम्ही S2S वापरून डेटा ट्रान्सफरबद्दल बोलत आहोत, एका उदाहरणाला क्लायंट म्हणतात, दुसरा सर्व्हर. क्लायंट डेटा पाठवतो, सर्व्हर प्राप्त करतो. त्यांच्या दरम्यान डेटा ट्रान्सफर कॉन्फिगर करण्याचे दोन मार्ग:

1. पुश करा. क्लायंटच्या उदाहरणावरून, रिमोट प्रोसेस ग्रुप (RPG) वापरून डेटा पाठवला जातो. सर्व्हरच्या उदाहरणावर, इनपुट पोर्ट वापरून डेटा प्राप्त होतो
2. खेचणे. सर्व्हर आरपीजी वापरून डेटा प्राप्त करतो, क्लायंट आउटपुट पोर्ट वापरून पाठवतो.

रोलिंगसाठी प्रवाह Apache Registry मध्ये संग्रहित केला जातो.

Apache NiFi Registry हा Apache NiFi चा उपप्रकल्प आहे जो प्रवाह संचयन आणि आवृत्ती नियंत्रणासाठी एक साधन प्रदान करतो. GIT चा एक प्रकार. रेजिस्ट्री स्थापित करणे, कॉन्फिगर करणे आणि कार्य करणे याबद्दल माहिती मिळू शकते अधिकृत दस्तऐवजीकरण. स्टोरेजसाठी प्रवाह प्रक्रिया गटामध्ये एकत्रित केला जातो आणि या फॉर्ममध्ये नोंदणीमध्ये संग्रहित केला जातो. आम्ही लेखात नंतर याकडे परत येऊ.

सुरुवातीला, जेव्हा N ही संख्या लहान असते, तेव्हा प्रवाह वितरित केला जातो आणि स्वीकार्य वेळेत व्यक्तिचलितपणे अद्यतनित केला जातो.

परंतु जसजसे N वाढते तसतसे समस्या अधिकाधिक वाढतात:

1. प्रवाह अद्यतनित करण्यासाठी अधिक वेळ लागतो. तुम्हाला सर्व सर्व्हरमध्ये लॉग इन करणे आवश्यक आहे
2. टेम्प्लेट अपडेट करताना चुका होतात. येथे त्यांनी ते अद्यतनित केले, परंतु येथे ते विसरले
3. मोठ्या संख्येने समान ऑपरेशन्स करताना मानवी चुका

हे सर्व आम्हाला या वस्तुस्थितीकडे आणते की आम्हाला प्रक्रिया स्वयंचलित करणे आवश्यक आहे. या समस्येचे निराकरण करण्यासाठी मी खालील मार्गांनी प्रयत्न केला:

1. NiFi ऐवजी MiNiFi वापरा
2. NiFi CLI
3. NiPyAPI

MiNiFi वापरणे

Apache MiNiFy - Apache NiFi चा उपप्रकल्प. MiNiFy हा एक कॉम्पॅक्ट एजंट आहे जो NiFi प्रमाणेच प्रोसेसर वापरतो, जो तुम्हाला NiFi प्रमाणेच प्रवाह तयार करण्यास अनुमती देतो. MiNiFy मध्ये फ्लो कॉन्फिगरेशनसाठी ग्राफिकल इंटरफेस नसल्यामुळे एजंटचे हलके स्वरूप इतर गोष्टींबरोबरच प्राप्त होते. MiNiFy मध्‍ये ग्राफिकल इंटरफेस नसल्‍याचा अर्थ minifi वर प्रवाह वितरीत करण्‍याची समस्या सोडवणे आवश्‍यक आहे. MiNiFy चा IOT मध्ये सक्रियपणे वापर केला जात असल्याने, त्यात अनेक घटक आहेत आणि अंतिम मिनीफाय घटनांमध्ये प्रवाह वितरीत करण्याची प्रक्रिया स्वयंचलित असणे आवश्यक आहे. एक परिचित कार्य, बरोबर?

आणखी एक उपप्रकल्प या समस्येचे निराकरण करण्यात मदत करेल - MiNiFi C2 सर्व्हर. हे उत्पादन कॉन्फिगरेशन रोलआउट आर्किटेक्चरमधील मध्यवर्ती बिंदू बनण्याचा हेतू आहे. वातावरण कसे कॉन्फिगर करावे - मध्ये वर्णन केले आहे हा लेख समस्या सोडवण्यासाठी Habré वर पुरेशी माहिती आहे. 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.

आम्हाला रेजिस्ट्रीमधून आवश्यक प्रवाह लोड करण्यासाठी, आम्हाला बकेटचे अभिज्ञापक (बकेट आयडेंटिफायर) आणि प्रवाह स्वतः (फ्लो आयडेंटिफायर) माहित असणे आवश्यक आहे. हा डेटा एकतर 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 उदाहरणे आवश्यक आहेत. आपण सर्व्हरवर डेटा हस्तांतरित करण्यासाठी पुश पद्धत निवडली आहे असे मानू या. डेटा ट्रान्सफर आयोजित करण्यासाठी, तुम्हाला जोडलेल्या रिमोट प्रोसेस ग्रुप (RPG) वर डेटा ट्रान्सफर सक्षम करणे आवश्यक आहे, जे आमच्या प्रवाहात आधीपासूनच समाविष्ट आहे.

CLI आणि इतर स्त्रोतांमधील दस्तऐवजीकरणामध्ये, मला डेटा ट्रान्सफर सक्षम करण्याचा मार्ग सापडला नाही. हे कसे करायचे हे आपल्याला माहित असल्यास, टिप्पण्यांमध्ये लिहा.

आमची बाश आहे आणि आम्ही शेवटपर्यंत जायला तयार आहोत, आम्हाला मार्ग सापडेल! ही समस्या सोडवण्यासाठी तुम्ही NiFi API वापरू शकता. चला खालील पद्धत वापरुया, वरील उदाहरणांमधून आयडी घेऊ (आमच्या बाबतीत ते 7f522a13-016e-1000-e504-d5b15587f2f3 आहे). NiFi API पद्धतींचे वर्णन येथे.

शरीरात तुम्हाला JSON पास करणे आवश्यक आहे, जसे:

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

हे कार्य करण्यासाठी पॅरामीटर्स भरणे आवश्यक आहे:
राज्य - डेटा ट्रान्सफर स्थिती. उपलब्ध: डेटा ट्रान्सफर सक्षम करण्यासाठी ट्रान्समिटिंग, अक्षम करण्यासाठी थांबवले
आवृत्ती - प्रोसेसर आवृत्ती

आवृत्ती तयार केल्यावर 0 वर डीफॉल्ट असेल, परंतु हे पॅरामीटर्स पद्धत वापरून मिळवता येतात

बॅश स्क्रिप्टच्या चाहत्यांसाठी, ही पद्धत योग्य वाटू शकते, परंतु माझ्यासाठी ती थोडी अवघड आहे - बॅश स्क्रिप्ट माझ्या आवडत्या नाहीत. पुढील पद्धत माझ्या मते अधिक मनोरंजक आणि सोयीस्कर आहे.

NiPyAPI

NiFi उदाहरणांशी संवाद साधण्यासाठी NiPyAPI ही पायथन लायब्ररी आहे. दस्तऐवजीकरण पृष्ठ लायब्ररीसोबत काम करण्यासाठी आवश्यक माहिती समाविष्ट आहे. जलद प्रारंभ मध्ये वर्णन केले आहे प्रकल्प github वर.

कॉन्फिगरेशन रोल आउट करण्यासाठी आमची स्क्रिप्ट पायथनमधील एक प्रोग्राम आहे. चला कोडिंगकडे वळू.
आम्ही पुढील कामासाठी कॉन्फिग्स सेट करतो. आम्हाला खालील पॅरामीटर्सची आवश्यकता असेल:

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

आम्ही पुढे शोधू शकतो, उदाहरणार्थ, नावाने.

मी टेम्पलेट अद्यतनित करण्याच्या प्रक्रियेचे वर्णन करणार नाही, मी फक्त असे म्हणेन की जर टेम्पलेटच्या नवीन आवृत्तीमध्ये प्रोसेसर जोडले गेले असतील तर रांगेत संदेशांच्या उपस्थितीत कोणतीही समस्या नाही. परंतु जर प्रोसेसर काढून टाकले गेले तर समस्या उद्भवू शकतात (निफाय तुम्हाला प्रोसेसर काढण्याची परवानगी देत ​​​​नाही जर संदेशाची रांग समोर जमा झाली असेल). मी या समस्येचे निराकरण कसे केले याबद्दल आपल्याला स्वारस्य असल्यास, कृपया मला लिहा आणि आम्ही या समस्येवर चर्चा करू. लेखाच्या शेवटी संपर्क. प्रक्रिया गट जोडण्याच्या पायरीवर जाऊ या.

स्क्रिप्ट डीबग करताना, मला एक वैशिष्ठ्य आढळले की प्रवाहाची नवीनतम आवृत्ती नेहमी खेचली जात नाही, म्हणून मी प्रथम ही आवृत्ती तपासण्याची शिफारस करतो:

nipyapi.versioning.get_latest_flow_ver

प्रक्रिया गट तैनात करा:

nipyapi.versioning.deploy_flow_version

आम्ही प्रोसेसर सुरू करतो:

nipyapi.canvas.schedule_process_group

सीएलआय बद्दलच्या ब्लॉकमध्ये असे लिहिले होते की रिमोट प्रोसेस ग्रुपमध्ये डेटा ट्रान्सफर स्वयंचलितपणे सक्षम होत नाही? स्क्रिप्टची अंमलबजावणी करताना मलाही ही समस्या आली. त्या वेळी, मी API वापरून डेटा ट्रान्सफर सुरू करू शकलो नाही आणि मी NiPyAPI लायब्ररीच्या विकसकाला लिहून सल्ला/मदत विचारण्याचे ठरवले. विकसकाने मला प्रतिसाद दिला, आम्ही समस्येवर चर्चा केली आणि त्याने लिहिले की त्याला "काहीतरी तपासण्यासाठी" वेळ हवा आहे. आणि मग, काही दिवसांनंतर, एक पत्र येते ज्यामध्ये पायथनमध्ये एक फंक्शन लिहिलेले आहे जे माझ्या लॉन्च समस्येचे निराकरण करते !!! त्या वेळी, NiPyAPI आवृत्ती 0.13.3 होती आणि अर्थातच असे काहीही नव्हते. परंतु आवृत्ती 0.14.0 मध्ये, जी अगदी अलीकडे रिलीझ झाली होती, हे कार्य आधीच लायब्ररीमध्ये समाविष्ट केले गेले होते. भेटा,

nipyapi.canvas.set_remote_process_group_transmission

म्हणून, NiPyAPI लायब्ररी वापरून, आम्ही रेजिस्ट्री कनेक्ट केली, प्रवाह आणला आणि प्रोसेसर आणि डेटा ट्रान्सफर देखील सुरू केला. मग तुम्ही कोड एकत्र करू शकता, सर्व प्रकारचे चेक जोडू शकता, लॉगिंग करू शकता आणि एवढेच. पण ती पूर्णपणे वेगळी कथा आहे.

मी विचारात घेतलेल्या ऑटोमेशन पर्यायांपैकी, शेवटचा पर्याय मला सर्वात कार्यक्षम वाटला. प्रथम, हा अजूनही पायथन कोड आहे, ज्यामध्ये तुम्ही सहाय्यक प्रोग्राम कोड एम्बेड करू शकता आणि प्रोग्रामिंग भाषेच्या सर्व फायद्यांचा लाभ घेऊ शकता. दुसरे म्हणजे, NiPyAPI प्रकल्प सक्रियपणे विकसित होत आहे आणि समस्या असल्यास आपण विकसकाला लिहू शकता. तिसरे म्हणजे, NiPyAPI हे अजून जटिल समस्या सोडवण्यासाठी NiFi शी संवाद साधण्याचे अधिक लवचिक साधन आहे. उदाहरणार्थ, संदेशाच्या रांगा आता प्रवाहात रिक्त आहेत की नाही आणि प्रक्रिया गट अद्यतनित केला जाऊ शकतो किंवा नाही हे निर्धारित करण्यासाठी.

इतकंच. मी NiFi मध्‍ये स्वयंचलित प्रवाह वितरणासाठी 3 पद्धतींचे वर्णन केले आहे, विकसकाला येऊ शकणार्‍या अडचणी, आणि स्वयंचलित वितरणासाठी कार्यरत कोड प्रदान केला आहे. जर तुम्हाला या विषयात माझ्याइतकीच रस असेल तर - लिहा

स्त्रोत: www.habr.com