Úvod
Na mojej súčasnej pracovnej pozícii som sa musel oboznámiť s touto technológiou. Začnem s malým pozadím. Na ďalšom mítingu nám tím povedal, že musíme vyvinúť integráciu s známym systémom. Pod integráciou sa rozumelo, že tento známý systém nám bude posielať požiadavky cez HTTP na určitý koncový bod, a my, ako to ani nie je prekvapujúce, budeme posielať späť odpovede vo forme SOAP správ. Zdá sa, že je to všetko jednoduché a triviálne. Z toho vyplýva, že treba...
Úloha
Vytvoriť 3 služby. Prvá z nich je Služba aktualizácie DB. Táto služba, po prijatí nových údajov zo strany systému, aktualizuje údaje v databáze a generuje akýsi súbor vo formáte CSV, ktorý sa prenáša do nasledujúceho systému. Následne sa vyvoláva koncový bod druhej služby – Služby transportu cez FTP, ktorá prijíma odoslaný súbor, validuje ho a ukladá do súborového úložiska cez FTP. Tretia služba – Služba prenosu údajov pre spotrebiteľa, pracuje asynchrónne s prvými dvoma. Prijíma požiadavku od vonkajšieho systému na získanie súboru, o ktorom sa hovorilo vyššie, berie hotový súbor odpovede, modifikuje ho (aktualizuje polia id, description, linkToFile) a posiela odpoveď vo forme SOAP správy. Teda v celku vyzerá situácia takto: prvé dve služby začínajú svoju prácu len vtedy, keď prídu údaje na aktualizáciu. Tretia služba funguje nepretržite, pretože je veľa spotrebiteľov informácií, približne 1000 požiadaviek na prijatie údajov za minútu. Služby sú neustále dostupné a ich inštancie sa nachádzajú v rôznych prostrediach, ako sú test, demo, pre-prod a prod. Nižšie je znázornená schéma práce týchto služieb. Hneď vysvetlím, že niektoré detaily sú zjednodušené na vyhnutie sa zbytočnej zložitosti.
Technické prehĺbenie
Pri plánovaní riešenia úlohy sme sa najprv rozhodli vytvoriť aplikácie v jazyku Java s použitím Spring frameworku, vyvažovača zaťaženia Nginx, databázy Postgres a ďalších technických a menej technických vecí. Keďže čas na spracovanie technického riešenia umožňoval zvážiť aj iné prístupy k tejto úlohe, náš pohľad sa obrátil na módnu technológiu Apache NIFI v určitých kruhoch. Hneď poviem, že táto technológia nám umožnila všimnúť si tieto 3 služby. V tomto článku bude opísaný vývoj služby na prenos súborov a služby na prenos údajov k spotrebiteľovi, avšak ak sa článok zapáči, napíšem aj o službe na aktualizáciu údajov v databáze.
Čo to je
NIFI predstavuje distribuovanú architektúru pre rýchle paralelné nahrávanie a spracovanie údajov, množstvo pluginov pre zdroje a transformácie, spravovanie verzií konfigurácií a mnoho ďalšieho. Príjemným bonusom je, že je veľmi jednoduchý na používanie. Triviálne procesy, ako getFile, sendHttpRequest a iné, možno zobraziť ako štvorce. Každý štvorec predstavuje určitý proces, ktorého interakciu je možné vidieť na obrázku nižšie. Podrobnejšia dokumentácia o interakcii a nastavení procesov je napísaná , pre tých, ktorí hovoria po rusky — . V dokumentácii je skvele popísané, ako rozbalit a spustiť NIFI, ako aj, ako vytvoriť procesy, teda štvorce.
Myšlienka napísať článok vznikla po dlhých hľadaní a štruktúrovaní získaných informácií do niečoho zrozumiteľného, ako aj z túžby trochu uľahčiť život budúcim vývojárom.
Príklad
Prezentovali sme príklad interakcie štvorcov. Celková schéma je dosť jednoduchá: prijímame HTTP požiadavku (teoreticky s súborom v tele požiadavky. Na demonštráciu možností NIFI, v tomto príklade požiadavka spúšťa proces získavania súboru z lokálnej FX), následne posielame späť odpoveď, že požiadavka bola prijatá, paralelne sa spúšťa proces získavania súboru z FX a ďalej proces presunu cez FTP do FX. Je potrebné zdôrazniť, že procesy spolu interagujú prostredníctvom takzvaného flowFile. Toto je základná entita v NIFI, ktorá uchováva atribúty a obsah. Obsah sú dáta, ktoré sú prezentované ako súbor prúdu. To znamená, ak ste prijali súbor z jedného štvorca a prenášate ho do druhého, obsahom bude váš súbor.
Ako môžete vidieť — na tomto obrázku je znázornený celkový proces. HandleHttpRequest — prijíma požiadavky, ReplaceText — generuje telo odpovede, HandleHttpResponse — poskytuje odpoveď. FetchFile — získava súbor z úložiska a odovzdáva ho štvorcu PutSftp — umiestňuje tento súbor na FTP na zadanú adresu. Teraz podrobnejšie o tomto procese.
V tomto prípade — request je všemu začiatok. Pozrime sa na jeho konfiguračné parametre.
Tu je všetko dosť triviálne, okrem StandartHttpContextMap — to je akýsi servis, ktorý umožňuje posielať a prijímať požiadavky. Podrobnejšie a dokonca s príkladmi si môžete pozrieť —
Ďalej sa pozrieme na konfiguračné parametre štvorca ReplaceText. Zamerajte sa na ReplacementValue — to je to, čo sa vráti používateľovi ako odpoveď. V nastaveniach môžete regulovať úroveň logovania, logy si môžete pozrieť {kam sme rozbalili nifi}/nifi-1.9.2/logs, tam sú tiež parametre failure/success — na základe týchto parametrov môžete regulovať proces ako celok. T.j., v prípade úspešného spracovania textu sa spustí proces odosielania odpovede používateľovi, a v inom prípade jednoducho zalogujeme neúspešný proces.
Vo vlastnostiach HandleHttpResponse nie je nič zaujímavé, okrem statusu pri úspešnom vytvorení odpovede.
S požiadavkou a odpoveďou sme si objasnili — prejdeme ďalej k získavaniu súboru a jeho umiestneniu na FTP server. FetchFile — získava súbor podľa cesty uvedenej v nastaveniach a odovzdáva ho do nasledujúceho procesu.
A ďalej štvorec PutSftp — umiestňuje súbor do úložiska súborov. Parametre konfigurácie si môžeme pozrieť nižšie.
Je potrebné si uvedomiť, že každý štvorec je samostatný proces, ktorý musí byť spustený. Preskúmali sme ten najjednoduchší príklad, ktorý si nevyžaduje žiadnu zložitú prispôsobenie. Teraz sa pozrieme na proces, ktorý je o niečo zložitejší, kde si trochu poskriptujeme na groovoch.
Zložitejší príklad
Služba prenosu údajov pre spotrebiteľa sa ukázala ako o niečo zložitejšia v dôsledku procesu modifikácie SOAP správy. Celkový proces je znázornený na obrázku nižšie.
Tu nie je myšlienka príliš zložitá: prijali sme požiadavku od spotrebiteľa, že potrebuje údaje, poslali sme odpoveď, že sme dostali správu, spustili sme proces načítania súboru odpovede, potom sme ho upravili s určitú logikou, po čom sme odovzdali súbor spotrebiteľovi vo forme SOAP správy na server.
Myslím, že nemá zmysel znovu opisovať tie štvorce, ktoré sme videli vyššie — prejdime rovno na nové. Ak potrebujete upraviť akýkoľvek súbor a bežné štvorce ako ReplaceText nevyhovujú, budete musieť napísať vlastný skript. To je možné vykonať pomocou štvorca ExecuteGroogyScript. Jeho nastavenia sú uvedené nižšie.
Existujú dva spôsoby, ako načítať skript do tohto bloku. Prvý — nahraním súboru so skriptom. Druhý — vložením skriptu do scriptBody. Pokiaľ viem, blok executeScript podporuje niekoľko programovacích jazykov — jedným z nich je groovy. Oklamem java vývojárov — na javu nie je možné písať skripty v takýchto blokoch. Pre tých, ktorí veľmi túžia — je potrebné vytvoriť svoj vlastný prispôsobený blok a pridať ho do systému NIFI. Celý tento proces si vyžaduje dosť dlhé tance s bubnom, ktorým sa nebudeme v rámci tohto článku zaoberať. Vybrala som jazyk groovy. Nižšie je uvedený testovací skript, ktorý jednoducho inkrementálne aktualizuje id v SOAP správe. Dôležité je poznamenať, že beriete súbor z flowFile, aktualizujete ho a nesmiete zabudnúť, že ho treba, aktualizovaný, vrátiť späť. Rovnako je potrebné poznamenať, že nie všetky knižnice sú pripojené. Môže sa stať, že vám nakoniec bude treba importovať jednu z knižníc. Ďalšou nevýhodou je, že skript v tomto bloku sa pomerne ťažko ladí. Existuje spôsob, ako sa pripojiť k JVM NIFI a začať proces ladenia. Osobne som spustila svoju lokálnu aplikáciu a simulovala získavanie súboru zo relácie. Ladením som sa tiež zaoberala lokálne. Chyby, ktoré sa objavia pri nahrávaní skriptu, sa dosť jednoducho googlia a píšu sa samotným NIFI do logu.
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("Chyba počas spracovania validate.groovy", e)
session.transfer(flowFile, REL_FAILURE)
}Vlastne na tomto končí prispôsobenie bloku. Aktualizovaný súbor sa ďalej prenáša do bloku, ktorý sa zaoberá odosielaním súboru na server. Nižšie sú uvedené nastavenia tohto bloku.
Opisujeme metódu, ktorou bude odosielané SOAP správy. Napíšeme, kam. Ďalej je potrebné uviesť, že ide skutočne o SOAP.
Pridávame niekoľko vlastností, ako je host a akcia (soapAction). Uložíme a skontrolujeme. Podrobnejšie informácie o tom, ako posielať SOAP požiadavky, nájdete.
Preskúmali sme niekoľko možností využitia procesov NIFI. Ako interagujú a aký je ich skutočný prínos. Prezentované príklady sú testovacie a trochu sa líšia od toho, čo sa skutočne používa v praxi. Dúfam, že tento článok bude pre vývojárov trochu užitočný. Ďakujem za pozornosť. Ak máte nejaké otázky, píšte. Pokúsim sa odpovedať.
Zdroj: habr.com
