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

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

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

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

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

खींच. सर्वर आरपीजी का उपयोग करके डेटा प्राप्त करता है, क्लाइंट आउटपुट पोर्ट का उपयोग करके भेजता है।

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

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

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

लेकिन जैसे-जैसे एन बढ़ता है, और अधिक समस्याएं होती हैं:

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

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

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

मिनीफाई का उपयोग करना

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

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

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

मिनीफ़ी में निफ़ी के सभी प्रोसेसर नहीं हैं
Minifi में CPU संस्करण NiFi में CPU संस्करणों से पीछे हैं।

लेखन के समय, 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 रजिस्ट्री वेब इंटरफ़ेस में प्राप्त किया जा सकता है। वेब इंटरफ़ेस इस तरह दिखता है:

सीएलआई का उपयोग करके, आप यह करते हैं:

#> 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 pg-start -pgid 7f522a13-016e-1000-e504-d5b15587f2f3 -u http://nifi:8080

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

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

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

मुख्य भाग में, आपको निम्नलिखित फ़ॉर्म का 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, которое подтягиваем

आगे मैं इस लाइब्रेरी के तरीकों के नाम डालूँगा, जिनका वर्णन किया गया है यहां.

हम रजिस्ट्री को निफ़ी इंस्टेंस का उपयोग करके कनेक्ट करते हैं

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

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

nipyapi.canvas.set_remote_process_group_transmission

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

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

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

स्रोत: www.habr.com

Apache NiFi में फ्लो डिलिवरी ऑटोमेशन

मिनीफाई का उपयोग करना

निफ़ी सीएलआई

NiPyAPI

एक टिप्पणी जोड़ें उत्तर रद्द