Введение Так получилось, что на моем нынешнем месте работы мне пришлось познакомиться с этой технологией.
Я начну с небольшой предыстории.
На следующей встрече нашей команде сказали, что нам нужно создать интеграцию с известная система .
Под интеграцией подразумевалось, что эта известная система будет отправлять нам запросы по HTTP к определенной конечной точке, а мы, как ни странно, будем отправлять обратно ответы в виде SOAP-сообщения.
Все кажется простым и тривиальным.
Отсюда следует, что вам нужно.
Задача
Создайте 3 сервиса.Первый из них — Служба обновления баз данных.
Этот сервис при поступлении новых данных из сторонней системы обновляет данные в базе данных и формирует файл в формате CSV для передачи его в следующую систему.
Называется конечная точка второго сервиса — FTP Transport Service, который принимает переданный файл, проверяет его и помещает в файловое хранилище через FTP. Третий сервис — сервис передачи потребительских данных — работает асинхронно с первыми двумя.
Он получает запрос от сторонней внешней системы на получение рассмотренного выше файла, берет готовый файл ответа, модифицирует его (обновляет поля id, описание, linkToFile) и отправляет ответ в виде SOAP-сообщения.
То есть общая картина такая: первые два сервиса начинают свою работу только тогда, когда приходят данные для обновления.
Третий сервис работает постоянно, поскольку потребителей информации много, около 1000 запросов данных в минуту.
Сервисы доступны постоянно, а их экземпляры расположены в разных средах, таких как тестовая, демонстрационная, предпроизводственная и рабочая.
Ниже представлена диаграмма того, как работают эти службы.
Сразу уточню, что некоторые детали упрощены во избежание лишней сложности.
Техническое углубление
Планируя решение проблемы, мы в первую очередь решили делать приложения на Java с использованием фреймворка Spring, балансировщика Nginx, базы данных Postgres и других технических и не очень технических вещей.Поскольку время разработки технического решения позволило нам рассмотреть и другие подходы к решению данной проблемы, наш взгляд упал на модную в определенных кругах технологию Apache NIFI. Скажу сразу, что данная технология позволила нам заметить эти 3 сервиса.
В этой статье будет описана разработка сервиса транспортировки файлов и сервиса передачи данных потребителю, но если статья окажется полезной, я напишу о сервисе обновления данных в базе данных.
Что это
NIFI — это распределенная архитектура для быстрой параллельной загрузки и обработки данных, большое количество плагинов для источников и преобразований, версионирования конфигураций и многое другое.Приятным бонусом является то, что им очень легко пользоваться.
Тривиальные процессы, такие как getFile, sendHttpRequest и другие, можно представить в виде квадратов.
Каждый квадрат представляет собой процесс, взаимодействие которого можно увидеть на рисунке ниже.
Написана более подробная документация по взаимодействию при настройке процесса.
Здесь , для тех, кто говорит по-русски - Здесь .
В документации прекрасно описано, как распаковать и запустить NIFI, а также как создавать процессы, известные также как квадраты.
Идея написать статью родилась после долгих поисков и структурирования полученной информации во что-то осознанное, а также желания немного облегчить жизнь будущим разработчикам.
Пример
Рассмотрен пример того, как квадраты взаимодействуют друг с другом.Общая схема достаточно проста: Получаем HTTP-запрос (Теоретически с файлом в теле запроса.
Для демонстрации возможностей NIFI в этом примере запрос запускает процесс получения файла из локального файлового хранилища.
), затем отправляем обратно ответ, что запрос получен, параллельно идет процесс получения файла с ФХ и далее процесс его перемещения по FTP на ФХ.
Стоит уточнить, что процессы взаимодействуют друг с другом через так называемый flowFile. Это базовый объект в NIFI, в котором хранятся атрибуты и контент. Контент — это данные, представленные файлом потока.
То есть, грубо говоря, если вы получаете файл с одного квадрата и передаете его в другой, то содержимое будет вашим файлом.
Как видите, на этой картинке показан общий процесс.
HandleHttpRequest — принимает запросы, replaceText — формирует тело ответа, HandleHttpResponse — отправляет ответ. FetchFile — получает файл из файлового хранилища, передает его в квадрат. PutSftp — кладет этот файл на FTP, по указанному адресу.
Теперь подробнее об этом процессе.
В данном случае просьба — это начало всего.
Давайте посмотрим на его параметры конфигурации.
Здесь все достаточно тривиально за исключением StandardHttpContextMap — это своего рода сервис, позволяющий отправлять и получать запросы.
Более подробно и даже на примерах можно посмотреть - Здесь Далее давайте посмотрим на параметры конфигурации replaceText квадрата.
Стоит обратить внимание на replaceValue — это то, что будет возвращено пользователю в виде ответа.
В настройках можно настроить уровень логирования, можно посмотреть логи {куда распаковали nifi}/nifi-1.9.2/logs, там же есть параметры неудачи/успеха - на основе этих параметров можно регулировать процесс в целом .
То есть в случае успешной обработки текста будет вызван процесс отправки ответа пользователю, а в другом случае мы просто логируем неуспешный процесс.
В свойствах HandleHttpResponse нет ничего особенно интересного, кроме статуса при успешном создании ответа.
С запросом и ответом разобрались — переходим к получению файла и размещению его на FTP-сервере.
FetchFile — получает файл по указанному в настройках пути и передает его следующему процессу.
А дальше квадрат PutSftp - помещает файл в файловое хранилище.
Параметры конфигурации мы видим ниже.
Стоит обратить внимание на то, что каждый квадрат – это отдельный процесс, который необходимо запустить.
Мы рассмотрели самый простой пример, не требующий какой-либо сложной настройки.
Далее мы рассмотрим процесс немного сложнее, где немного распишем пазы.
Более сложный пример
Служба передачи данных потребителю оказалась немного сложнее из-за процесса модификации SOAP-сообщения.Общий процесс показан на рисунке ниже.
Здесь идея тоже не особо сложная: мы получили запрос от потребителя о том, что ему нужны данные, отправили ответ о том, что он получил сообщение, запустили процесс получения файла ответов, затем отредактировали его с определенной логикой, а затем передал файл потребителю в виде сообщения SOAP на сервер.
Думаю, нет необходимости заново описывать те квадраты, которые мы видели выше – перейдем сразу к новым.
Если вам нужно отредактировать какой-либо файл и обычные квадраты типа replaceText не подходят, придется написать собственный скрипт. Это можно сделать с помощью квадрата ExecuteGroogyScript. Его настройки представлены ниже.
Есть два варианта загрузки скрипта в этот квадрат. Первый — путем скачивания файла со скриптом.
Второй — вставка скрипта в scriptBody. Насколько я знаю, квадрат ExecuteScript поддерживает несколько языков — один из них groovy. Разочарую java-разработчиков — в таких квадратах скрипты на java не напишешь.
Для тех, кто очень хочет, нужно создать свой собственный квадрат и добавить его в систему NIFI. Вся эта операция сопровождается довольно долгим танцем с бубном, которым мы не будем заниматься в этой статье.
Я выбрал заводной язык.
Ниже приведен тестовый сценарий, который просто постепенно обновляет идентификатор в сообщении SOAP. Это важно отметить.
Берешь файл из flowFile и обновляешь, не забывая, что его нужно положить обратно туда, обновленный.
Также стоит отметить, что включены не все библиотеки.
Может случиться так, что вам все равно придется импортировать одну из библиотек.
Еще одним недостатком является то, что скрипт в этом квадрате довольно сложно отладить.
Существует способ подключиться к NIFI JVM и начать процесс отладки.
Лично я запускал локальное приложение и имитировал получение файла из сессии.
Я также делал отладку локально.
Ошибки, возникающие при загрузке скрипта, довольно легко гуглятся и записываются самим НИФИ в лог.
Собственно, на этом кастомизация квадрата заканчивается.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. Как они взаимодействуют и в чем их реальная польза? Рассмотренные примеры являются тестовыми и немного отличаются от того, что реально происходит в бою.
Надеюсь, эта статья будет немного полезна разработчикам.
Спасибо за внимание.
Если есть вопросы, пишите.
Я постараюсь ответить.
Теги: #java #nifi #Apache #программирование #Groovy #Flow #Habr #java #Groovy & Grails #Apache #Big Data
-
Угроза Безопасности Перьевого Накопителя
19 Oct, 24 -
Как Определить, Какие Обратные Ссылки Лучше?
19 Oct, 24 -
Считаем Ли Мы Деньги Открыто? Введение.
19 Oct, 24
