Wprowadzenie
Zdarzyło się, że w moim obecnym miejscu pracy musiałem zapoznać się z tą technologią. Zacznę od krótkiej historii. znanego systemu. Pod integracją rozumiano, że ten znany system będzie nam wysyłać zapytania przez HTTP na określony punkt końcowy, a my, co może być zaskakujące, będziemy z powrotem wysyłać odpowiedzi w postaci wiadomości SOAP. Wydaje się to proste i trywialne. Z tego wynika, że trzeba…
Zadanie
Utworzyć 3 usługi. Pierwsza z nich to Usługa aktualizacji bazy danych. Ta usługa, po otrzymaniu nowych danych z systemu zewnętrznego, aktualizuje dane w bazie danych i generuje pewien plik w formacie CSV, aby przekazać go do następnego systemu. Wywoływany jest punkt końcowy drugiej usługi — Usługi transportu przez FTP, która otrzymuje przesłany plik, waliduje go i zapisuje w magazynie plików przez FTP. Trzecia usługa — Usługa przesyłania danych do konsumenta, działa asynchronicznie w porównaniu do dwóch pierwszych. Odbiera zapytanie z zewnętrznego systemu, dotyczące pliku, o którym mowa powyżej, pobiera gotowy plik odpowiedzi, modyfikuje go (aktualizuje pola id, description, linkToFile) i wysyła odpowiedź w postaci wiadomości SOAP. W skrócie, obraz przedstawia się następująco: pierwsze dwie usługi zaczynają działać tylko wtedy, gdy przychodzą dane do aktualizacji. Trzecia usługa działa nieprzerwanie, ponieważ jest wielu konsumentów informacji, około 1000 zapytań o dane na minutę. Usługi są zawsze dostępne, a ich instancje znajdują się w różnych środowiskach, takich jak testowe, demo, preprodukcja i produkcja. Poniżej przedstawiona jest schemat pracy tych usług. Od razu wyjaśnię, że niektóre szczegóły zostały uproszczone, aby uniknąć zbędnej złożoności.
Techniczne zagłębienie
Podczas planowania rozwiązania problemu, pierwotnie zdecydowano się na stworzenie aplikacji w języku Java z wykorzystaniem frameworka Spring, balancerem Nginx, bazą danych Postgres oraz innymi technicznymi i mniej technicznymi komponentami. Ponieważ czas na opracowanie rozwiązania technicznego pozwalał na rozważenie innych podejść do tego zadania, uwagę zwróciła modna w pewnych kręgach technologia Apache NIFI. Od razu zaznaczam, że ta technologia pozwoliła nam dostrzec te trzy usługi. W tym artykule zostanie opisana praca nad usługą transportu pliku i usługą przesyłania danych do odbiorcy, jednak jeśli artykuł się spodoba, napiszę o usłudze aktualizacji danych w bazie danych.
Czym to jest
NIFI to rozproszona architektura do szybkiego równoległego ładowania i przetwarzania danych, z dużą ilością wtyczek dla źródeł i transformacji, wersjonowaniem konfiguracji i wieloma innymi funkcjami. Miłym plusem jest to, że jest bardzo prosty w użyciu. Trywialne procesy, takie jak getFile, sendHttpRequest i inne — można przedstawić w postaci kwadratów. Każdy kwadrat reprezentuje pewien proces, którego interakcję można zobaczyć na poniższym rysunku. Szczegółowa dokumentacja dotycząca interakcji i konfiguracji procesów jest napisana , dla tych, którzy mówią po rosyjsku — . W dokumentacji doskonale opisano, jak rozpakować i uruchomić NIFI, a także jak tworzyć procesy, które są kwadratami
Pomysł napisania artykułu pojawił się po długotrwałych poszukiwaniach i strukturyzacji zebranych informacji w coś zrozumiałego, a także z chęci ułatwienia życia przyszłym programistom.
Przykład
Rozważmy przykład, jak kwadraty ze sobą współpracują. Ogólny schemat jest dość prosty: Otrzymujemy żądanie HTTP (teoretycznie z plikiem w treści żądania. Dla zobrazowania możliwości NIFI, w tym przykładzie żądanie uruchamia proces pobierania pliku z lokalnego FX), następnie wysyłamy z powrotem odpowiedź, że żądanie zostało odebrane, równocześnie uruchamia się proces pobierania pliku z FX, a następnie proces przenoszenia go przez FTP do FX. Warto wyjaśnić, że procesy współdziałają ze sobą za pośrednictwem tzw. flowFile. To podstawowy byt w NIFI, który przechowuje atrybuty i zawartość. Zawartość — to dane, które są reprezentowane przez plik strumieniowy. Mówiąc w uproszczeniu, jeśli otrzymałeś plik z jednego kwadratu i przekazujesz go do drugiego, zawartością będzie twój plik.
Jak zauważyłeś — na tym rysunku przedstawiono ogólny proces. HandleHttpRequest — przyjmuje żądania, ReplaceText — generuje treść odpowiedzi, HandleHttpResponse — zwraca odpowiedź. FetchFile — pobiera plik z przechowalni plików i przekazuje go do kwadratu PutSftp — umieszcza ten plik na FTP, pod wskazanym adresem. Teraz bardziej szczegółowo o tym procesie.
W tym przypadku — request to wszystko, co zaczyna. Przyjrzyjmy się jego parametrom konfiguracyjnym.
Tutaj wszystko jest dość trywialne, z wyjątkiem StandardHttpContextMap — to pewnego rodzaju usługa, która umożliwia wysyłanie i odbieranie żądań. Można to zobaczyć bardziej szczegółowo, nawet z przykładami —
Dalej przyjrzymy się parametrom konfiguracyjnym kwadratu ReplaceText. Należy zwrócić uwagę na ReplacementValue — to, co wróci do użytkownika jako odpowiedź. W ustawieniach można regulować poziom logowania, logi można sprawdzić {gdzie rozpakowano nifi}/nifi-1.9.2/logs, tam są także parametry failure/success — na podstawie tych parametrów można regulować proces w całości. To znaczy, w przypadku pomyślnego przetwarzania tekstu — wywoła się proces wysłania odpowiedzi do użytkownika, a w innym przypadku po prostu zalogujemy nieudany proces.
W właściwościach HandleHttpResponse nie ma nic szczególnie interesującego, poza statusem przy pomyślnym utworzeniu odpowiedzi.
Z żądaniem odpowiedzi się uporaliśmy — przejdźmy dalej do otrzymywania pliku i umieszczania go na serwerze FTP. FetchFile — pobiera plik z określonej w ustawieniach ścieżki i przekazuje go do następnego procesu.
A następnie kwadrat PutSftp — umieszcza plik w magazynie plików. Parametry konfiguracyjne możemy zobaczyć poniżej.
Warto zwrócić uwagę, że każdy kwadrat to oddzielny proces, który musi być uruchomiony. Rozpatrzyliśmy najprostszy przykład, który nie wymaga żadnej skomplikowanej personalizacji. Następnie rozważymy nieco bardziej złożony proces, w którym trochę napiszemy na Groovy.
Bardziej złożony przykład
Usługa przesyłania danych do konsumenta okazała się nieco bardziej skomplikowana z powodu procesu modyfikacji wiadomości SOAP. Ogólny proces przedstawiono na rysunku poniżej.
Tutaj idea również nie jest zbyt skomplikowana: otrzymaliśmy zapytanie od konsumenta, że potrzebuje danych, wysłaliśmy odpowiedź, że otrzymaliśmy wiadomość, uruchomiliśmy proces otrzymywania pliku odpowiedzi, następnie edytowaliśmy go zgodnie z określoną logiką, po czym przekazaliśmy plik konsumentowi w postaci wiadomości SOAP na serwer.
Myślę, że nie ma sensu opisywać na nowo tych kwadratów, które widzieliśmy wcześniej — przejdźmy od razu do nowych. Jeśli musisz edytować jakiś plik, a standardowe kwadraty typu ReplaceText nie pasują, będziesz musiał napisać własny skrypt. Można to zrobić za pomocą kwadratu ExecuteGroogyScript. Ustawienia są przedstawione poniżej.
Istnieją dwie metody ładowania skryptu do tego kwadratu. Pierwsza to wgranie pliku ze skryptem. Druga to wstawienie skryptu do scriptBody. O ile mi wiadomo, kwadrat executeScript obsługuje kilka języków programowania — jednym z nich jest groovy. Rozczaruję programistów java — w javie nie można pisać skryptów w takich kwadratach. Dla tych, którzy naprawdę chcą — trzeba stworzyć swój własny niestandardowy kwadrat i wprowadzić go do systemu NIFI. Cała ta operacja wiąże się z dość długimi tańcami wokół bębna, którymi nie będziemy się zajmować w ramach tego artykułu. Wybrałam język groovy. Poniżej znajduje się testowy skrypt, który po prostu inkrementalnie aktualizuje id w wiadomości SOAP. Ważne jest, aby pamiętać, że bierzesz plik z flowFile, aktualizujesz go, nie zapominając włożyć go z powrotem. Również należy zauważyć, że nie wszystkie biblioteki są podłączone. Może się zdarzyć, że będziesz musiał zaimportować jedną z bibliotek. Kolejnym minusem jest to, że skrypt w tym kwadracie jest dość trudny do debugowania. Istnieje sposób na połączenie się z JVM NIFI i rozpoczęcie procesu debugowania. Osobiście uruchomiłam lokalną aplikację i symulowałam odbieranie pliku z sesji. Debugowaniem również zajmowałam się lokalnie. Błędy, które pojawiają się podczas ładowania skryptu, są dość łatwe do znalezienia i są rejestrowane przez NIFI w logach.
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)
}Na tym kończy się dostosowywanie kwadratu. Uaktualniony plik jest następnie przekazywany do kwadratu, który zajmuje się wysyłaniem pliku na serwer. Poniżej przedstawiono ustawienia tego kwadratu.
Opisujemy metodę, za pomocą której będzie przekazywana wiadomość SOAP. Wskazujemy, dokąd. Następnie należy określić, że jest to dokładnie SOAP.
Dodajemy kilka właściwości, takich jak host i akcja (soapAction). Zapisujemy i sprawdzamy. Więcej informacji na temat wysyłania żądań SOAP można znaleźć.
Przeanalizowaliśmy kilka przypadków użycia procesów NIFI. Jak one współdziałają i jakie przynoszą rzeczywiste korzyści. Rozważane przykłady są testowe i nieco różnią się od tego, co dzieje się w produkcji. Mam nadzieję, że ten artykuł będzie nieco przydatny dla programistów. Dziękuję za uwagę. Jeśli masz jakiekolwiek pytania – pisz. Postaram się odpowiedzieć.
Źródło: habr.com
