Apache NIFI - Təcrübədə Xüsusiyyətlərə Qısa Baxış

Giriş

Elə oldu ki, indiki iş yerimdə bu texnologiya ilə tanış olmalı oldum. Bir az arxa planla başlayacağam. Növbəti görüşdə komandamızla inteqrasiya yaratmağımız lazım olduğu bildirildi məlum sistem. İnteqrasiya dedikdə o nəzərdə tutulurdu ki, bu tanınmış sistem HTTP vasitəsilə bizə müəyyən son nöqtəyə sorğular göndərəcək və biz də qəribə də olsa, cavabları SOAP mesajı şəklində geri göndərəcəyik. Hər şey sadə və mənasız görünür. Buradan belə çıxır ki, sizə lazımdır...

Tapşırıq

3 xidmət yaradın. Bunlardan birincisi Verilənlər Bazasının Yenilənməsi Xidmətidir. Bu xidmət üçüncü tərəf sistemindən yeni məlumatlar daxil olduqda verilənlər bazasındakı məlumatları yeniləyir və onu növbəti sistemə ötürmək üçün CSV formatında fayl yaradır. İkinci xidmətin son nöqtəsi deyilir - ötürülən faylı qəbul edən, onu təsdiqləyən və FTP vasitəsilə fayl anbarına yerləşdirən FTP Nəqliyyat Xidməti. Üçüncü xidmət, istehlakçı məlumatlarının ötürülməsi xidməti, ilk ikisi ilə asinxron işləyir. O, yuxarıda müzakirə olunan faylı almaq üçün üçüncü tərəfin xarici sistemindən sorğu alır, hazır cavab faylını götürür, onu dəyişdirir (id, təsvir, linkToFile sahələrini yeniləyir) və cavabı SOAP mesajı şəklində göndərir. Yəni, ümumi mənzərə belədir: ilk iki xidmət yalnız yeniləmə üçün məlumatlar gəldikdə işə başlayır. Üçüncü xidmət daim işləyir, çünki çoxlu məlumat istehlakçıları var, dəqiqədə məlumat üçün təxminən 1000 sorğu var. Xidmətlər daim mövcuddur və onların nümunələri sınaq, demo, pre-istehsal və məhsul kimi müxtəlif mühitlərdə yerləşir. Aşağıda bu xidmətlərin necə işlədiyini göstərən bir diaqram var. Dərhal aydınlaşdırım ki, lazımsız mürəkkəbliyin qarşısını almaq üçün bəzi detallar sadələşdirilib.

Texniki dərinləşdirmə

Problemin həllini planlaşdırarkən, ilk növbədə Spring framework, Nginx balanslaşdırıcı, Postgres verilənlər bazası və digər texniki və o qədər də texniki olmayan şeylərdən istifadə edərək Java-da tətbiqlər etmək qərarına gəldik. Texniki həllin işlənib hazırlanması vaxtı bizə bu problemin həllinə başqa yanaşmaları nəzərdən keçirməyə imkan verdiyi üçün nəzərimiz müəyyən dairələrdə dəbdə olan Apache NIFI texnologiyasına düşdü. Dərhal deyim ki, bu texnologiya bizə bu 3 xidməti görməyə imkan verdi. Bu məqalə fayl daşıma xidmətinin və istehlakçıya məlumat ötürmə xidmətinin inkişafını təsvir edəcək, lakin məqalə faydalı olarsa, verilənlər bazasında məlumatların yenilənməsi xidməti haqqında yazacağam.

Nədir?

NIFI, məlumatların sürətli paralel yüklənməsi və emalı, mənbələr və transformasiyalar üçün çoxlu sayda plaginlər, konfiqurasiyaların versiyalaşdırılması və daha çox şey üçün paylanmış bir arxitekturadır. Gözəl bonus ondan ibarətdir ki, istifadə etmək çox asandır. GetFile, sendHttpRequest və başqaları kimi əhəmiyyətsiz proseslər kvadrat şəklində təqdim edilə bilər. Hər kvadrat bir prosesi təmsil edir, onun qarşılıqlı təsirini aşağıdakı şəkildə görmək olar. Prosesin qurulması ilə qarşılıqlı əlaqə haqqında daha ətraflı sənədlər yazılmışdır burada , rus dilində danışanlar üçün - burada. Sənədlər NIFI-nin necə açılacağını və işlədilməsini, həmçinin kvadratlar kimi tanınan proseslərin necə yaradılacağını mükəmməl şəkildə təsvir edir.
Məqalə yazmaq ideyası uzun axtarışdan və alınan məlumatları şüurlu bir şeyə çevirdikdən, həmçinin gələcək tərtibatçılar üçün həyatı bir az asanlaşdırmaq istəyindən sonra yaranıb.

Misal

Kvadratların bir-biri ilə qarşılıqlı əlaqəsinə dair bir nümunə nəzərdən keçirilir. Ümumi sxem olduqca sadədir: Biz HTTP sorğusu alırıq (Nəzəri olaraq, sorğunun mətnində fayl ilə. NIFI-nin imkanlarını nümayiş etdirmək üçün bu nümunədə sorğu yerli fayl yaddaşından faylın qəbulu prosesinə başlayır. ), sonra sorğunun alındığına dair cavabı geri göndəririk, paralel olaraq FH-dən faylın qəbulu prosesi və sonra onun FTP vasitəsilə FH-yə köçürülməsi prosesi. Proseslərin flowFile adlanan vasitəsilə bir-biri ilə qarşılıqlı əlaqədə olduğunu aydınlaşdırmağa dəyər. Bu, NIFI-də atributları və məzmunu saxlayan əsas qurumdur. Məzmun axın faylı ilə təmsil olunan məlumatdır. Yəni, kobud desək, bir kvadratdan fayl qəbul edib digərinə köçürsəniz, məzmun sizin faylınız olacaq.

Göründüyü kimi, bu şəkil ümumi prosesi göstərir. HandleHttpRequest - sorğuları qəbul edir, ReplaceText - cavab orqanı yaradır, HandleHttpResponse - cavab göndərir. FetchFile - fayl anbarından fayl qəbul edir, onu kvadrat PutSftp-ə köçürür - bu faylı FTP-də, göstərilən ünvanda yerləşdirir. İndi bu proses haqqında daha çox.

Bu halda istək hər şeyin başlanğıcıdır. Onun konfiqurasiya parametrlərinə baxaq.

StandardHttpContextMap istisna olmaqla, burada hər şey olduqca mənasızdır - bu, sorğu göndərməyə və qəbul etməyə imkan verən bir növ xidmətdir. Daha ətraflı və hətta nümunələrlə görə bilərsiniz - burada

Sonra kvadratın ReplaceText konfiqurasiya parametrlərinə baxaq. ReplacementValue-a diqqət yetirməyə dəyər - bu, istifadəçiyə cavab şəklində qaytarılacaq. Parametrlərdə siz giriş səviyyəsini tənzimləyə bilərsiniz, qeydləri görə bilərsiniz {nifi-1.9.2/logs paketdən çıxartdığınız yerdə}/nifi-XNUMX/logs, həmçinin uğursuzluq/uğur parametrləri var - bu parametrlərə əsasən prosesi bütövlükdə tənzimləyə bilərsiniz. . Yəni, mətnin uğurlu işlənməsi halında, istifadəçiyə cavabın göndərilməsi prosesi çağırılacaq, başqa bir halda isə sadəcə olaraq uğursuz prosesi qeyd edəcəyik.

HandleHttpResponse xassələrində cavabın uğurla yaradıldığı statusdan başqa maraqlı heç nə yoxdur.

Sorğu və cavabı sıraladıq - faylın qəbuluna və FTP serverinə yerləşdirməyə keçək. FetchFile - parametrlərdə göstərilən yolda faylı qəbul edir və növbəti prosesə ötürür.

Və sonra PutSftp kvadratı - faylı fayl anbarına yerləşdirir. Aşağıdakı konfiqurasiya parametrlərini görə bilərik.

Hər kvadratın işə salınmalı olan ayrı bir proses olduğuna diqqət yetirməyə dəyər. Heç bir mürəkkəb fərdiləşdirmə tələb etməyən ən sadə nümunəyə baxdıq. Sonra, bir az daha mürəkkəb prosesə baxacağıq, burada yivlərə bir az yazacağıq.

Daha mürəkkəb nümunə

İstehlakçıya məlumat ötürmə xidməti SOAP mesajının dəyişdirilməsi prosesi səbəbindən bir az daha mürəkkəb oldu. Ümumi proses aşağıdakı şəkildə göstərilmişdir.

Burada ideya da o qədər də mürəkkəb deyil: biz istehlakçıdan ona məlumat lazım olduğu barədə sorğu aldıq, mesaj aldığına dair cavab göndərdik, cavab faylının qəbulu prosesinə başladıq, sonra onu müəyyən məntiqlə redaktə etdik və sonra faylı serverə SOAP mesajı şəklində istehlakçıya ötürdü.

Düşünürəm ki, yuxarıda gördüyümüz kvadratları yenidən təsvir etməyə ehtiyac yoxdur - gəlin birbaşa yenilərinə keçək. Hər hansı bir faylı redaktə etmək lazımdırsa və adi ReplaceText tipli kvadratlar uyğun deyilsə, öz skriptinizi yazmalı olacaqsınız. Bu ExecuteGroogyScript kvadratından istifadə etməklə edilə bilər. Onun parametrləri aşağıda təqdim olunur.

Skripti bu kvadrata yükləmək üçün iki seçim var. Birincisi, faylı skriptlə yükləməkdir. İkincisi, scriptBody-ə bir skript daxil etməklə. Bildiyimə görə, executeScript kvadratı bir neçə dili dəstəkləyir - onlardan biri gözəldir. Mən java tərtibatçılarını məyus edəcəm - belə kvadratlarda java-da skript yaza bilməzsiniz. Həqiqətən istəyənlər üçün öz fərdi kvadratınızı yaratmalı və onu NIFI sisteminə əlavə etməlisiniz. Bütün bu əməliyyat, bu məqalədə bəhs etməyəcəyimiz bir dəf ilə kifayət qədər uzun bir rəqslə müşayiət olunur. Mən gözəl dil seçdim. Aşağıda sadəcə olaraq SOAP mesajında ​​id-i tədricən yeniləyən test skriptidir. Qeyd etmək vacibdir. Siz faylı flowFile-dan götürüb yeniləyirsiniz, unutmayın ki, onu yenidən ora qoymaq lazımdır, yenilənir. Bütün kitabxanaların daxil olmadığını da qeyd etmək lazımdır. Ola bilər ki, siz hələ də liblərdən birini idxal etməlisiniz. Başqa bir mənfi cəhət odur ki, bu kvadratdakı skriptin düzəldilməsi olduqca çətindir. NIFI JVM-ə qoşulmaq və sazlama prosesinə başlamaq üçün bir yol var. Şəxsən mən yerli tətbiqi işə saldım və sessiyadan fayl almağı simulyasiya etdim. Mən də yerli olaraq ayıklama etdim. Skripti yükləyərkən ortaya çıxan səhvlər Google üçün olduqca asandır və NIFI-nin özü tərəfindən jurnala yazılır.

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)
}

Əslində, kvadratın fərdiləşdirilməsi burada başa çatır. Sonra, yenilənmiş fayl faylın serverə göndərilməsindən məsul olan kvadrata köçürülür. Aşağıda bu kvadrat üçün parametrlər verilmişdir.

Biz SOAP mesajının ötürülməsi üsulunu təsvir edirik. Harada yazırıq. Sonra bunun SOAP olduğunu göstərməlisiniz.

Host və fəaliyyət (soapAction) kimi bir neçə xüsusiyyət əlavə edin. Saxlayırıq və yoxlayırıq. Siz SOAP sorğularını necə göndərmək barədə ətraflı məlumatı görə bilərsiniz burada

NIFI proseslərindən istifadə üçün bir neçə varianta baxdıq. Onlar necə qarşılıqlı əlaqə qururlar və onların əsl faydası nədir? Baxılan nümunələr sınaq nümunələridir və əslində döyüşdə baş verənlərdən bir qədər fərqlidir. Ümid edirəm ki, bu məqalə tərtibatçılar üçün bir az faydalı olacaq. Diqqətinizə görə təşəkkürlər. Hər hansı bir sualınız varsa, yazın. Cavab verməyə çalışacağam.

Mənbə: www.habr.com