नमस्कार!

कार्य निम्नानुसार छ - त्यहाँ एक प्रवाह छ, माथिको चित्रमा प्रस्तुत गरिएको छ, जसलाई N सर्भरहरूमा रोल आउट गर्न आवश्यक छ। Apache NiFi। प्रवाह परीक्षण - एउटा फाइल उत्पन्न भई अर्को NiFi उदाहरणमा पठाइँदैछ। डाटा स्थानान्तरण NiFi साइट टु साइट प्रोटोकल प्रयोग गरेर हुन्छ।

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

मामिलाहरूमा जहाँ हामी S2S प्रयोग गरेर डाटा स्थानान्तरणको बारेमा कुरा गर्दैछौं, एउटा उदाहरणलाई ग्राहक भनिन्छ, दोस्रो सर्भर। ग्राहकले डाटा पठाउँछ, सर्भरले प्राप्त गर्छ। तिनीहरू बीच डाटा स्थानान्तरण कन्फिगर गर्न दुई तरिकाहरू:

1. पुस। ग्राहक उदाहरणबाट, डाटा रिमोट प्रोसेस ग्रुप (RPG) प्रयोग गरेर पठाइन्छ। सर्भर उदाहरणमा, डेटा इनपुट पोर्ट प्रयोग गरेर प्राप्त हुन्छ
2. पुल। सर्भरले RPG प्रयोग गरेर डाटा प्राप्त गर्दछ, क्लाइन्टले आउटपुट पोर्ट प्रयोग गरेर पठाउँछ।

रोल आउटको लागि प्रवाह Apache Registry मा भण्डारण गरिएको छ।

Apache NiFi रजिस्ट्री 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 उदाहरणहरूमा प्रवाह डेलिभर गर्ने प्रक्रिया स्वचालित हुनु आवश्यक छ। एक परिचित कार्य, हैन?

अर्को उपप्रोजेक्टले यो समस्या समाधान गर्न मद्दत गर्नेछ - MiNiFi C2 सर्भर। यो उत्पादन कन्फिगरेसन रोलआउट वास्तुकलामा केन्द्रीय बिन्दु बन्नको लागि हो। वातावरण कसरी कन्फिगर गर्ने - मा वर्णन गरिएको छ यो लेख समस्या समाधान गर्न Habré मा पर्याप्त जानकारी छ। MiNiFi, C2 सर्भरसँग संयोजनमा, स्वचालित रूपमा यसको कन्फिगरेसन अपडेट गर्दछ। यस दृष्टिकोणको एकमात्र कमजोरी यो हो कि तपाईंले C2 सर्भरमा टेम्प्लेटहरू सिर्जना गर्नुपर्नेछ; रजिस्ट्रीमा साधारण प्रतिबद्धता पर्याप्त छैन।

माथिको लेखमा वर्णन गरिएको विकल्प काम गरिरहेको छ र कार्यान्वयन गर्न गाह्रो छैन, तर हामीले निम्न बिर्सनु हुँदैन:

1. Minifi सँग nifi बाट सबै प्रोसेसरहरू छैनन्
2. Minifi प्रोसेसर संस्करणहरू 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 प्रयोग गर्न सक्नुहुन्छ। निम्न विधि प्रयोग गरौं, माथिका उदाहरणहरूबाट ID लिनुहोस् (हाम्रो अवस्थामा यो 7f522a13-016e-1000-e504-d5b15587f2f3 हो)। NiFi API विधिहरूको विवरण यहाँ.

शरीरमा तपाईले JSON पास गर्न आवश्यक छ, जस्तै:

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

मापदण्डहरू जुन यसलाई काम गर्न भर्न आवश्यक छ:
राज्य - डाटा स्थानान्तरण स्थिति। उपलब्ध: डाटा स्थानान्तरण सक्षम गर्न ट्रान्समिटिङ, असक्षम गर्न रोकियो
संस्करण - प्रोसेसर संस्करण

संस्करण सिर्जना गर्दा ० मा पूर्वनिर्धारित हुनेछ, तर यी प्यारामिटरहरू विधि प्रयोग गरेर प्राप्त गर्न सकिन्छ

bash स्क्रिप्टका प्रशंसकहरूका लागि, यो विधि उपयुक्त लाग्न सक्छ, तर यो मेरो लागि अलि गाह्रो छ - bash स्क्रिप्टहरू मेरो मनपर्ने होइनन्। अर्को विधि मेरो विचारमा थप रोचक र सुविधाजनक छ।

NiPyAPI

NiPyAPI NiFi उदाहरणहरूसँग अन्तरक्रिया गर्नको लागि पाइथन पुस्तकालय हो। कागजात पृष्ठ पुस्तकालय संग काम गर्न को लागी आवश्यक जानकारी समावेश गर्दछ। द्रुत सुरुवात मा वर्णन गरिएको छ ड्राफ्ट 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

अर्को, यो प्रक्रिया समूह पहिले नै थपिएको छ कि छैन भनेर बुझ्न महत्त्वपूर्ण छ। प्रक्रिया समूह निर्देशांक अनुसार राखिएको छ र एक को शीर्ष मा दोस्रो घटक superimposed एक स्थिति उत्पन्न हुन सक्छ। मैले जाँच गरें, यो हुन सक्छ :) सबै थपिएको प्रक्रिया समूहहरू प्राप्त गर्न हामीले विधि प्रयोग गर्छौं

nipyapi.canvas.list_all_process_groups

हामी थप खोज्न सक्छौं, उदाहरणका लागि, नामद्वारा।

म टेम्प्लेट अपडेट गर्ने प्रक्रियाको वर्णन गर्नेछैन, म मात्र भन्छु कि यदि टेम्प्लेटको नयाँ संस्करणमा प्रोसेसरहरू थपिएका छन् भने, लामहरूमा सन्देशहरूको उपस्थितिमा कुनै समस्या छैन। तर यदि प्रोसेसरहरू हटाइयो भने, त्यसपछि समस्याहरू उत्पन्न हुन सक्छ (निफिले तपाईंलाई प्रोसेसर हटाउन अनुमति दिँदैन यदि सन्देश लाम यसको अगाडि जम्मा भएको छ)। यदि तपाइँ मैले यो समस्या कसरी समाधान गरें भन्नेमा रुचि राख्नुहुन्छ भने, कृपया मलाई लेख्नुहोस् र हामी यस बिन्दुमा छलफल गर्नेछौं। लेखको अन्त्यमा सम्पर्कहरू। प्रक्रिया समूह थप्ने चरणमा जाऔं।

स्क्रिप्ट डिबग गर्दा, मैले एउटा विशेषता भेट्टाएँ कि प्रवाहको नवीनतम संस्करण सधैं तानिएको छैन, त्यसैले म पहिले यो संस्करण जाँच गर्न सिफारिस गर्छु:

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 सँग अन्तरक्रिया गर्नको लागि अझ लचिलो उपकरण हो। उदाहरणका लागि, सन्देश लामहरू अब प्रवाहमा खाली छन् कि छैनन् र प्रक्रिया समूह अद्यावधिक गर्न सकिन्छ कि भनेर निर्धारण गर्न।

यति नै। मैले NiFi मा प्रवाह वितरण स्वचालित गर्न 3 दृष्टिकोणहरू वर्णन गरें, विकासकर्ताले सामना गर्न सक्ने समस्याहरू, र स्वचालित डेलिभरीको लागि कार्य कोड प्रदान गरे। यदि तपाइँ यस विषयमा म जस्तै रुचि राख्नुहुन्छ भने - लेख्नुहोस्!

स्रोत: www.habr.com