🥇Apache NIFI - Кароткі агляд магчымасцяў на практыцы

Увядзенне

Так атрымалася, што на маім бягучым месцы працы мне прыйшлося пазнаёміцца з дадзенай тэхналогіяй. Пачну з невялікай перадгісторыі. На чарговым мітынгу, нашай камандзе сказалі, што трэба стварыць інтэграцыю з вядомай сістэмай. Пад інтэграцыяй мелася на ўвазе, што гэтая вядомая сістэма будзе нам слаць запыты праз HTTP на пэўны эндпоінт, а мы, як гэта ні дзіўна, слаць назад адказы ў выглядзе SOAP паведамлення. Накшталт усё проста і трывіяльна. З гэтага вынікае што трэба…

Задача

Стварыць 3 сэрвісы. Першы з іх - Сэрвіс абнаўлення БД. Гэты сэрвіс, пры паступленні новых дадзеных з іншай сістэмы, абнаўляе дадзеныя ў базе дадзеных і генеруе нейкі файл у фармаце CSV, для перадачы яго ў наступную сістэму. Выклікаецца эндпоінт другога сэрвісу - Сэрвісу транспарціроўкі праз FTP, які атрымлівае перададзены файл, валідуе яго, і кладзе ў файлавае сховішча праз FTP. Трэці сэрвіс - Сэрвіс перадачы дадзеных спажыўцу, працуе асінхронна з першымі двума. Ён прымае запыт ад іншай знешняй сістэмы, на атрыманне файла аб якім ішла гаворка вышэй, бярэ гатовы файл адказу, мадыфікуе яго (абнаўляе палі id, description, linkToFile) і пасылае адказ у выглядзе SOAP паведамлення. Т е ў цэлым карціна наступная: першыя два сэрвісу пачынаюць сваю працу толькі тады, калі дашлі дадзеныя для абнаўлення. Трэці сэрвіс працуе ўвесь час паколькі спажыўцоў інфармацыі шмат, каля 1000 запытаў на атрыманне дадзеных у хвіліну. Сэрвісы даступныя ўвесь час і іх інстанцы размяшчаюцца на розных асяродках, такіх як тэст, дэма, препрод і прод. Ніжэй прадстаўлена схема працы гэтых сервісаў. Адразу растлумачу, што некаторыя дэталі спрошчаны для пазбягання лішняй складанасці.

Тэхнічнае паглыбленне

Пры планаванні рашэння задачы, спачатку вырашылі зрабіць прыкладанні на java з выкарыстаннем Spring framework, балансавальнікам Nginx, базай дадзеных Postgres і іншымі тэхнічнымі і не вельмі штукамі. Паколькі час на прапрацоўку тэхнічнага рашэння дазваляла разгледзець іншыя падыходы рашэння гэтай задачы, погляд упаў на модную ў пэўных колах тэхналогію Apache NIFI. Адразу скажу, што гэтая тэхналогія дазволіла заўважыць нам гэтыя 3 сэрвісы. У гэтым артыкуле будзе апісана распрацоўка сэрвісу транспарціроўкі файла і сэрвісу перадачы даных спажыўцу, аднак калі артыкул зойдзе, напішу пра сэрвіс абнаўлення даных у БД.

Што гэта такое

NIFI уяўляе сабой размеркаваную архітэктуру для хуткай паралельнай загрузкі і апрацоўкі дадзеных, вялікая колькасць плагінаў для крыніц і пераўтварэнняў, версіяванне канфігурацый і многае іншае. Прыемным бонусам з'яўляецца тое, што ён вельмі просты ў выкарыстанні. Трывіяльныя працэсы, такія як getFile, sendHttpRequest і іншыя - можна прадставіць у выглядзе квадратаў. Кожны квадрат уяўляе нейкі працэс, узаемадзеянне якога можна ўбачыць на малюнку ніжэй. Больш падрабязная дакументацыя па ўзаемадзеянні наладзе працэсаў напісана тут , для тых каму на рускай тут. У дакументацыі выдатна распісана як распакаваць і запусціць NIFI, а гэтак жа, як стварыць працэсы, яны ж квадраты.
Ідэя напісаць артыкул нарадзілася пасля працяглых пошукаў і структуравання атрыманай інфармацыі ў нешта ўсвядомленае, а гэтак жа жаданне крыху аблегчыць жыццё будучым распрацоўшчыкамі.

Прыклад

Разгледжаны прыклад таго, як узаемадзейнічаюць квадраты паміж сабой. Агульная схема даволі простая: Атрымліваем HTTP запыт (У тэорыі з файлам у целе запыту. Для дэманстрацыі магчымасцяў NIFI, у дадзеным прыкладзе запыт стартуе працэс атрымання файла з лакальнага ФХ), далей адсылаем зваротна, што запыт атрыманы, раўналежна запускаецца працэс атрымання файла з ФХ і далей працэс перамяшчэнне яго праз FTP у ФХ. Варта патлумачыць, што працэсы ўзаемадзейнічаюць паміж сабой з дапамогай так званага flowFile. Гэта базавая сутнасць у NIFI, якая захоўвае ў сабе атрыбуты і змесціва. Змесціва - дадзеныя якія прадстаўлены файлам патоку. Т я груба кажучы, калі вы атрымалі файл з аднаго квадрата і перадаеце яго ў іншы, кантэнтам будзе ваш файл.

Як вы можаце заўважыць - на гэтым малюнку намаляваны агульны працэс. HandleHttpRequest - прымае запыты, ReplaceText - генеруе цела адказу, HandleHttpResponse - аддае адказ. FetchFile - атрымлівае файл з файлавага сховішча перадае яго квадрату PutSftp - кладзе гэты файл на FTP, па паказаным адрасе. Цяпер падрабязней аб гэтым працэсе.

У дадзеным выпадку – request усяму пачатак. Паглядзім яго параметры канфігурацыі.

Тут усё даволі трывіяльна за выключэннем StandartHttpContextMap - гэта нейкі сэрвіс які які дазваляе пасылаць і прымаць запыты. Больш падрабязна і нават з прыкладамі можна паглядзець - тут

Далей паглядзім параметры канфігурацыі ReplaceText квадрата. У ёй варта звярнуць увагу на ReplacementValue - гэта тое, што вернецца карыстачу ў выглядзе адказу. У settings можна рэгуляваць узровень лагіравання, логі можна паглядзець {куды распакавалі nifi}/nifi-1.9.2/logs там жа ёсць параметры failure/success - грунтуючыся на гэтыя параметры можна рэгуляваць працэс у цэлым. Т е ў выпадку паспяховай апрацоўкі тэксту - выклічацца працэс адпраўкі адказу карыстачу, а ў іншым выпадку мы проста залагуем няўдалы працэс.

Ва ўласцівасцях HandleHttpResponse асабліва нічога цікавага няма акрамя статуту пры паспяховым стварэнні адказу.

З запытам адказам разабраліся - пяройдзем далей да атрымання файла і памяшканнем яго на FTP сервер. FetchFile - атрымлівае файл па паказаным у наладах шляху і перадае яго ў наступны працэс.

І далей квадрат PutSftp – змяшчае файл у файлавае сховішча. Параметры канфігурацыі можам убачыць ніжэй.

Варта звярнуць увагу на тое, што кожны квадрат - гэта асобны працэс, які павінен быць запушчаны. Мы разгледзелі самы просты прыклад які не патрабуе якой-небудзь складанай кастамізіцыі. Далей разгледзім працэс крыху больш складана, дзе крыху папішам на грувях.

Больш складаны прыклад

Сэрвіс перадачы дадзеных спажыўцу атрымаўся крыху больш складана за кошт працэсу мадыфікацыі SOAP паведамлення. Агульны працэс прадстаўлены на малюнку ніжэй.

Тут ідэя таксама не асоба складаная: атрымалі запыт ад спажыўца, што яму патрэбныя дадзеныя, адправілі адказ, што атрымалі паведамленне, запусцілі працэс атрымання файла адказу, далей адрэдагавалі яго з вызначанай логікай, пасля чаго перадалі файл спажыўцу ў выглядзе SOAP паведамлення на сервер.

Думаю не варта апісваць нанова тыя квадраты, якія мы бачылі вышэй - пяройдзем адразу да новых. Калі вам неабходна рэдагаваць які-небудзь файл і звычайныя квадраты тыпу ReplaceText не падыходзяць, вам давядзецца пісаць свой скрыпт. Зрабіць гэта можна з дапамогай квадрата ExecuteGroogyScript. Наладкі яго прадстаўлены ніжэй.

Ёсць два варыянты загрузкі скрыпту ў гэты квадрат. Першы - гэта шляхам загрузкі файла са скрыптам. Другі - устаўкай скрыпту ў scriptBody. На колькі я ведаю, квадрат executeScript падтрымлівае некалькі ЯП - адзін з іх groovy. Расчарую java распрацоўнікаў - на java нельга пісаць скрыпты ў такіх квадратах. Для тых каму вельмі хочацца - трэба стварыць свой кастамны квадрат і падкінуць яго ў сістэму NIFI. Уся гэтая аперацыя суправаджаецца даволі працяглымі танцамі з бубнам, якімі мы не будзем у рамках гэтага артыкула займацца. Я выбрала мову groovy. Ніжэй прадстаўлены тэставы скрыпт які проста інкрыментна абнаўляе id у SOAP паведамленні. Важна адзначыць. Вы бераце файл з flowFile абнаўляеце яго, не варта забываць, што яго трэба, абноўлены, зваротна туды пакласці. Таксама варта адзначыць, што не ўсе бібліятэкі падключаны. Можа атрымацца так, што вам усёткі давядзецца імпартаваць адну з либ. Мінусам яшчэ з'яўляецца тое, што скрыпт у дадзеным квадраце даволі цяжка дэбажыць. Ёсць спосаб падлучыцца да JVM NIFI і пачаць працэс адладкі. Асабіста я запускала ў сябе лакальнае прыкладанне і імітавала атрыманне файла з сесіі. Адладкай таксама займалася лакальна. Памылкі, якія вылазяць пры загрузцы скрыпту даволі лёгка куляцца і пішуцца самім 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.

Дадаем некалькі ўласцівасцяў такіх як хост і дзеянне(soapAction). Захоўваем, правяраем. Больш падрабязна як дасылаць SOAP запыты можна паглядзець тут

Мы разгледзелі некалькі варыянтаў выкарыстання працэсаў NIFI. Як яны ўзаемадзейнічаюць і якая ад іх рэальная карысць. Разгледжаныя прыклады з'яўляюцца тэставымі і крыху адрозніваюцца ад таго, што рэальна на баі. Спадзяюся, гэты артыкул будзе крыху карыснай для распрацоўшчыкаў. Дзякуй за ўвагу. Калі ёсць якія-небудзь пытанні - пішыце. Пастараюся адказаць.

Крыніца: habr.com

Apache NIFI - Кароткі агляд магчымасцяў на практыцы