Apache NIFI - सरावातील संधींचे संक्षिप्त विहंगावलोकन

परिचय

असे घडले की माझ्या सध्याच्या कामाच्या ठिकाणी मला या तंत्रज्ञानाशी परिचित व्हावे लागले. मी थोड्या पार्श्वभूमीपासून सुरुवात करेन. पुढील मीटिंगमध्ये, आमच्या टीमला सांगण्यात आले की आम्हाला एकीकरण तयार करण्याची आवश्यकता आहे ज्ञात प्रणाली. एकत्रीकरणाचा अर्थ असा होता की ही सुप्रसिद्ध प्रणाली आम्हाला HTTP द्वारे विशिष्ट एंडपॉईंटवर विनंत्या पाठवेल आणि आम्ही, विचित्रपणे, SOAP संदेशाच्या रूपात प्रतिसाद परत पाठवू. सर्व काही साधे आणि क्षुल्लक वाटते. यावरून हे लक्षात येते की आपल्याला आवश्यक आहे ...

उद्दिष्ट

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

तांत्रिक खोलीकरण

समस्येचे निराकरण करण्याची योजना आखताना, आम्ही प्रथम स्प्रिंग फ्रेमवर्क, Nginx बॅलन्सर, पोस्टग्रेस डेटाबेस आणि इतर तांत्रिक गोष्टींचा वापर करून Java मध्ये ऍप्लिकेशन बनवण्याचा निर्णय घेतला आणि तांत्रिक गोष्टींचा वापर केला नाही. तांत्रिक उपाय विकसित करण्याच्या वेळेमुळे आम्हाला या समस्येचे निराकरण करण्यासाठी इतर पध्दतींचा विचार करण्याची परवानगी मिळाली, आमची नजर अपाचे एनआयएफआय तंत्रज्ञानाकडे गेली, जी विशिष्ट मंडळांमध्ये फॅशनेबल आहे. मी लगेच म्हणेन की या तंत्रज्ञानामुळे आम्हाला या 3 सेवा लक्षात आल्या. हा लेख फाईल वाहतूक सेवेच्या विकासाचे आणि ग्राहकांना डेटा ट्रान्सफर सेवेचे वर्णन करेल, परंतु लेख उपयुक्त असल्यास, मी डेटाबेसमधील डेटा अद्यतनित करण्याच्या सेवेबद्दल लिहीन.

हे काय आहे?

NIFI हे डेटाचे जलद समांतर लोडिंग आणि प्रक्रिया, स्त्रोत आणि परिवर्तनांसाठी मोठ्या संख्येने प्लगइन, कॉन्फिगरेशनची आवृत्ती आणि बरेच काही यासाठी वितरित आर्किटेक्चर आहे. एक चांगला बोनस म्हणजे ते वापरणे खूप सोपे आहे. क्षुल्लक प्रक्रिया जसे की getFile, sendHttpRequest आणि इतरांना स्क्वेअर म्हणून प्रस्तुत केले जाऊ शकते. प्रत्येक चौरस एक प्रक्रिया दर्शवितो, ज्याचा परस्परसंवाद खालील आकृतीमध्ये पाहिला जाऊ शकतो. प्रक्रिया सेटअप परस्परसंवादावर अधिक तपशीलवार दस्तऐवजीकरण लिहिले गेले आहे येथे , जे रशियन बोलतात त्यांच्यासाठी - येथे. दस्तऐवजीकरण NIFI कसे अनपॅक करावे आणि कसे चालवावे, तसेच प्रक्रिया कशी तयार करावी, ज्याला स्क्वेअर म्हणूनही ओळखले जाते याचे उत्तम वर्णन केले आहे.
लेख लिहिण्याची कल्पना दीर्घ शोधानंतर आणि प्राप्त झालेल्या माहितीची जाणीवपूर्वक रचना केल्यानंतर, तसेच भविष्यातील विकसकांसाठी जीवन थोडे सोपे बनवण्याच्या इच्छेनंतर जन्माला आला.

उदाहरण:

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

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

या प्रकरणात, विनंती ही प्रत्येक गोष्टीची सुरुवात आहे. चला त्याचे कॉन्फिगरेशन पॅरामीटर्स पाहू.

StandardHttpContextMap अपवाद वगळता येथे सर्व काही अगदी क्षुल्लक आहे - ही एक प्रकारची सेवा आहे जी तुम्हाला विनंत्या पाठवू आणि प्राप्त करू देते. अधिक तपशीलवार आणि अगदी उदाहरणांसह, आपण पाहू शकता - येथे

पुढे, स्क्वेअरचे ReplaceText कॉन्फिगरेशन पॅरामीटर्स पाहू. रिप्लेसमेंट व्हॅल्यूकडे लक्ष देणे योग्य आहे - हे असे आहे जे वापरकर्त्याला प्रतिसादाच्या रूपात परत केले जाईल. सेटिंग्जमध्ये तुम्ही लॉगिंगची पातळी समायोजित करू शकता, तुम्ही लॉग पाहू शकता {जिथे तुम्ही nifi-1.9.2/logs अनपॅक केले आहे, तेथे अपयश/यशाचे मापदंड देखील आहेत - या पॅरामीटर्सच्या आधारावर तुम्ही संपूर्ण प्रक्रियेचे नियमन करू शकता. . म्हणजेच, यशस्वी मजकूर प्रक्रियेच्या बाबतीत, वापरकर्त्याला प्रतिसाद पाठविण्याची प्रक्रिया कॉल केली जाईल आणि दुसर्या प्रकरणात आम्ही अयशस्वी प्रक्रिया लॉग करू.

जेव्हा प्रतिसाद यशस्वीरित्या तयार केला जातो तेव्हा स्थिती वगळता HandleHttpResponse गुणधर्मांमध्ये विशेष मनोरंजक काहीही नाही.

आम्ही विनंती आणि प्रतिसाद क्रमवारी लावला आहे - चला फाइल प्राप्त करण्याकडे आणि FTP सर्व्हरवर ठेवण्यासाठी पुढे जाऊया. FetchFile - सेटिंग्जमध्ये निर्दिष्ट केलेल्या मार्गावर एक फाइल प्राप्त करते आणि ती पुढील प्रक्रियेसाठी पास करते.

आणि नंतर PutSftp स्क्वेअर - फाइलला फाइल स्टोरेजमध्ये ठेवते. आम्ही खाली कॉन्फिगरेशन पॅरामीटर्स पाहू शकतो.

प्रत्येक स्क्वेअर ही एक स्वतंत्र प्रक्रिया आहे जी लाँच करणे आवश्यक आहे याकडे लक्ष देणे योग्य आहे. आम्ही सर्वात सोप्या उदाहरणाकडे पाहिले ज्यासाठी कोणत्याही जटिल सानुकूलनाची आवश्यकता नाही. पुढे, आम्ही प्रक्रिया थोडी अधिक क्लिष्ट पाहू, जिथे आम्ही खोबणीवर थोडे लिहू.

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

SOAP संदेश सुधारित करण्याच्या प्रक्रियेमुळे ग्राहकांना डेटा ट्रान्सफर सेवा थोडी अधिक क्लिष्ट झाली. सामान्य प्रक्रिया खालील आकृतीमध्ये दर्शविली आहे.

येथे कल्पना देखील विशेषतः क्लिष्ट नाही: आम्हाला ग्राहकांकडून एक विनंती प्राप्त झाली की त्याला डेटा आवश्यक आहे, त्याला एक संदेश मिळाला आहे असा प्रतिसाद पाठवला, प्रतिसाद फाइल प्राप्त करण्याची प्रक्रिया सुरू केली, नंतर विशिष्ट तर्काने ती संपादित केली आणि नंतर सर्व्हरवर SOAP संदेशाच्या रूपात फाईल ग्राहकांना हस्तांतरित केली.

मला वाटते की आपण वर पाहिलेल्या चौरसांचे पुन्हा वर्णन करण्याची गरज नाही - चला थेट नवीनकडे जाऊया. जर तुम्हाला कोणतीही फाईल संपादित करायची असेल आणि सामान्य रिप्लेस टेक्स्ट प्रकार योग्य नसतील, तर तुम्हाला तुमची स्वतःची स्क्रिप्ट लिहावी लागेल. हे ExecuteGroogyScript स्क्वेअर वापरून करता येते. त्याची सेटिंग्ज खाली सादर केली आहेत.

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

होस्ट आणि ॲक्शन (साबण ॲक्शन) सारखे अनेक गुणधर्म जोडा. आम्ही जतन करतो आणि तपासतो. SOAP विनंत्या कशा पाठवायच्या याबद्दल तुम्ही अधिक तपशील पाहू शकता येथे

आम्ही NIFI प्रक्रिया वापरण्यासाठी अनेक पर्याय पाहिले. ते कसे संवाद साधतात आणि त्यांचा खरा फायदा काय आहे? विचारात घेतलेली उदाहरणे चाचणीची आहेत आणि प्रत्यक्षात लढाईत काय होते त्यापेक्षा थोडी वेगळी आहेत. मला आशा आहे की हा लेख विकसकांसाठी थोडा उपयुक्त असेल. आपण लक्ष दिल्याबद्दल धन्यवाद. तुम्हाला काही प्रश्न असतील तर लिहा. मी उत्तर देण्याचा प्रयत्न करेन.

स्त्रोत: www.habr.com