सभी को नमस्कार!

कार्य निम्नानुसार है - ऊपर चित्र में प्रस्तुत किया गया प्रवाह है, जिसे अपाचे NiFi के साथ एन सर्वर पर रोल किया जाना चाहिए। फ्लो टेस्ट - फाइल जेनरेट की जा रही है और दूसरे NiFi इंस्टेंस पर भेजी जा रही है। डेटा को NiFi साइट से साइट प्रोटोकॉल का उपयोग करके प्रसारित किया जाता है।

NiFi साइट से साइट (S2S) NiFi उदाहरणों के बीच डेटा स्थानांतरित करने का एक सुरक्षित, आसानी से अनुकूलन योग्य तरीका है। देखें कि S2S प्रलेखन में कैसे काम करता है और यह महत्वपूर्ण है कि S2S को यहाँ देखने के लिए NiFi उदाहरण को कॉन्फ़िगर करना न भूलें।

उन मामलों में जब S2S का उपयोग करके डेटा ट्रांसफर की बात आती है - एक उदाहरण क्लाइंट कहा जाता है, दूसरा सर्वर। क्लाइंट डेटा भेजता है, सर्वर भेजता है। उनके बीच डेटा ट्रांसफर कॉन्फ़िगर करने के दो तरीके:

पुश। क्लाइंट उदाहरण से, दूरस्थ प्रक्रिया समूह (आरपीजी) का उपयोग करके डेटा भेजा जाता है। एक सर्वर उदाहरण पर, इनपुट पोर्ट का उपयोग करके डेटा प्राप्त किया जाता है।
पुल। सर्वर आरपीजी का उपयोग करके डेटा प्राप्त करता है, क्लाइंट आउटपुट पोर्ट का उपयोग करके भेजता है।

हम अपाचे रजिस्ट्री में रोलिंग के लिए प्रवाह को स्टोर करते हैं।

अपाचे NiFi रजिस्ट्री Apache NiFi का एक उपप्रोजेक्ट है जो प्रवाह और संस्करण नियंत्रण के भंडारण के लिए एक उपकरण प्रदान करता है। एक प्रकार का कण्ठ। रजिस्ट्री के साथ स्थापित करने, कॉन्फ़िगर करने और काम करने की जानकारी आधिकारिक दस्तावेज में पाई जा सकती है। भंडारण के लिए प्रवाह को एक प्रक्रिया समूह में जोड़ा जाता है और रजिस्ट्री में संग्रहीत किया जाता है। लेख में आगे हम इस पर लौटेंगे।

शुरुआत में, जब एन एक छोटी संख्या होती है, तो प्रवाह स्वीकार्य समय में हाथों द्वारा वितरित और अद्यतन किया जाता है।

लेकिन एन के विकास के साथ, अधिक समस्याएं हैं:

अद्यतन प्रवाह में अधिक समय लगता है। सभी सर्वरों पर जाना आवश्यक है
टेम्प्लेट अपडेट करने में त्रुटियाँ हैं। यहां उन्होंने अपडेट किया, लेकिन यहां वे भूल गए
बड़ी संख्या में एक ही प्रकार के संचालन करते समय मानवीय त्रुटियां

यह सब हमें इस तथ्य की ओर ले जाता है कि हमें प्रक्रिया को स्वचालित करने की आवश्यकता है। मैंने इस समस्या को हल करने के लिए निम्नलिखित तरीके आजमाए:

NiFi के बजाय MiNiFi का उपयोग करें
निफी सीएलआई
NiPyAPI

MiNiFi का उपयोग करना

Apache MiNiFy Apache NiFi का एक उपप्रोजेक्ट है। MiNiFy एक कॉम्पैक्ट एजेंट है जो NiFi के समान प्रोसेसर का उपयोग करता है, जिससे आप NiFi के समान प्रवाह बना सकते हैं। एजेंट की लपट अन्य बातों के अलावा, इस तथ्य के कारण प्राप्त होती है कि MiNiFy में फ्लो कॉन्फ़िगरेशन के लिए ग्राफिकल इंटरफ़ेस नहीं है। MiNiFy में एक ग्राफिकल इंटरफ़ेस की कमी का मतलब है कि मिनीफी में प्रवाह वितरण की समस्या को हल करना आवश्यक है। चूंकि MiNiFy IOT में सक्रिय रूप से उपयोग किया जाता है, इसलिए कई घटक हैं और अंतिम मिनीफाई इंस्टेंसेस में प्रवाह को स्वचालित करने की आवश्यकता है। एक परिचित कार्य, सही?

एक अन्य उपप्रोजेक्ट इस समस्या को हल करने में मदद करेगा - MiNiFi C2 सर्वर। यह उत्पाद रोलिंग कॉन्फ़िगरेशन की वास्तुकला में एक केंद्रीय बिंदु होने का इरादा है। पर्यावरण को कैसे कॉन्फ़िगर करें - इस लेख में वर्णित हैबे और कार्य को हल करने के लिए पर्याप्त जानकारी। C2 सर्वर के साथ संयोजन में MiNiFi स्वचालित रूप से घर पर कॉन्फ़िगरेशन को अपडेट करता है। इस दृष्टिकोण का एकमात्र दोष यह है कि आपको C2 सर्वर पर टेम्पलेट बनाने होंगे, एक सरल रजिस्ट्री कमिट पर्याप्त नहीं है।

उपरोक्त लेख में वर्णित विकल्प काम कर रहा है और इसे लागू करना मुश्किल है, लेकिन निम्नलिखित को मत भूलना:

Minifi में nifi से सभी प्रोसेसर नहीं होते हैं
Minifi के प्रोसेसर संस्करण NiFi में प्रोसेसर संस्करणों के पीछे हैं।

लेखन के समय, NiFi का नवीनतम संस्करण 1.9.2 है। नवीनतम MiNiFi संस्करण का प्रोसेसर संस्करण 1.7.0 है। प्रोसेसर को MiNiFi में जोड़ा जा सकता है, लेकिन NiFi और MiNiFi प्रोसेसर के बीच संस्करण अंतर के कारण यह काम नहीं कर सकता है।

निफी सीएलआई

आधिकारिक वेबसाइट पर उपकरण के विवरण को देखते हुए, यह प्रवाह वितरण या प्रक्रिया नियंत्रण के क्षेत्र में 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 इंस्टेंस की आवश्यकता होती है। मान लें कि आप सर्वर पर डेटा स्थानांतरित करने के लिए पुश विधि का चयन करते हैं। डेटा स्थानांतरण को व्यवस्थित करने के लिए, आपको जोड़ा गया रिमोट प्रोसेस ग्रुप (आरपीजी) पर डेटा ट्रांसमिशन (ट्रांसमिटिंग सक्षम करना) सक्षम करना होगा, जो पहले से ही हमारे प्रवाह में शामिल है।

सीएलआई और अन्य स्रोतों में प्रलेखन में, मुझे डेटा स्थानांतरण को सक्षम करने का कोई तरीका नहीं मिला। यदि आप जानते हैं कि यह कैसे करना है, तो कृपया टिप्पणियों में लिखें।

चूंकि हमारे पास बैश है और हम अंत तक जाने के लिए तैयार हैं - हम एक रास्ता निकाल लेंगे! आप इस समस्या को हल करने के लिए NiFi API का उपयोग कर सकते हैं। हम निम्नलिखित विधि का उपयोग करते हैं, हम ऊपर के उदाहरणों से आईडी लेते हैं (हमारे मामले में यह 7f522a13-016e-1000-e504-d5b15587f2f3 है)। नीफी एपीआई विधियों का वर्णन यहाँ दिया गया है ।

शरीर में, आपको निम्न फॉर्म का JSON पास करना होगा:

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

"काम" करने के लिए जो पैरामीटर भरे जाने चाहिए:
राज्य - डेटा अंतरण स्थिति। डेटा ट्रांसफर को सक्षम करने के लिए उपलब्ध ट्रांसमिशन, बंद करने के लिए बंद
संस्करण - प्रोसेसर संस्करण

संस्करण बनाया जाने पर 0 पर डिफ़ॉल्ट होगा, लेकिन इन मापदंडों को विधि का उपयोग करके प्राप्त किया जा सकता है

बैश स्क्रिप्ट के प्रेमियों के लिए, यह विधि उपयुक्त लग सकती है, लेकिन यह मेरे लिए कठिन है - बैश स्क्रिप्ट मेरे पसंदीदा नहीं हैं। मेरी राय में निम्नलिखित विधि अधिक दिलचस्प और आरामदायक है।

NiPyAPI

NiPyAPI NiFi उदाहरणों के साथ बातचीत करने के लिए एक पायथन लाइब्रेरी है। प्रलेखन पृष्ठ में पुस्तकालय के साथ काम करने के लिए आवश्यक जानकारी है। जीथब प्रोजेक्ट में त्वरित शुरुआत का वर्णन किया गया है।

कॉन्फ़िगरेशन रोल आउट करने के लिए हमारी स्क्रिप्ट एक पायथन प्रोग्राम है। हम कोडिंग को पास करते हैं।
हम आगे के काम के लिए कॉन्फ़िगर कॉन्फ़िगर करते हैं। हमें निम्नलिखित मापदंडों की आवश्यकता होगी:

 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

सीएलआई ब्लॉक में, यह लिखा गया था कि दूरस्थ प्रक्रिया समूह में डेटा स्थानांतरण स्वचालित रूप से चालू नहीं होता है? स्क्रिप्ट को लागू करते समय, मैं इस समस्या में भी भाग गया। उस समय, मैं एपीआई का उपयोग करके डेटा ट्रांसफर शुरू नहीं कर सका और मैंने NiPyAPI लाइब्रेरी के डेवलपर को लिखने और सलाह / मदद के लिए कहने का फैसला किया। डेवलपर ने मुझे जवाब दिया, हमने समस्या पर चर्चा की और उन्होंने लिखा कि उन्हें "कुछ जाँचने" के लिए समय चाहिए। और अब, कुछ दिनों के बाद, एक पत्र आता है जिसमें एक पायथन फ़ंक्शन लिखा जाता है जो मेरी लॉन्च की समस्या को हल करता है !!! उस समय, NiPyAPI संस्करण 0.13.3 था और निश्चित रूप से, इसमें ऐसा कुछ नहीं था। लेकिन संस्करण 0.14.0 में, जिसे हाल ही में जारी किया गया था, इस फ़ंक्शन को पहले ही पुस्तकालय में शामिल किया गया है। मिलिए,

 nipyapi.canvas.set_remote_process_group_transmission

इसलिए, NiPyAPI लाइब्रेरी का उपयोग करते हुए, हमने रजिस्ट्री, लुढ़का हुआ प्रवाह, और यहां तक कि प्रोसेसर और डेटा ट्रांसफर भी शुरू किया। फिर आप कोड को कंघी कर सकते हैं, सभी प्रकार के चेक, लॉगिंग और यह सब जोड़ सकते हैं। लेकिन यह पूरी तरह से अलग कहानी है।

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

वह सब है। मैंने NiFi में फ्लो डिलीवरी को स्वचालित करने के लिए 3 दृष्टिकोणों का वर्णन किया, एक डेवलपर को नुकसान हो सकता है और डिलीवरी को स्वचालित करने के लिए एक कार्य कोड प्रदान किया जा सकता है। यदि आप इस विषय में रुचि रखते हैं - लिखो!

अपाचे NiFi प्रवाह वितरण स्वचालन

MiNiFi का उपयोग करना

निफी सीएलआई

NiPyAPI

More articles: