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

परिचय

ऐसा हुआ कि मेरे वर्तमान कार्यस्थल पर मुझे इस तकनीक से परिचित होना पड़ा। मैं थोड़ी पृष्ठभूमि से शुरुआत करूंगा। अगली बैठक में, हमारी टीम को बताया गया कि हमें एकीकरण बनाने की आवश्यकता है ज्ञात प्रणाली. एकीकरण से तात्पर्य यह था कि यह प्रसिद्ध प्रणाली हमें HTTP के माध्यम से एक विशिष्ट समापन बिंदु पर अनुरोध भेजेगी, और हम, अजीब तरह से, SOAP संदेश के रूप में प्रतिक्रियाएँ भेजेंगे। सब कुछ सरल और तुच्छ लगता है. इससे यह पता चलता है कि आपको...

कार्य

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

तकनीकी गहनता

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

यह क्या है

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

उदाहरण

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

जैसा कि आप देख सकते हैं, यह चित्र सामान्य प्रक्रिया को दर्शाता है। HandleHttpRequest - अनुरोध स्वीकार करता है, ReplaceText - एक प्रतिक्रिया निकाय बनाता है, HandleHttpResponse - एक प्रतिक्रिया भेजता है। FetchFile - फ़ाइल संग्रहण से एक फ़ाइल प्राप्त करता है, इसे स्क्वायर पुटस्फ़्टप में स्थानांतरित करता है - इस फ़ाइल को निर्दिष्ट पते पर FTP पर रखता है। अब इस प्रक्रिया के बारे में और अधिक।

इस मामले में, अनुरोध हर चीज़ की शुरुआत है। आइए इसके कॉन्फ़िगरेशन पैरामीटर देखें।

मानकHttpContextMap के अपवाद के साथ यहां सब कुछ काफी तुच्छ है - यह एक प्रकार की सेवा है जो आपको अनुरोध भेजने और प्राप्त करने की अनुमति देती है। अधिक विस्तार से और उदाहरण सहित भी आप देख सकते हैं - यहां

इसके बाद, आइए वर्ग के रिप्लेसटेक्स्ट कॉन्फ़िगरेशन मापदंडों को देखें। रिप्लेसमेंटवैल्यू पर ध्यान देने योग्य है - यही वह है जो उपयोगकर्ता को प्रतिक्रिया के रूप में लौटाया जाएगा। सेटिंग्स में आप लॉगिंग के स्तर को समायोजित कर सकते हैं, आप लॉग देख सकते हैं {जहाँ आपने निफ़ी को अनपैक किया है}/निफ़ी-1.9.2/लॉग, विफलता/सफलता पैरामीटर भी हैं - इन मापदंडों के आधार पर आप पूरी प्रक्रिया को विनियमित कर सकते हैं . यानी, सफल टेक्स्ट प्रोसेसिंग के मामले में, उपयोगकर्ता को प्रतिक्रिया भेजने की प्रक्रिया को कॉल किया जाएगा, और दूसरे मामले में हम असफल प्रक्रिया को लॉग करेंगे।

जब प्रतिक्रिया सफलतापूर्वक बनाई जाती है तो स्थिति को छोड़कर HandleHttpResponse गुणों में कुछ भी विशेष रूप से दिलचस्प नहीं है।

हमने अनुरोध और प्रतिक्रिया को सुलझा लिया है - आइए फ़ाइल प्राप्त करने और उसे एफ़टीपी सर्वर पर रखने के लिए आगे बढ़ें। FetchFile - सेटिंग्स में निर्दिष्ट पथ पर एक फ़ाइल प्राप्त करता है और इसे अगली प्रक्रिया में भेजता है।

और फिर putSftp वर्ग - फ़ाइल को फ़ाइल संग्रहण में रखता है। हम नीचे कॉन्फ़िगरेशन पैरामीटर देख सकते हैं।

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

अधिक जटिल उदाहरण

SOAP संदेश को संशोधित करने की प्रक्रिया के कारण उपभोक्ता को डेटा स्थानांतरण सेवा थोड़ी अधिक जटिल हो गई। सामान्य प्रक्रिया नीचे दिए गए चित्र में दिखाई गई है।

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

मुझे लगता है कि उन वर्गों का फिर से वर्णन करने की कोई आवश्यकता नहीं है जो हमने ऊपर देखे हैं - चलिए सीधे नए पर चलते हैं। यदि आपको किसी फ़ाइल को संपादित करने की आवश्यकता है और साधारण ReplaceText प्रकार के वर्ग उपयुक्त नहीं हैं, तो आपको अपनी स्वयं की स्क्रिप्ट लिखनी होगी। यह ExecuteGrogyScript वर्ग का उपयोग करके किया जा सकता है। इसकी सेटिंग्स नीचे प्रस्तुत की गई हैं।

इस वर्ग में स्क्रिप्ट लोड करने के लिए दो विकल्प हैं। पहला है स्क्रिप्ट के साथ फ़ाइल डाउनलोड करना। दूसरा है स्क्रिप्टबॉडी में एक स्क्रिप्ट डालना। जहाँ तक मुझे पता है, एक्ज़ीक्यूटस्क्रिप्ट स्क्वायर कई भाषाओं का समर्थन करता है - उनमें से एक ग्रूवी है। मैं जावा डेवलपर्स को निराश करूंगा - आप जावा में ऐसे वर्गों में स्क्रिप्ट नहीं लिख सकते। जो लोग वास्तव में ऐसा करना चाहते हैं, उनके लिए आपको अपना स्वयं का कस्टम वर्ग बनाना होगा और इसे NIFI सिस्टम में जोड़ना होगा। यह पूरा ऑपरेशन डफ के साथ काफी लंबे नृत्य के साथ होता है, जिस पर हम इस लेख में चर्चा नहीं करेंगे। मैंने ग्रूवी भाषा चुनी. नीचे एक परीक्षण स्क्रिप्ट है जो SOAP संदेश में आईडी को धीरे-धीरे अपडेट करती है। यह बात ध्यान देने योग्य है। आप फ़ाइल को फ़्लोफ़ाइल से लें और उसे अपडेट करें, यह न भूलें कि आपको इसे वापस वहीं रखना होगा, अपडेट किया हुआ। यह भी ध्यान देने योग्य है कि सभी पुस्तकालय शामिल नहीं हैं। ऐसा हो सकता है कि आपको अभी भी किसी एक 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 संदेश प्रसारित किया जाएगा। हम कहाँ लिखते हैं. आगे आपको यह बताना होगा कि यह SOAP है।

होस्ट और एक्शन (soapAction) जैसे कई गुण जोड़ें। हम सहेजते हैं और जांचते हैं। आप SOAP अनुरोध भेजने के तरीके के बारे में अधिक विवरण देख सकते हैं यहां

हमने NIFI प्रक्रियाओं का उपयोग करने के लिए कई विकल्पों पर गौर किया। वे कैसे बातचीत करते हैं और उनका वास्तविक लाभ क्या है? जिन उदाहरणों पर विचार किया गया है वे परीक्षण वाले हैं और वास्तव में युद्ध में जो होता है उससे थोड़े अलग हैं। मुझे उम्मीद है कि यह लेख डेवलपर्स के लिए थोड़ा उपयोगी होगा। आपके ध्यान देने के लिए धन्यवाद! यदि आपके कोई प्रश्न हों तो लिखें. मैं उत्तर देने का प्रयास करूंगा.

स्रोत: www.habr.com