Apache NIFI - პრაქტიკაში შესაძლებლობების მოკლე მიმოხილვა

შესავალი

ისე მოხდა, რომ ჩემს ამჟამინდელ სამუშაო ადგილზე მომიწია ამ ტექნოლოგიის გაცნობა. დავიწყებ ცოტა ფონზე. მომდევნო შეხვედრაზე ჩვენს გუნდს უთხრეს, რომ გვჭირდებოდა ინტეგრაციის შექმნა ცნობილი სისტემა. ინტეგრაციაში იგულისხმებოდა, რომ ეს ცნობილი სისტემა გამოგვიგზავნიდა მოთხოვნებს HTTP-ის საშუალებით კონკრეტულ საბოლოო წერტილამდე, ჩვენ კი, უცნაურია, პასუხებს SOAP შეტყობინების სახით ვუგზავნიდით. ყველაფერი მარტივი და ტრივიალური ჩანს. აქედან გამომდინარეობს, რომ საჭიროა...

დავალება

შექმენით 3 სერვისი. პირველი მათგანი არის მონაცემთა ბაზის განახლების სერვისი. ეს სერვისი, როდესაც ახალი მონაცემები მოდის მესამე მხარის სისტემიდან, განაახლებს მონაცემთა ბაზაში არსებულ მონაცემებს და აგენერირებს ფაილს CSV ფორმატში, რათა გადაიტანოს იგი შემდეგ სისტემაში. მეორე სერვისის ბოლო წერტილი ეწოდება - FTP Transport Service, რომელიც იღებს გადაცემულ ფაილს, ამოწმებს მას და ათავსებს ფაილების შესანახად FTP-ის საშუალებით. მესამე სერვისი, მომხმარებლის მონაცემთა გადაცემის სერვისი, მუშაობს ასინქრონულად პირველ ორთან. ის იღებს მოთხოვნას მესამე მხარის გარე სისტემისგან ზემოთ განხილული ფაილის მისაღებად, იღებს მზა საპასუხო ფაილს, ცვლის მას (ახლებს id, აღწერა, linkToFile ველებს) და აგზავნის პასუხს SOAP შეტყობინების სახით. ანუ, საერთო სურათი ასეთია: პირველი ორი სერვისი იწყებს მუშაობას მხოლოდ მაშინ, როდესაც მოვიდა განახლების მონაცემები. მესამე სერვისი მუდმივად მუშაობს, რადგან ბევრია ინფორმაციის მომხმარებელი, დაახლოებით 1000 მოთხოვნა მონაცემზე წუთში. სერვისები მუდმივად ხელმისაწვდომია და მათი ინსტანციები განლაგებულია სხვადასხვა გარემოში, როგორიცაა ტესტი, დემო, წინასწარი წარმოება და პროდუქცია. ქვემოთ მოცემულია დიაგრამა, თუ როგორ მუშაობს ეს სერვისები. ნება მომეცით დაუყოვნებლივ განვმარტო, რომ ზოგიერთი დეტალი გამარტივდა ზედმეტი სირთულის თავიდან ასაცილებლად.

ტექნიკური გაღრმავება

პრობლემის გადაჭრის დაგეგმვისას, ჯერ გადავწყვიტეთ აპლიკაციების გაკეთება Java-ში Spring Framework-ის, Nginx-ის ბალანსერის, Postgres-ის მონაცემთა ბაზის და სხვა ტექნიკური და არც ისე ტექნიკური ნივთების გამოყენებით. ვინაიდან ტექნიკური გადაწყვეტის შემუშავების დრომ მოგვცა საშუალება განგვეხილა ამ პრობლემის გადაჭრის სხვა მიდგომები, ჩვენი მზერა დაეცა Apache NIFI ტექნოლოგიაზე, რომელიც მოდურია გარკვეულ წრეებში. მაშინვე ვიტყვი, რომ ამ ტექნოლოგიამ საშუალება მოგვცა შეგვენახა ეს 3 სერვისი. ეს სტატია აღწერს ფაილების ტრანსპორტირების სერვისის განვითარებას და მომხმარებლისთვის მონაცემთა გადაცემის სერვისს, მაგრამ თუ სტატია სასარგებლოა, დავწერ მონაცემთა ბაზაში მონაცემთა განახლების სერვისზე.

რა არის ეს?

NIFI არის განაწილებული არქიტექტურა მონაცემთა სწრაფი პარალელური ჩატვირთვისა და დამუშავებისთვის, დიდი რაოდენობით დანამატები წყაროებისა და ტრანსფორმაციისთვის, კონფიგურაციების ვერსიებისთვის და მრავალი სხვა. სასიამოვნო ბონუსი ის არის, რომ მისი გამოყენება ძალიან მარტივია. ტრივიალური პროცესები, როგორიცაა getFile, sendHttpRequest და სხვა, შეიძლება წარმოდგენილი იყოს კვადრატებად. თითოეული კვადრატი წარმოადგენს პროცესს, რომლის ურთიერთქმედება შეგიძლიათ იხილოთ ქვემოთ მოცემულ ფიგურაში. დაიწერა უფრო დეტალური დოკუმენტაცია პროცესის დაყენების ურთიერთქმედების შესახებ აქ მათთვის, ვინც რუსულად საუბრობს - აქ. დოკუმენტაცია შესანიშნავად აღწერს, თუ როგორ უნდა გახსნათ და გაუშვათ NIFI, ასევე როგორ შევქმნათ პროცესები, რომლებიც ასევე ცნობილია როგორც კვადრატები.
სტატიის დაწერის იდეა გაჩნდა ხანგრძლივი ძიების და მიღებული ინფორმაციის რაღაც ცნობიერად სტრუქტურირების შემდეგ, ისევე როგორც სურვილი, ცოტათი გაუადვილოს ცხოვრება მომავალ დეველოპერებს.

მაგალითი

განხილულია მაგალითი იმისა, თუ როგორ ურთიერთქმედებენ კვადრატები ერთმანეთთან. ზოგადი სქემა საკმაოდ მარტივია: ჩვენ ვიღებთ HTTP მოთხოვნას (თეორიულად, ფაილით მოთხოვნის სხეულში. NIFI-ს შესაძლებლობების დემონსტრირებისთვის, ამ მაგალითში მოთხოვნა იწყებს ფაილის მიღების პროცესს ადგილობრივი ფაილების საცავიდან. ), შემდეგ ვუგზავნით პასუხს, რომ მოთხოვნა მიღებულია, პარალელურად მიმდინარეობს ფაილის FH-დან მიღების პროცესი და შემდეგ FTP-ით FH-ში გადატანის პროცესი. გასარკვევია, რომ პროცესები ერთმანეთთან ურთიერთქმედებენ ე.წ. flowFile-ის საშუალებით. ეს არის NIFI-ის საბაზისო ერთეული, რომელიც ინახავს ატრიბუტებსა და შინაარსს. შინაარსი არის მონაცემები, რომლებიც წარმოდგენილია ნაკადის ფაილით. ანუ, უხეშად რომ ვთქვათ, თუ თქვენ მიიღებთ ფაილს ერთი კვადრატიდან და გადაიტანეთ მეორეში, კონტენტი იქნება თქვენი ფაილი.

როგორც ხედავთ, ეს სურათი გვიჩვენებს ზოგად პროცესს. HandleHttpRequest - იღებს მოთხოვნებს, ReplaceText - წარმოქმნის პასუხს, HandleHttpResponse - აგზავნის პასუხს. FetchFile - იღებს ფაილს ფაილის საცავიდან, გადასცემს მას კვადრატულ PutSftp - აყენებს ამ ფაილს FTP-ზე, მითითებულ მისამართზე. ახლა უფრო მეტი ამ პროცესის შესახებ.

ამ შემთხვევაში თხოვნა ყველაფრის დასაწყისია. მოდით შევხედოთ მისი კონფიგურაციის პარამეტრებს.

აქ ყველაფერი საკმაოდ ტრივიალურია, გარდა StandardHttpContextMap - ეს არის ერთგვარი სერვისი, რომელიც საშუალებას გაძლევთ გაგზავნოთ და მიიღოთ მოთხოვნები. უფრო დეტალურად და თუნდაც მაგალითებით, შეგიძლიათ ნახოთ - აქ

შემდეგი, მოდით შევხედოთ კვადრატის ReplaceText კონფიგურაციის პარამეტრებს. ღირს ReplacementValue-ზე ყურადღების მიქცევა - ეს არის ის, რაც უბრუნდება მომხმარებელს პასუხის სახით. პარამეტრებში შეგიძლიათ დაარეგულიროთ ანგარიშების დონე, შეგიძლიათ იხილოთ ჟურნალები {სადაც ამოალაგეთ nifi}/nifi-1.9.2/logs, ასევე არის წარუმატებლობის/წარმატების პარამეტრები - ამ პარამეტრებზე დაყრდნობით შეგიძლიათ პროცესი მთლიანად დაარეგულიროთ. . ანუ ტექსტის წარმატებული დამუშავების შემთხვევაში გამოიძახება მომხმარებლისთვის პასუხის გაგზავნის პროცესი, სხვა შემთხვევაში კი უბრალოდ ჩავწერთ წარუმატებელ პროცესს.

არაფერია განსაკუთრებით საინტერესო HandleHttpResponse თვისებებში, გარდა სტატუსისა, როდესაც პასუხი წარმატებით იქმნება.

ჩვენ დავახარისხეთ მოთხოვნა და პასუხი - გადავიდეთ ფაილის მიღებაზე და FTP სერვერზე განთავსებაზე. FetchFile - იღებს ფაილს პარამეტრებში მითითებულ გზაზე და გადასცემს შემდეგ პროცესს.

და შემდეგ PutSftp კვადრატი - ათავსებს ფაილს ფაილის საცავში. ჩვენ ვხედავთ კონფიგურაციის პარამეტრებს ქვემოთ.

ყურადღება უნდა მიაქციოთ იმ ფაქტს, რომ თითოეული კვადრატი ცალკე პროცესია, რომელიც უნდა დაიწყოს. ჩვენ გადავხედეთ უმარტივეს მაგალითს, რომელიც არ საჭიროებს რაიმე რთულ პერსონალიზაციას. შემდეგი, ჩვენ შევხედავთ პროცესს ცოტა უფრო რთულად, სადაც ცოტას დავწერთ ღარები.

უფრო რთული მაგალითი

მომხმარებლისთვის მონაცემთა გადაცემის სერვისი ცოტა უფრო რთული აღმოჩნდა SOAP შეტყობინების შეცვლის პროცესის გამო. ზოგადი პროცესი ნაჩვენებია ქვემოთ მოცემულ ფიგურაში.

აქ იდეა ასევე არ არის განსაკუთრებით რთული: მივიღეთ მომხმარებლისგან მოთხოვნა, რომ მას სჭირდებოდა მონაცემები, გავგზავნეთ პასუხი, რომ მიიღო შეტყობინება, დავიწყეთ საპასუხო ფაილის მიღების პროცესი, შემდეგ დავადაქტირეთ გარკვეული ლოგიკით და შემდეგ გადასცა ფაილი მომხმარებელს SOAP შეტყობინების სახით სერვერზე.

ვფიქრობ, საჭირო არ არის კიდევ ერთხელ აღვწერო ის კვადრატები, რომლებიც ზემოთ ვნახეთ - პირდაპირ ახალზე გადავიდეთ. თუ რაიმე ფაილის რედაქტირება გჭირდებათ და ჩვეულებრივი ReplaceText ტიპის კვადრატები არ არის შესაფერისი, მოგიწევთ საკუთარი სკრიპტის დაწერა. ეს შეიძლება გაკეთდეს ExecuteGroogyScript კვადრატის გამოყენებით. მისი პარამეტრები წარმოდგენილია ქვემოთ.

ამ კვადრატში სკრიპტის ჩატვირთვის ორი ვარიანტი არსებობს. პირველი არის ფაილის ჩამოტვირთვა სკრიპტით. მეორე არის სკრიპტის ჩასმა scriptBody-ში. რამდენადაც მე ვიცი, executeScript კვადრატი მხარს უჭერს რამდენიმე ენას - ერთი მათგანი არის groovy. მე გაუცრუებ იმედს java დეველოპერებს - ასეთ კვადრატებში java-ში სკრიპტების დაწერა არ შეიძლება. მათთვის, ვისაც ნამდვილად სურს, თქვენ უნდა შექმნათ თქვენი საკუთარი მოედანი და დაამატოთ ის NIFI სისტემაში. მთელ ამ ოპერაციას თან ახლავს საკმაოდ ხანგრძლივი ცეკვა ტამბურით, რომელსაც ამ სტატიაში არ შევეხებით. მე ავარჩიე გრილის ენა. ქვემოთ მოცემულია სატესტო სკრიპტი, რომელიც უბრალოდ თანდათანობით ანახლებს ID-ს SOAP შეტყობინებაში. მნიშვნელოვანია აღინიშნოს. თქვენ იღებთ ფაილს flowFile-დან და განაახლებთ, არ დაგავიწყდეთ, რომ უნდა დააბრუნოთ იქ, განახლებული. აღსანიშნავია ისიც, რომ ყველა ბიბლიოთეკა არ შედის. შეიძლება მოხდეს, რომ მაინც მოგიწიოთ ერთ-ერთი lib-ის იმპორტი. კიდევ ერთი მინუსი არის ის, რომ ამ კვადრატში სკრიპტის გამართვა საკმაოდ რთულია. არსებობს გზა 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