Wprowadzenie
Tak się złożyło, że w moim obecnym miejscu pracy musiałem zapoznać się z tą technologią. Zacznę od małego tła. Na kolejnym spotkaniu powiedziano naszemu zespołowi, że musimy stworzyć integrację z znany system. Przez integrację rozumieno, że ten dobrze znany system będzie wysyłał nam żądania za pośrednictwem protokołu HTTP do określonego punktu końcowego, a my, co dziwne, będziemy odsyłać odpowiedzi w postaci komunikatu SOAP. Wszystko wydaje się proste i banalne. Z tego wynika, że potrzebujesz...
Zadanie
Utwórz 3 usługi. Pierwszą z nich jest Usługa Aktualizacji Baz Danych. Usługa ta, gdy napływają nowe dane z zewnętrznego systemu, aktualizuje dane w bazie danych i generuje plik w formacie CSV w celu przesłania ich do następnego systemu. Punkt końcowy drugiej usługi nazywany jest usługą transportową FTP, która odbiera przesłany plik, sprawdza go i umieszcza w magazynie plików za pośrednictwem protokołu FTP. Trzecia usługa, usługa przesyłania danych konsumenckich, działa asynchronicznie z dwoma pierwszymi. Otrzymuje żądanie z zewnętrznego systemu zewnętrznego o przyjęcie omawianego powyżej pliku, pobiera gotowy plik odpowiedzi, modyfikuje go (aktualizuje identyfikator, opis, pola linkToFile) i wysyła odpowiedź w postaci komunikatu SOAP. Oznacza to, że ogólny obraz jest następujący: pierwsze dwie usługi rozpoczynają pracę dopiero po nadejściu danych do aktualizacji. Trzecia usługa działa stale, ponieważ odbiorców informacji jest wielu, około 1000 żądań danych na minutę. Usługi są dostępne stale, a ich instancje zlokalizowane są w różnych środowiskach, takich jak testowe, demonstracyjne, przedprodukcyjne i prod. Poniżej znajduje się schemat działania tych usług. Od razu wyjaśniam, że niektóre szczegóły zostały uproszczone, aby uniknąć niepotrzebnej komplikacji.
Pogłębienie techniczne
Planując rozwiązanie problemu, zdecydowaliśmy się najpierw na wykonanie aplikacji w Javie z wykorzystaniem frameworka Spring, balansera Nginx, bazy danych Postgres i innych rzeczy technicznych i mniej technicznych. Ponieważ czas opracowania rozwiązania technicznego pozwolił nam rozważyć inne podejścia do rozwiązania tego problemu, nasz wzrok padł na modną w niektórych kręgach technologię Apache NIFI. Od razu powiem, że dzięki tej technologii udało nam się dostrzec te 3 usługi. W tym artykule zostanie opisany rozwój usługi transportu plików i usługi przesyłania danych do konsumenta, ale jeśli artykuł będzie przydatny, napiszę o usłudze aktualizacji danych w bazie danych.
Co to jest
NIFI to rozproszona architektura do szybkiego równoległego ładowania i przetwarzania danych, duża liczba wtyczek do źródeł i transformacji, wersjonowanie konfiguracji i wiele więcej. Miłym bonusem jest to, że jest bardzo łatwy w użyciu. Trywialne procesy, takie jak getFile, sendHttpRequest i inne, można przedstawić w postaci kwadratów. Każdy kwadrat reprezentuje proces, którego interakcję można zobaczyć na poniższym rysunku. Napisano bardziej szczegółową dokumentację dotyczącą interakcji konfiguracji procesu , dla tych, którzy mówią po rosyjsku - . Dokumentacja doskonale opisuje sposób rozpakowania i uruchomienia NIFI oraz tworzenia procesów, zwanych też kwadratami
Pomysł napisania artykułu zrodził się po długich poszukiwaniach i uporządkowaniu otrzymanych informacji w coś świadomego, a także chęci ułatwienia życia przyszłym programistom.
Przykład
Rozważono przykład interakcji kwadratów ze sobą. Ogólny schemat jest dość prosty: Otrzymujemy żądanie HTTP (w teorii z plikiem w treści żądania. Aby zademonstrować możliwości NIFI, w tym przykładzie żądanie rozpoczyna proces odbierania pliku z lokalnego magazynu plików) ), wówczas odsyłamy odpowiedź, że żądanie zostało przyjęte, równolegle z procesem otrzymania pliku z FH i następnie przeniesieniem go poprzez FTP do FH. Warto doprecyzować, że procesy współdziałają ze sobą poprzez tzw. flowFile. Jest to podstawowa jednostka w NIFI, która przechowuje atrybuty i treść. Treść to dane reprezentowane przez plik strumieniowy. Oznacza to, że z grubsza jeśli otrzymasz plik z jednego kwadratu i przeniesiesz go do innego, zawartość będzie Twoim plikiem.
Jak widać, to zdjęcie pokazuje ogólny proces. HandleHttpRequest – akceptuje żądania, ZamieńText – generuje treść odpowiedzi, HandleHttpResponse – wysyła odpowiedź. FetchFile - pobiera plik z magazynu plików, przesyła go do kwadratu PutSftp - umieszcza ten plik na FTP, pod podanym adresem. Teraz więcej o tym procesie.
W tym przypadku prośba jest początkiem wszystkiego. Przyjrzyjmy się jego parametrom konfiguracyjnym.
Wszystko tutaj jest dość banalne z wyjątkiem StandardHttpContextMap - jest to rodzaj usługi, która pozwala na wysyłanie i odbieranie żądań. Bardziej szczegółowo, a nawet z przykładami, możesz zobaczyć -
Następnie przyjrzyjmy się parametrom konfiguracyjnym ZamieńText kwadratu. Warto zwrócić uwagę na opcję FrequencyValue – to właśnie ona zostanie zwrócona użytkownikowi w formie odpowiedzi. W ustawieniach możesz dostosować poziom logowania, możesz zobaczyć logi {gdzie rozpakowałeś nifi}/nifi-1.9.2/logs, są też parametry niepowodzenia/sukcesu - na podstawie tych parametrów możesz regulować proces jako całość . Oznacza to, że w przypadku pomyślnego przetworzenia tekstu zostanie wywołany proces wysyłania odpowiedzi do użytkownika, a w innym przypadku po prostu zarejestrujemy nieudany proces.
We właściwościach HandleHttpResponse nie ma nic szczególnie interesującego poza statusem pomyślnego utworzenia odpowiedzi.
Uporządkowaliśmy żądanie i odpowiedź - przejdźmy do odebrania pliku i umieszczenia go na serwerze FTP. FetchFile - pobiera plik pod ścieżką określoną w ustawieniach i przekazuje go do następnego procesu.
A następnie kwadrat PutSftp - umieszcza plik w magazynie plików. Poniżej możemy zobaczyć parametry konfiguracyjne.
Warto zwrócić uwagę na fakt, że każdy kwadrat to osobny proces, który należy uruchomić. Przyjrzeliśmy się najprostszemu przykładowi, który nie wymaga żadnego skomplikowanego dostosowywania. Następnie przyjrzymy się procesowi nieco bardziej skomplikowanemu, gdzie napiszemy trochę na rowkach.
Bardziej złożony przykład
Usługa przesyłania danych do konsumenta okazała się nieco bardziej skomplikowana ze względu na proces modyfikacji komunikatu SOAP. Ogólny proces pokazano na poniższym rysunku.
Tutaj pomysł też nie jest szczególnie skomplikowany: otrzymaliśmy od konsumenta żądanie, że potrzebuje danych, wysłaliśmy odpowiedź, że otrzymał wiadomość, rozpoczęliśmy proces otrzymania pliku odpowiedzi, następnie zredagowaliśmy go według określonej logiki, a następnie przekazał plik konsumentowi w formie komunikatu SOAP na serwer.
Myślę, że nie ma potrzeby opisywać ponownie kwadratów, które widzieliśmy powyżej - przejdźmy od razu do nowych. Jeśli potrzebujesz edytować jakiś plik i zwykłe kwadraty typu ZamieńText nie są odpowiednie, będziesz musiał napisać własny skrypt. Można to zrobić za pomocą kwadratu ExecuteGroogyScript. Poniżej przedstawiono jego ustawienia.
Istnieją dwie możliwości załadowania skryptu do tego kwadratu. Pierwszy polega na pobraniu pliku ze skryptem. Drugi polega na wstawieniu skryptu do scriptBody. O ile wiem, kwadrat wykonawczy obsługuje kilka języków - jeden z nich jest groovy. Rozczaruję programistów Java - w takich kwadratach nie można pisać skryptów w Javie. Dla tych, którzy naprawdę chcą, musisz stworzyć własny kwadrat i dodać go do systemu NIFI. Całości tej operacji towarzyszy dość długi taniec z tamburynem, którym nie będziemy się zajmować w tym artykule. Wybrałem odlotowy język. Poniżej znajduje się skrypt testowy, który po prostu stopniowo aktualizuje identyfikator w komunikacie SOAP. Ważne jest, aby pamiętać. Bierzesz plik z flowFile i aktualizujesz go, nie zapominaj, że musisz go tam umieścić z powrotem, zaktualizowany. Warto również zauważyć, że nie wszystkie biblioteki są uwzględnione. Może się zdarzyć, że nadal będziesz musiał zaimportować jedną z bibliotek. Kolejną wadą jest to, że skrypt w tym kwadracie jest dość trudny do debugowania. Istnieje sposób na połączenie się z maszyną NIFI JVM i rozpoczęcie procesu debugowania. Osobiście uruchomiłem aplikację lokalną i symulowałem odbiór pliku z sesji. Debugowałem też lokalnie. Błędy pojawiające się podczas ładowania skryptu są dość łatwe dla Google i są zapisywane przez samą NIFI w dzienniku.
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)
}Właściwie na tym kończy się dostosowywanie kwadratu. Następnie zaktualizowany plik przesyłany jest do kwadratu, który odpowiada za przesłanie pliku na serwer. Poniżej znajdują się ustawienia tego kwadratu.
Opisujemy metodę przesyłania komunikatu SOAP. Piszemy gdzie. Następnie musisz wskazać, że jest to SOAP.
Dodaj kilka właściwości, takich jak host i akcja (soapAction). Zapisujemy i sprawdzamy. Możesz zobaczyć więcej szczegółów na temat wysyłania żądań SOAP
Przyjrzeliśmy się kilku opcjom wykorzystania procesów NIFI. Jak na siebie oddziałują i jaka jest ich realna korzyść? Podane przykłady mają charakter testowy i różnią się nieco od tego, co faktycznie dzieje się w walce. Mam nadzieję, że ten artykuł będzie choć trochę przydatny dla programistów. Dziękuję za uwagę. Jeżeli masz jakieś pytania napisz. Spróbuję odpowiedzieć.
Źródło: www.habr.com