Apache NIFI - кратък преглед на възможностите на практика

въведение

Така се случи, че на сегашното ми място на работа трябваше да се запозная с тази технология. Ще започна с малко предистория. На следващия митинг на нашия екип беше казано, че трябва да създадем интеграция с известна система. Интегрирането означаваше, че тази добре позната система ще ни изпраща заявки чрез HTTP до конкретна крайна точка и, колкото и да е странно, ние ще изпращаме обратно отговори под формата на SOAP съобщение. Всичко изглежда просто и тривиално. Това, което следва от това е…

Задача

Създайте 3 услуги. Първата е услугата за актуализиране на база данни. Тази услуга, при получаване на нови данни от система на трета страна, актуализира данните в базата данни и генерира файл в CSV формат, за да го прехвърли към следващата система. Извиква се крайната точка на втората услуга - FTP Transportation Service, която получава прехвърления файл, валидира го и го поставя във файловото хранилище чрез FTP. Третата услуга - услуга за пренос на данни към потребителя, работи асинхронно с първите две. Той получава заявка от външна система на трета страна за получаване на файл, който беше обсъден по-горе, взема готов файл с отговори, модифицира го (актуализира идентификатора, описанието, полетата linkToFile) и изпраща отговор под формата на SOAP съобщение. Тоест като цяло картината е следната: първите две услуги започват работа само когато данните за актуализиране са пристигнали. Третата услуга работи постоянно, защото има много потребители на информация, около 1000 заявки за данни в минута. Услугите са винаги достъпни и техните екземпляри са разположени в различни среди, като например тест, демо, preprod и prod. По-долу има диаграма за това как работят тези услуги. Нека обясня веднага, че някои детайли са опростени, за да се избегне ненужната сложност.

Техническо задълбочаване

Когато планирахме решение на проблема, първо решихме да правим приложения в java, използвайки Spring framework, Nginx балансьор, Postgres база данни и други технически и не много технически неща. Тъй като времето за разработване на техническо решение направи възможно разглеждането на други подходи за решаване на този проблем, погледът падна върху технологията Apache NIFI, модерна в определени кръгове. Веднага трябва да кажа, че тази технология ни позволи да забележим тези 3 услуги. Тази статия ще опише разработването на услуга за транспортиране на файлове и услуга за трансфер на данни към потребителя, но ако статията се появи, ще пиша за услугата за актуализиране на данни в базата данни.

Какво е

NIFI е разпределена архитектура за бързо паралелно зареждане и обработка на данни, голям брой плъгини за източници и трансформации, версия на конфигурацията и много други. Приятен бонус е, че е много лесен за използване. Тривиални процеси като getFile, sendHttpRequest и други могат да бъдат представени като квадрати. Всеки квадрат представлява определен процес, чието взаимодействие може да се види на фигурата по-долу. Написана е по-подробна документация за взаимодействието при настройка на процеса тук , за тези, които на руски - тук. Документацията перфектно описва как се разопакова и стартира NIFI, както и как се създават процеси, те също са квадрати
Идеята за написване на статия се роди след дълго търсене и структуриране на получената информация в нещо съзнателно, както и желанието да улесним малко живота на бъдещите разработчици.

Пример

Разглежда се пример за това как квадратите взаимодействат помежду си. Общата схема е доста проста: Получаваме HTTP заявка (На теория, с файл в тялото на заявката. За да демонстрираме възможностите на NIFI, в този пример заявката стартира процеса на получаване на файл от локалния FH), след това изпращаме обратно отговор, че заявката е получена, успоредно с процеса на получаване на файл от 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 поддържа няколко PL - един от тях е groovy. Ще разочаровам разработчиците на java - не можете да пишете скриптове в java в такива квадрати. За тези, които наистина искат, трябва да създадете свой собствен квадрат и да го хвърлите в системата NIFI. Цялата тази операция е придружена от доста дълги танци с тамбура, които няма да разглеждаме в тази статия. Избрах грууви езика. По-долу е тестов скрипт, който просто постепенно актуализира идентификатора в SOAP съобщението. Важно е да се отбележи. Вземате файл от flowFile и го актуализирате, не забравяйте, че имате нужда от него, актуализиран, върнете го там. Също така си струва да се отбележи, че не всички библиотеки са свързани. Може да се окаже, че все още трябва да импортирате някоя от библиотеките. Недостатъкът е, че скриптът в този квадрат е доста труден за отстраняване на грешки. Има начин да се свържете с 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