Apache NIFI - Краток преглед на карактеристиките во пракса

Вовед

Така се случи на моето сегашно место на работа да морам да се запознаам со оваа технологија. Ќе почнам со малку позадина. На следниот состанок, на нашиот тим му беше кажано дека треба да создадеме интеграција со познат систем. Под интеграција се подразбираше дека овој добро познат систем ќе ни испраќа барања преку HTTP до одредена крајна точка, а ние, чудно е доволно, ќе испраќаме одговори во форма на SOAP порака. Сè изгледа едноставно и тривијално. Од ова произлегува дека ви треба ...

Задача

Создадете 3 услуги. Првиот од нив е услугата за ажурирање на базата на податоци. Оваа услуга, кога пристигнуваат нови податоци од систем од трета страна, ги ажурира податоците во базата на податоци и генерира датотека во CSV формат за да ги пренесе на следниот систем. Крајната точка на втората услуга се нарекува - FTP Transport Service, која ја прима пренесената датотека, ја потврдува и ја става во складирање на датотеки преку FTP. Третата услуга, услугата за пренос на податоци од потрошувачите, работи асинхроно со првите две. Добива барање од надворешен систем од трета страна да ја прими датотеката дискутирана погоре, ја зема подготвената датотека за одговор, ја менува (ја ажурира полињата id, опис, linkToFile) и го испраќа одговорот во форма на SOAP порака. Односно, целокупната слика е следна: првите две сервиси ја започнуваат својата работа само кога ќе пристигнат податоците за ажурирање. Третата услуга работи постојано бидејќи има многу корисници на информации, околу 1000 барања за податоци во минута. Услугите се достапни постојано и нивните примероци се наоѓаат во различни средини, како што се тест, демо, претпродукција и продукција. Подолу е дијаграм за тоа како функционираат овие услуги. Веднаш да објаснам дека некои детали се поедноставени за да се избегне непотребна сложеност.

Техничко продлабочување

Кога планиравме решение за проблемот, најпрво решивме да правиме апликации во Java со помош на Spring frame, Nginx balancer, 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