Въведение
Стана така, че на текущото ми работно място трябваше да се запозная с тази технология. Ще започна с малка предистория. На поредната среща нашият екип бе информиран, че трябва да създаде интеграция с известна система. Под интеграцията се имаше предвид, че тази известна система ще ни изпраща заявки през 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 поставя файла в файловото хранилище. Конфигурационните параметри можете да видите по-долу.
Следва да се обърне внимание на факта, че всеки квадрат е отделен процес, който трябва да бъде задействан. Разгледахме най-простия пример, който не изисква сложна персонализация. Сега ще разгледаме процес, който е малко по-сложен, където ще пишем малко на Groovy.
По-сложен пример
Услугата за пренос на данни към потребителя се оказа малко по-сложна заради процеса на модификация на SOAP съобщението. Общият процес е представен на изображението по-долу.
Идеята тук също не е особено сложна: получихме запитване от потребителя, че му трябват данни, изпратихме отговор, че сме получили съобщението, задействали сме процеса на получаване на файла с отговор, след което го редактирахме с определена логика, и накрая предадохме файла на потребителя под формата на SOAP съобщение на сървъра.
Не мисля, че трябва да описвам отново квадратите, които видяхме по-горе — направо ще преминем към новите. Ако е необходимо да редактирате определен файл и обикновените квадратчета, като ReplaceText, не са подходящи, ще трябва да напишете свой скрипт. Това може да се направи с помощта на квадрата ExecuteGroovyScript. Неговите настройки са представени по-долу.
Има два варианта за зареждане на скрипта в този квадрат. Първият — чрез зареждане на файла със скрипта. Вторият — чрез вмъкване на скрипта в scriptBody. Няма да изненадвам 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
