Apache NIFI - Тәжірибедегі мүмкіндіктерге қысқаша шолу

Кіріспе

Қазіргі жұмыс орнымда осы технологиямен танысуға тура келді. Мен кішкене фоннан бастайын. Келесі кездесуде біздің командаға интеграция құру керектігі айтылды белгілі жүйе. Интеграция арқылы бұл белгілі жүйе бізге HTTP арқылы белгілі бір соңғы нүктеге сұраулар жіберетінін білдіреді және біз, таңқаларлық, жауаптарды SOAP хабарламасы түрінде жіберетін едік. Барлығы қарапайым және тривиальды болып көрінеді. Осыдан келіп шығады...

Мақсаты

3 қызметті құру. Олардың біріншісі – Мәліметтер базасын жаңарту қызметі. Бұл қызмет үшінші тарап жүйесінен жаңа деректер келгенде, дерекқордағы деректерді жаңартады және келесі жүйеге тасымалдау үшін CSV пішіміндегі файлды жасайды. Екінші қызметтің соңғы нүктесі - FTP тасымалдау қызметі деп аталады, ол тасымалданатын файлды қабылдайды, оны тексереді және FTP арқылы файлдарды сақтауға қояды. Үшінші қызмет, тұтынушы деректерін тасымалдау қызметі, алғашқы екеуімен асинхронды түрде жұмыс істейді. Ол жоғарыда қарастырылған файлды алу үшін үшінші тараптың сыртқы жүйесінен сұрауды алады, дайын жауап файлын алады, оны өзгертеді (id, сипаттама, linkToFile өрістерін жаңартады) және жауапты SOAP хабарламасы түрінде жібереді. Яғни, жалпы сурет келесідей: алғашқы екі қызмет жаңартуға арналған деректер келгенде ғана жұмысын бастайды. Үшінші сервис үздіксіз жұмыс істейді, себебі ақпаратты тұтынушылар көп, минутына деректерге шамамен 1000 сұраныс. Қызметтер үнемі қол жетімді және олардың даналары сынақ, демонстрация, алдын ала өндіру және өнім сияқты әртүрлі орталарда орналасқан. Төменде осы қызметтердің жұмыс істеу схемасы берілген. Бірден түсіндіруге рұқсат етіңіздер, кейбір мәліметтер қажетсіз күрделілікті болдырмау үшін жеңілдетілген.

Техникалық тереңдету

Мәселені шешуді жоспарлаған кезде, біз алдымен Spring Framework, Nginx балансизаторы, Postgres дерекқоры және басқа да техникалық және техникалық емес нәрселерді пайдалана отырып, Java тілінде қосымшалар жасауды шештік. Техникалық шешімді әзірлеу уақыты осы мәселені шешудің басқа тәсілдерін қарастыруға мүмкіндік бергендіктен, біздің көзқарасымыз белгілі бір ортада сәнге айналған Apache NIFI технологиясына түсті. Мен бірден айтамын, бұл технология бізге осы 3 қызметті байқауға мүмкіндік берді. Бұл мақалада тұтынушыға файлдарды тасымалдау қызметі мен деректерді беру қызметінің дамуы сипатталады, бірақ мақала пайдалы болса, мен дерекқордағы деректерді жаңарту қызметі туралы жазамын.

Бұл не

NIFI - деректерді жылдам параллель жүктеуге және өңдеуге, дереккөздер мен түрлендірулерге арналған плагиндердің үлкен саны, конфигурацияларды нұсқалау және т.б. үшін бөлінген архитектура. Жақсы бонус - оны пайдалану өте оңай. GetFile, sendHttpRequest және басқалары сияқты тривиальды процестерді шаршы түрінде көрсетуге болады. Әрбір шаршы процесті білдіреді, оның әрекеттесуін төмендегі суреттен көруге болады. Процесті орнатудың өзара әрекеттесуі туралы толығырақ құжаттама жазылды осында , орысша сөйлейтіндер үшін - осында. Құжаттама NIFI-ны қалай ашу және іске қосу керектігін, сондай-ақ квадраттар деп аталатын процестерді қалай жасау керектігін тамаша сипаттайды.
Мақала жазу идеясы ұзақ іздеуден және алынған ақпаратты саналы нәрсеге айналдырудан кейін, сондай-ақ болашақ әзірлеушілер үшін өмірді сәл жеңілдетуге ұмтылудан кейін туды.

Мысал:

Квадраттардың бір-бірімен әрекеттесуінің мысалы қарастырылады. Жалпы схема өте қарапайым: Біз HTTP сұрауын аламыз (Теорияда сұраудың негізгі бөлігінде файлы бар. NIFI мүмкіндіктерін көрсету үшін бұл мысалда сұрау жергілікті файлдар қоймасынан файлды қабылдау процесін бастайды. ), содан кейін біз сұраудың алынғаны туралы жауапты кері жібереміз, сонымен қатар FH-дан файлды алу процесі, содан кейін оны FTP арқылы FH-ге жылжыту процесі. Процесстер flowFile деп аталатын файл арқылы бір-бірімен өзара әрекеттесетінін түсінген жөн. Бұл атрибуттар мен мазмұнды сақтайтын NIFI негізгі нысаны. Мазмұн – ағындық файл арқылы көрсетілетін деректер. Яғни, шамамен айтқанда, егер сіз бір шаршыдан файлды қабылдап, оны екіншісіне тасымалдасаңыз, мазмұн сіздің файлыңыз болады.

Көріп отырғаныңыздай, бұл сурет жалпы процесті көрсетеді. HandleHttpRequest – сұрауларды қабылдайды, ReplaceText – жауап мәтінін жасайды, HandleHttpResponse – жауапты жібереді. FetchFile - файлды сақтау қоймасынан файлды қабылдайды, PutSftp шаршысына ауыстырады - бұл файлды FTP-ге, көрсетілген мекенжайға қояды. Енді бұл процесс туралы толығырақ.

Бұл жағдайда сұрау – бәрінің басы. Оның конфигурация параметрлерін қарастырайық.

StandardHttpContextMap қолданбасын қоспағанда, мұнда барлығы өте тривиальды - бұл сұрауларды жіберуге және алуға мүмкіндік беретін қызмет түрі. Толығырақ және тіпті мысалдармен сіз көре аласыз - осында

Әрі қарай шаршының ReplaceText конфигурация параметрлерін қарастырайық. ReplacementValue-ге назар аударған жөн - бұл пайдаланушыға жауап түрінде қайтарылатын нәрсе. Параметрлерде журнал жүргізу деңгейін реттей аласыз, журналдарды {nifi-1.9.2/logs қаптамасынан шығарған жерде көре аласыз, сонымен қатар сәтсіздік/сәттілік параметрлері бар - осы параметрлер негізінде процесті тұтастай реттеуге болады. . Яғни, мәтінді өңдеу сәтті болған жағдайда, пайдаланушыға жауап жіберу процесі шақырылады, ал басқа жағдайда біз сәтсіз процесті тіркеуге аламыз.

HandleHttpResponse сипаттарында жауап сәтті жасалған күйден басқа ерекше қызықты ештеңе жоқ.

Біз сұрауды және жауапты сұрыптадық - файлды қабылдауға және оны FTP серверіне орналастыруға көшейік. FetchFile – параметрлерде көрсетілген жол бойынша файлды қабылдап, келесі процеске береді.

Содан кейін PutSftp квадраты - файлды файлдар қоймасына орналастырады. Төменде конфигурация параметрлерін көре аламыз.

Әрбір шаршының іске қосылуы керек жеке процесс екеніне назар аударған жөн. Біз күрделі теңшеуді қажет етпейтін ең қарапайым мысалды қарастырдық. Әрі қарай, біз процесті біршама күрделірек қарастырамыз, онда біз ойықтарға аздап жазамыз.

Неғұрлым күрделі мысал

Тұтынушыға деректерді беру қызметі SOAP хабарламасын өзгерту процесіне байланысты біршама күрделірек болды. Жалпы процесс төмендегі суретте көрсетілген.

Мұнда идея да аса күрделі емес: біз тұтынушыдан деректер қажет екендігі туралы сұрау алдық, хабарлама алғаны туралы жауап жібердік, жауап файлын алу процесін бастады, содан кейін оны белгілі бір логикамен өңдедік, содан кейін файлды тұтынушыға SOAP хабарламасы түрінде серверге жіберді.

Менің ойымша, біз жоғарыда көрген шаршыларды қайта сипаттаудың қажеті жоқ - тікелей жаңаларына көшейік. Кез келген файлды өңдеу қажет болса және қарапайым ReplaceText түріндегі квадраттар жарамсыз болса, өзіңіздің сценарийіңізді жазуыңыз керек. Мұны ExecuteGroogyScript шаршысының көмегімен жасауға болады. Оның параметрлері төменде берілген.

Бұл шаршыға сценарийді жүктеудің екі нұсқасы бар. Біріншісі - сценарийі бар файлды жүктеп алу. Екіншісі - сценарийді scriptBody ішіне енгізу арқылы. Менің білуімше, executeScript квадраты бірнеше тілдерді қолдайды - олардың бірі өте қызықты. Мен java әзірлеушілерінің көңілін қалдырамын - мұндай квадраттарда java тілінде сценарий жаза алмайсыз. Шынымен қалағандар үшін сіз өзіңіздің жеке квадратыңызды жасап, оны NIFI жүйесіне қосуыңыз керек. Бүкіл операция бубенмен ұзақ бимен бірге жүреді, біз бұл мақалада оны қарастырмаймыз. Мен керемет тілді таңдадым. Төменде SOAP хабарламасындағы идентификаторды біртіндеп жаңартатын сынақ сценарийі берілген. Айта кету керек. Сіз flowFile файлынан файлды алып, оны жаңартасыз, оны сол жерге қайта қою керек екенін ұмытпаңыз, жаңартылды. Сонымен қатар, барлық кітапханалар қамтылмағанын атап өткен жөн. Сіз әлі де libs біреуін импорттауыңыз керек болуы мүмкін. Тағы бір кемшілігі - бұл шаршыдағы сценарийді жөндеу өте қиын. 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 екенін көрсету керек.

Хост және әрекет (soapAction) сияқты бірнеше сипаттарды қосыңыз. Біз сақтаймыз және тексереміз. SOAP сұрауларын жіберу туралы қосымша мәліметтерді көре аласыз осында

Біз NIFI процестерін пайдаланудың бірнеше нұсқасын қарастырдық. Олар қалай өзара әрекеттеседі және олардың нақты пайдасы қандай? Қарастырылған мысалдар сынақ болып табылады және шайқаста болған жағдайдан біршама ерекшеленеді. Бұл мақала әзірлеушілер үшін аздап пайдалы болады деп үміттенемін. Назарларыңызға рахмет. Сұрақтарыңыз болса жазыңыздар. Мен жауап беруге тырысамын.

Ақпарат көзі: www.habr.com