Apache NIFI - Stručný prehľad príležitostí v praxi

Úvod

Stalo sa, že na mojom súčasnom pracovisku som sa musel zoznámiť s touto technológiou. Začnem malým pozadím. Na ďalšom stretnutí bolo nášmu tímu povedané, že musíme vytvoriť integráciu s známy systém. Integráciou sa myslelo, že tento známy systém nám posiela požiadavky cez HTTP na konkrétny koncový bod a my, napodiv, spätne posielame odpovede vo forme SOAP správy. Všetko sa zdá byť jednoduché a triviálne. Z toho vyplýva, že potrebujete...

Úloha

Vytvorte 3 služby. Prvou z nich je Služba aktualizácie databázy. Táto služba, keď prídu nové údaje zo systému tretej strany, aktualizuje údaje v databáze a vygeneruje súbor vo formáte CSV na prenos do ďalšieho systému. Koncový bod druhej služby sa nazýva - FTP Transportná služba, ktorá prijme prenášaný súbor, overí ho a uloží ho do úložiska súborov cez FTP. Tretia služba, služba prenosu dát pre spotrebiteľov, pracuje asynchrónne s prvými dvoma. Prijme požiadavku od externého systému tretej strany na prijatie vyššie uvedeného súboru, vezme pripravený súbor odpovedí, upraví ho (aktualizuje polia id, popis, linkToFile) a odošle odpoveď vo forme správy SOAP. To znamená, že celkový obraz je takýto: prvé dve služby začnú svoju činnosť až po príchode údajov na aktualizáciu. Tretia služba funguje neustále, pretože existuje veľa spotrebiteľov informácií, približne 1000 žiadostí o údaje za minútu. Služby sú dostupné neustále a ich inštancie sa nachádzajú v rôznych prostrediach, ako sú testovacie, demo, predprodukčné a prod. Nižšie je uvedený diagram, ako tieto služby fungujú. Dovoľte mi hneď objasniť, že niektoré detaily boli zjednodušené, aby sa predišlo zbytočnej zložitosti.

Technické prehĺbenie

Pri plánovaní riešenia problému sme sa najskôr rozhodli robiť aplikácie v Jave pomocou frameworku Spring, Nginx balancer, Postgres databázu a ďalšie technické a nie až tak technické veci. Keďže čas na vývoj technického riešenia nám umožnil zvážiť aj iné prístupy k riešeniu tohto problému, náš pohľad padol na technológiu Apache NIFI, ktorá je v určitých kruhoch módna. Hneď poviem, že táto technológia nám umožnila všimnúť si tieto 3 služby. Tento článok popisuje vývoj služby prenosu súborov a služby prenosu údajov spotrebiteľovi, ale ak je článok užitočný, napíšem o službe na aktualizáciu údajov v databáze.

Čo je to

NIFI je distribuovaná architektúra pre rýchle paralelné načítanie a spracovanie dát, veľké množstvo pluginov pre zdroje a transformácie, verzovanie konfigurácií a mnoho ďalšieho. Príjemným bonusom je, že sa veľmi ľahko používa. Triviálne procesy ako getFile, sendHttpRequest a iné môžu byť reprezentované ako štvorce. Každý štvorec predstavuje proces, ktorého interakciu je možné vidieť na obrázku nižšie. Bola napísaná podrobnejšia dokumentácia o interakciách nastavenia procesu tu pre tých, ktorí hovoria po rusky - tu. Dokumentácia dokonale popisuje, ako rozbaliť a spustiť NIFI, ako aj vytvoriť procesy, známe tiež ako štvorce
Nápad napísať článok sa zrodil po dlhom hľadaní a štruktúrovaní získaných informácií do niečoho vedomého, ako aj túžbe uľahčiť budúcim vývojárom život.

Príklad

Uvažuje sa o príklade interakcie štvorcov. Všeobecná schéma je celkom jednoduchá: Prijmeme požiadavku HTTP (Teoreticky so súborom v tele požiadavky. Aby sme demonštrovali schopnosti NIFI, v tomto príklade požiadavka spustí proces prijatia súboru z lokálneho úložiska súborov ), potom pošleme späť odpoveď, že požiadavka bola prijatá, súbežne prebieha proces prijatia súboru z FH a následne proces presunu cez FTP do FH. Stojí za to objasniť, že procesy sa navzájom ovplyvňujú prostredníctvom takzvaného flowFile. Toto je základná entita v NIFI, ktorá ukladá atribúty a obsah. Obsah sú údaje, ktoré predstavuje súbor prúdu. Zhruba povedané, ak prijmete súbor z jedného štvorca a prenesiete ho na druhý, obsahom bude váš súbor.

Ako vidíte, tento obrázok ukazuje všeobecný proces. HandleHttpRequest – prijíma požiadavky, ReplaceText – vygeneruje telo odpovede, HandleHttpResponse – odošle odpoveď. FetchFile - prijme súbor zo súboru, prenesie ho do štvorca PutSftp - umiestni tento súbor na FTP na zadanú adresu. Teraz viac o tomto procese.

Žiadosť je v tomto prípade začiatkom všetkého. Pozrime sa na jeho konfiguračné parametre.

Všetko je tu celkom triviálne s výnimkou StandardHttpContextMap - ide o druh služby, ktorá vám umožňuje odosielať a prijímať požiadavky. Podrobnejšie a dokonca aj s príkladmi môžete vidieť - tu

Ďalej sa pozrime na konfiguračné parametre štvorca ReplaceText. Stojí za to venovať pozornosť ReplacementValue - to je to, čo sa používateľovi vráti vo forme odpovede. V nastaveniach si môžete upraviť úroveň protokolovania, vidíte protokoly {kde ste rozbalili nifi}/nifi-1.9.2/logs, sú tam aj parametre zlyhania/úspechu - na základe týchto parametrov môžete regulovať proces ako celok . To znamená, že v prípade úspešného spracovania textu sa zavolá proces odoslania odpovede používateľovi a v inom prípade neúspešný proces jednoducho zaprotokolujeme.

Vo vlastnostiach HandleHttpResponse nie je nič mimoriadne zaujímavé okrem stavu, keď je odpoveď úspešne vytvorená.

Vyriešili sme požiadavku a odpoveď – prejdime k prijatiu súboru a jeho umiestneniu na FTP server. FetchFile – prijme súbor na ceste zadanej v nastaveniach a odovzdá ho ďalšiemu procesu.

A potom štvorec PutSftp - umiestni súbor do úložiska súborov. Konfiguračné parametre môžeme vidieť nižšie.

Stojí za to venovať pozornosť skutočnosti, že každý štvorec je samostatný proces, ktorý sa musí spustiť. Pozreli sme sa na najjednoduchší príklad, ktorý nevyžaduje žiadne zložité prispôsobenie. Ďalej sa pozrieme na proces trochu komplikovanejšie, kde budeme písať trochu o drážkach.

Zložitejší príklad

Služba prenosu dát spotrebiteľovi sa ukázala byť trochu komplikovanejšia kvôli procesu úpravy SOAP správy. Všeobecný postup je znázornený na obrázku nižšie.

Myšlienka tu tiež nie je obzvlášť zložitá: dostali sme požiadavku od spotrebiteľa, že potrebuje údaje, poslali sme odpoveď, že dostal správu, spustili sme proces prijatia súboru odpovedí, potom sme ho upravili s určitou logikou a potom preniesol súbor spotrebiteľovi vo forme správy SOAP na server.

Myslím, že nie je potrebné znova opisovať tie štvorce, ktoré sme videli vyššie - prejdime rovno k novým. Ak potrebujete upraviť akýkoľvek súbor a bežné štvorce typu ReplaceText nie sú vhodné, budete si musieť napísať vlastný skript. Môžete to urobiť pomocou štvorca ExecuteGroogyScript. Jeho nastavenia sú uvedené nižšie.

Existujú dve možnosti načítania skriptu do tohto štvorca. Prvým je stiahnutie súboru so skriptom. Druhým je vloženie skriptu do scriptBody. Pokiaľ viem, štvorec executeScript podporuje niekoľko jazykov - jeden z nich je groovy. Sklamem vývojárov java - v takýchto štvorcoch nemôžete písať skripty v jave. Pre tých, ktorí naozaj chcú, si musíte vytvoriť svoj vlastný štvorec a pridať ho do systému NIFI. Celú túto operáciu sprevádza pomerne dlhý tanec s tamburínou, ktorému sa v tomto článku nebudeme venovať. Vybral som si groovy jazyk. Nižšie je uvedený testovací skript, ktorý jednoducho prírastkovo aktualizuje id v správe SOAP. Je dôležité poznamenať. Zoberiete súbor z flowFile a aktualizujete ho, nezabudnite, že ho tam musíte vrátiť, aktualizovaný. Je tiež potrebné poznamenať, že nie všetky knižnice sú zahrnuté. Môže sa stať, že ešte budete musieť importovať jednu z knižníc. Ďalšou nevýhodou je, že skript v tomto štvorci sa dosť ťažko ladí. Existuje spôsob, ako sa pripojiť k NIFI JVM a spustiť proces ladenia. Osobne som spustil lokálnu aplikáciu a simuloval príjem súboru z relácie. Lokálne som tiež ladil. Chyby, ktoré sa objavia pri načítaní skriptu, Google pomerne ľahko pozná a do denníka ich zapisuje samotná 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)
}

Vlastne tu sa prispôsobenie námestia končí. Potom sa aktualizovaný súbor prenesie do štvorca, ktorý je zodpovedný za odoslanie súboru na server. Nižšie sú uvedené nastavenia pre tento štvorec.

Opisujeme spôsob, akým sa bude správa SOAP prenášať. Píšeme kde. Ďalej musíte uviesť, že ide o SOAP.

Pridajte niekoľko vlastností, ako je hostiteľ a akcia (soapAction). Uložíme a skontrolujeme. Môžete vidieť ďalšie podrobnosti o tom, ako odosielať požiadavky SOAP tu

Pozreli sme sa na niekoľko možností využitia procesov NIFI. Ako sa vzájomne ovplyvňujú a aký je ich skutočný prínos? Uvažované príklady sú testovacie a mierne sa líšia od toho, čo sa v skutočnosti deje v boji. Dúfam, že tento článok bude pre vývojárov trochu užitočný. Ďakujem za tvoju pozornosť. Ak máte nejaké otázky, napíšte. Pokúsim sa odpovedať.

Zdroj: hab.com