🥇Apache NIFI - अभ्यासमा क्षमताहरूको संक्षिप्त सिंहावलोकन

परिचय

यो यस्तो भयो कि मेरो हालको कामको ठाउँमा मैले यो प्रविधिसँग परिचित हुनुपर्‍यो। म सानो पृष्ठभूमिबाट सुरु गर्छु। अर्को बैठकमा, हाम्रो टोलीलाई हामीसँग एकीकरण सिर्जना गर्न आवश्यक छ भनिएको थियो ज्ञात प्रणाली। एकीकरणको मतलब यो थियो कि यो प्रख्यात प्रणालीले HTTP मार्फत हामीलाई अनुरोधहरू एक विशिष्ट अन्त्य बिन्दुमा पठाउनेछ, र हामी, अनौठो रूपमा, SOAP सन्देशको रूपमा प्रतिक्रियाहरू फिर्ता पठाउनेछौं। सबै कुरा सरल र तुच्छ देखिन्छ। यसबाट यो पछ्याउँछ कि तपाईलाई आवश्यक छ ...

उद्देश्य

3 सेवाहरू सिर्जना गर्नुहोस्। ती मध्ये पहिलो डाटाबेस अपडेट सेवा हो। यो सेवा, जब तेस्रो-पक्ष प्रणालीबाट नयाँ डाटा आउँछ, डाटाबेसमा डाटा अपडेट गर्दछ र अर्को प्रणालीमा स्थानान्तरण गर्न CSV ढाँचामा फाइल उत्पन्न गर्दछ। दोस्रो सेवाको अन्तिम बिन्दु भनिन्छ - FTP यातायात सेवा, जसले स्थानान्तरण गरिएको फाइल प्राप्त गर्दछ, यसलाई प्रमाणित गर्दछ, र FTP मार्फत फाइल भण्डारणमा राख्छ। तेस्रो सेवा, उपभोक्ता डेटा स्थानान्तरण सेवा, पहिलो दुई संग एसिन्क्रोनस काम गर्दछ। यसले तेस्रो-पक्ष बाह्य प्रणालीबाट माथि छलफल गरिएको फाइल प्राप्त गर्न अनुरोध प्राप्त गर्दछ, तयार प्रतिक्रिया फाइल लिन्छ, यसलाई परिमार्जन गर्दछ (आईडी, विवरण, लिङ्कटोफाइल फिल्डहरू अद्यावधिक गर्दछ) र प्रतिक्रियालाई SOAP सन्देशको रूपमा पठाउँदछ। अर्थात्, समग्र तस्विर निम्नानुसार छ: पहिलो दुई सेवाहरूले आफ्नो काम तब सुरु गर्छन् जब अद्यावधिकको लागि डाटा आइपुगेको छ। तेस्रो सेवाले निरन्तर काम गर्दछ किनभने त्यहाँ धेरै जानकारीका उपभोक्ताहरू छन्, प्रति मिनेट डाटाको लागि लगभग 1000 अनुरोधहरू छन्। सेवाहरू निरन्तर उपलब्ध छन् र तिनीहरूका उदाहरणहरू परीक्षण, डेमो, पूर्व-उत्पादन र उत्पादन जस्ता विभिन्न वातावरणहरूमा अवस्थित छन्। तल यी सेवाहरूले कसरी काम गर्छन् भन्ने रेखाचित्र छ। मलाई तुरुन्तै स्पष्ट गरौं कि केहि विवरणहरू अनावश्यक जटिलताबाट बच्न सरलीकृत गरिएको छ।

प्राविधिक गहनता

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

यो के हो?

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

उदाहरण:

वर्गहरू कसरी एकअर्कासँग अन्तरक्रिया गर्छन् भन्ने उदाहरण मानिन्छ। सामान्य योजना एकदम सरल छ: हामीले HTTP अनुरोध प्राप्त गर्छौं (सिद्धान्तमा, अनुरोधको मुख्य भागमा फाइलको साथ। NIFI को क्षमताहरू प्रदर्शन गर्न, यस उदाहरणमा अनुरोधले स्थानीय फाइल भण्डारणबाट फाइल प्राप्त गर्ने प्रक्रिया सुरु गर्छ। ), त्यसपछि हामी FH बाट फाइल प्राप्त गर्ने प्रक्रिया र त्यसपछि FTP मार्फत FH मा सार्ने प्रक्रियाको समानान्तरमा अनुरोध प्राप्त भएको प्रतिक्रिया पठाउँछौं। यो स्पष्ट गर्न लायक छ कि प्रक्रियाहरू तथाकथित flowFile मार्फत एक अर्कासँग अन्तरक्रिया गर्दछ। यो NIFI मा आधार इकाई हो जसले विशेषताहरू र सामग्री भण्डार गर्दछ। सामग्री स्ट्रिम फाइल द्वारा प्रतिनिधित्व गरिएको डाटा हो। त्यो हो, लगभग बोल्दै, यदि तपाईंले एउटा वर्गबाट फाइल प्राप्त गर्नुभयो र अर्कोमा स्थानान्तरण गर्नुभयो भने, सामग्री तपाईंको फाइल हुनेछ।

तपाईं देख्न सक्नुहुन्छ, यो तस्वीर सामान्य प्रक्रिया देखाउँछ। HandleHttpRequest - अनुरोधहरू स्वीकार गर्दछ, ReplaceText - एक प्रतिक्रिया शरीर उत्पन्न गर्दछ, HandleHttpResponse - प्रतिक्रिया पठाउँछ। FetchFile - फाइल भण्डारणबाट फाइल प्राप्त गर्दछ, यसलाई वर्ग PutSftp मा स्थानान्तरण गर्दछ - यो फाइल FTP मा, निर्दिष्ट ठेगानामा राख्छ। अब यो प्रक्रिया बारे थप।

यस अवस्थामा, अनुरोध सबै कुराको सुरुवात हो। यसको कन्फिगरेसन प्यारामिटरहरू हेरौं।

यहाँ सबै कुरा मानकHttpContextMap को अपवादको साथ एकदम मामूली छ - यो एक प्रकारको सेवा हो जसले तपाईंलाई अनुरोधहरू पठाउन र प्राप्त गर्न अनुमति दिन्छ। थप विवरणमा र उदाहरणहरूको साथ पनि, तपाइँ देख्न सक्नुहुन्छ - यहाँ

अर्को, स्क्वायरको ReplaceText कन्फिगरेसन प्यारामिटरहरू हेरौं। यो ReplacementValue मा ध्यान दिन लायक छ - यो एक प्रतिक्रिया को रूप मा प्रयोगकर्ता लाई फिर्ता गरिने छ। सेटिङहरूमा तपाईंले लगिङको स्तर समायोजन गर्न सक्नुहुन्छ, तपाईंले लगहरू देख्न सक्नुहुन्छ {जहाँ तपाईंले अनप्याक गर्नुभयो nifi}/nifi-1.9.2/logs, त्यहाँ असफलता/सफलता प्यारामिटरहरू पनि छन् - यी प्यारामिटरहरूको आधारमा तपाईंले सम्पूर्ण प्रक्रियालाई नियमन गर्न सक्नुहुन्छ। । त्यो हो, सफल पाठ प्रशोधनको अवस्थामा, प्रयोगकर्तालाई प्रतिक्रिया पठाउने प्रक्रियालाई कल गरिनेछ, र अर्को केसमा हामी असफल प्रक्रियालाई मात्र लग गर्नेछौं।

त्यहाँ HandleHttpResponse गुणहरूमा कुनै प्रतिक्रिया सफलतापूर्वक सिर्जना हुँदा स्थिति बाहेक केही रोचक छैन।

हामीले अनुरोध र प्रतिक्रियालाई क्रमबद्ध गरेका छौं - फाईल प्राप्त गर्न र FTP सर्भरमा राख्नको लागि अगाडि बढौं। FetchFile - सेटिङहरूमा निर्दिष्ट मार्गमा फाइल प्राप्त गर्दछ र यसलाई अर्को प्रक्रियामा पास गर्दछ।

र त्यसपछि PutSftp वर्ग - फाइल भण्डारणमा फाइल राख्छ। हामी तल कन्फिगरेसन प्यारामिटरहरू देख्न सक्छौं।

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

थप जटिल उदाहरण

SOAP सन्देश परिमार्जन गर्ने प्रक्रियाको कारण उपभोक्तालाई डाटा स्थानान्तरण सेवा अलि बढी जटिल भयो। सामान्य प्रक्रिया तलको चित्रमा देखाइएको छ।

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

मलाई लाग्छ कि हामीले माथि देखेका ती वर्गहरूलाई फेरि वर्णन गर्न आवश्यक छैन - सीधा नयाँहरूमा जाऔं। यदि तपाइँ कुनै पनि फाइल सम्पादन गर्न आवश्यक छ र साधारण ReplaceText प्रकार वर्गहरू उपयुक्त छैनन् भने, तपाइँ तपाइँको आफ्नै लिपि लेख्नु पर्छ। यो ExecuteGroogyScript वर्ग प्रयोग गरेर गर्न सकिन्छ। यसको सेटिङहरू तल प्रस्तुत गरिएको छ।

यस वर्गमा लिपि लोड गर्नका लागि दुईवटा विकल्पहरू छन्। पहिलो स्क्रिप्टको साथ फाइल डाउनलोड गरेर हो। दोस्रो scriptBody मा स्क्रिप्ट सम्मिलित गरेर हो। जहाँसम्म मलाई थाहा छ, executeScript वर्गले धेरै भाषाहरूलाई समर्थन गर्दछ - ती मध्ये एउटा ग्रूवी हो। म जाभा विकासकर्ताहरूलाई निराश गर्नेछु - तपाइँ त्यस्ता वर्गहरूमा जाभामा लिपिहरू लेख्न सक्नुहुन्न। वास्तवमा चाहनेहरूका लागि, तपाईंले आफ्नै अनुकूलन वर्ग सिर्जना गर्न र NIFI प्रणालीमा थप्न आवश्यक छ। यो सम्पूर्ण अपरेशन टम्बोरिन संग धेरै लामो नृत्य संग छ, जुन हामी यस लेख मा सम्झौता गर्दैनौं। मैले ग्रोवी भाषा रोजें। तल एक परीक्षण स्क्रिप्ट हो जसले SOAP सन्देशमा आईडीलाई बढ्दो रूपमा अपडेट गर्दछ। यो नोट गर्न महत्त्वपूर्ण छ। तपाईंले flowFile बाट फाइल लिनुभयो र यसलाई अद्यावधिक गर्नुभयो, नबिर्सनुहोस् कि तपाईंले यसलाई त्यहाँ फिर्ता राख्न आवश्यक छ, अद्यावधिक गर्नुहोस्। यो पनि ध्यान दिन लायक छ कि सबै पुस्तकालयहरू समावेश छैनन्। यो हुन सक्छ कि तपाईंले अझै पनि libs मध्ये एक आयात गर्नुपर्छ। अर्को नकारात्मक पक्ष यो छ कि यस वर्गमा लिपि डिबग गर्न धेरै गाह्रो छ। NIFI JVM मा जडान गर्ने र डिबगिङ प्रक्रिया सुरु गर्ने तरिका छ। व्यक्तिगत रूपमा, मैले एक स्थानीय अनुप्रयोग सुरु गरें र सत्रबाट फाइल प्राप्त गर्ने सिमुलेटेड। मैले स्थानीय रूपमा डिबगिङ पनि गरें। स्क्रिप्ट लोड गर्दा देखिने त्रुटिहरू Google को लागि धेरै सजिलो छन् र लगमा NIFI आफैले लेखेका छन्।

import org.apache.commons.io.IOUtils
import groovy.xml.XmlUtil
import java.nio.charset.*
import groovy.xml.StreamingMarkupBuilder

def flowFile = session.get()
if (!flowFile) return
try {
    flowFile = session.write(flowFile, { inputStream, outputStream ->
        String result = IOUtils.toString(inputStream, "UTF-8");
        def recordIn = new XmlSlurper().parseText(result)
        def element = recordIn.depthFirst().find {
            it.name() == 'id'
        }

        def newId = Integer.parseInt(element.toString()) + 1
        def recordOut = new XmlSlurper().parseText(result)
        recordOut.Body.ClientMessage.RequestMessage.RequestContent.content.MessagePrimaryContent.ResponseBody.id = newId

        def res = new StreamingMarkupBuilder().bind { mkp.yield recordOut }.toString()
        outputStream.write(res.getBytes(StandardCharsets.UTF_8))
} as StreamCallback)
     session.transfer(flowFile, REL_SUCCESS)
}
catch(Exception e) {
    log.error("Error during processing of validate.groovy", e)
    session.transfer(flowFile, REL_FAILURE)
}

वास्तवमा, यहाँ वर्गको अनुकूलन समाप्त हुन्छ। अर्को, अद्यावधिक गरिएको फाइल वर्गमा स्थानान्तरण गरिएको छ, जुन फाइल सर्भरमा पठाउन जिम्मेवार छ। तल यस वर्गका लागि सेटिङहरू छन्।

हामीले सोप सन्देश पठाइने विधिको वर्णन गर्छौं। हामी कहाँ लेख्छौं। अर्को तपाईंले यो SOAP हो भनेर संकेत गर्न आवश्यक छ।

धेरै गुणहरू थप्नुहोस् जस्तै होस्ट र कार्य (soapAction)। हामी बचत र जाँच। तपाईंले SOAP अनुरोधहरू कसरी पठाउने भन्ने बारे थप विवरणहरू हेर्न सक्नुहुन्छ यहाँ

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

स्रोत: www.habr.com

Apache NIFI - अभ्यासमा सुविधाहरूको संक्षिप्त अवलोकन