अपाचे NIFI - अभ्यास में सुविधाओं का संक्षिप्त अवलोकन

परिचय

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

कार्य

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

तकनीकी गहरीकरण

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

यह क्या है

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

उदाहरण

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

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

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

यहाँ सब कुछ StandartHttpContextMap के अपवाद के साथ बहुत मामूली है - यह एक ऐसी सेवा है जो आपको अनुरोध भेजने और प्राप्त करने की अनुमति देती है। आप अधिक विवरण और यहां तक ​​कि उदाहरण देख सकते हैं

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

HandleHttpResponse संपत्तियों के पास सफल प्रतिक्रिया निर्माण के लिए स्थिति के अलावा कुछ खास नहीं है।

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

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

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

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

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

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

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

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

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

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