Apache NIFI — Stručný přehled možností v praxi

Úvod

Na mém současném pracovišti jsem měl příležitost se s touto technologií seznámit. Začnu krátkým historickým pozadím. Na jednom z meetingů nám bylo řečeno, že je potřeba vytvořit integraci s známým systémem. Pod integrací bylo myšleno, že tento známý systém nám bude posílat dotazy přes HTTP na určité rozhraní, a my, jak to může znít podivně, budeme posílat zpět odpovědi ve formátu SOAP zprávy. Vypadá to jako jednoduchý a triviální úkol. Z toho vyplývá, že je potřeba…

Úkol

Vytvořit 3 služby. První z nich je Služba pro aktualizaci databáze. Tato služba při příjmu nových dat ze třetí strany aktualizuje údaje v databázi a generuje soubor ve formátu CSV pro jeho předání dalšímu systému. Volá se rozhraní druhé služby — Služby pro transport přes FTP, která přijímá předaný soubor, ověřuje ho a ukládá do úložiště souborů přes FTP. Třetí služba — Služba pro předávání dat koncovému uživateli, pracuje asynchronně s prvními dvěma. Přijímá dotaz od externí třetí strany na soubor, o kterém jsme hovořili dříve, bere hotový soubor odpovědi, modifikuje ho (aktualizuje pole id, description, linkToFile) a posílá odpověď ve formátu SOAP zprávy. Celkově se situace má takto: první dvě služby začínají svou práci teprve, když přijde data pro aktualizaci. Třetí služba pracuje neustále, protože je mnoho uživatelů informací, přibližně 1000 dotazů na příjem dat za minutu. Služby jsou k dispozici neustále a jejich instance se nacházejí v různých prostředích, jako je test, demo, předprod a produkce. Níže je uvedena schéma práce těchto služeb. Ihned si dovoluji objasnit, že některé detaily jsou zjednodušeny, aby se předešlo zbytečné složitosti.

Technické prohloubení

Při plánování řešení úlohy jsme nejprve zvažovali vytvoření aplikací v Javě s využitím frameworku Spring, vyvažovače Nginx, databáze Postgres a dalších technických i netechnických aspektů. Vzhledem k tomu, že čas na vypracování technického řešení dovoloval zvažovat různé přístupy k této úloze, zaměřili jsme se na populární technologii Apache NIFI v určitých kruzích. Hned říkám, že tato technologie nám umožnila identifikovat tyto 3 služby. V tomto článku bude popsána vývoj služby pro přenos souborů a služby pro předávání dat konečnému uživateli, ale pokud článek zaujme, napíšu také o službě pro aktualizaci dat v databázi.

Co to je

NIFI představuje distribuovanou architekturu pro rychlé paralelní načítání a zpracování dat, velké množství pluginů pro zdroje a transformace, verzování konfigurací a mnoho dalšího. Příjemným bonusem je, že je velmi jednoduchý na používání. Triviální procesy, jako je getFile, sendHttpRequest a další, lze vizualizovat jako čtverce. Každý čtverec představuje určitý proces, jehož interakci lze vidět na obrázku níže. Podrobnější dokumentace o interakci a nastavení procesů je napsána zde , pro ty, kteří mluví rusky — zde. V dokumentaci je skvěle popsáno, jak rozbalit a spustit NIFI, a také, jak vytvářet procesy, které jsou také čtverce
Nápad napsat článek se zrodil po dlouhém hledání a strukturování získaných informací do něčeho smysluplného, stejně jako touha trochu usnadnit život budoucím vývojářům.

Příklad

Byly prozkoumány příklady, jak spolu čtverce interagují. Celková schéma je docela jednoduchá: přijmeme HTTP dotaz (teoreticky s souborem v těle dotazu. Pro demonstraci možností NIFI, v tomto příkladu dotaz spouští proces získání souboru z místního úložiště), poté odešleme zpět odpověď, že dotaz byl přijat, současně se spouští proces získání souboru z úložiště a dále proces jeho přesunutí přes FTP do úložiště. Je třeba vysvětlit, že procesy spolu komunikují prostřednictvím takzvaného flowFile. Toto je základní entita v NIFI, která uchovává atributy a obsah. Obsah — jsou data, která jsou reprezentována souborem toku. To znamená, že pokud získáte soubor z jednoho čtverce a předáváte ho do druhého, obsahem bude váš soubor.

Jak můžete vidět — na tomto obrázku je znázorněn celkový proces. HandleHttpRequest — přijímá dotazy, ReplaceText — generuje tělo odpovědi, HandleHttpResponse — vrací odpověď. FetchFile — získává soubor z úložiště a předává ho čtverci PutSftp — ukládá tento soubor na FTP na určenou adresu. Nyní se podívejme podrobněji na tento proces.

V tomto případě — dotaz je všemu začátkem. Podívejme se na jeho konfigurační parametry.

Zde je to celkem triviální s výjimkou StandartHttpContextMap — je to jistá služba, která umožňuje odesílat a přijímat dotazy. Podrobnější informace a dokonce i s příklady si můžete zobrazit — zde

Dále se podívejme na konfigurační parametry čtverce ReplaceText. Je důležité věnovat pozornost ReplacementValue — to je to, co se vrátí uživateli ve formě odpovědi. V nastavení lze regulovat úroveň logování, logy si můžete prohlédnout {kam jsme rozbalili nifi}/nifi-1.9.2/logs, tam jsou také parametry failure/success — na základě těchto parametrů lze regulovat celý proces. Tedy v případě úspěšného zpracování textu — vyvolá se proces odeslání odpovědi uživateli, a v opačném případě prostě zaznamenáme neúspěšný proces.

Ve vlastnostech HandleHttpResponse není nic zajímavého, kromě stavu při úspěšném vytvoření odpovědi.

S dotazem a odpovědí jsme se seznámili — pojďme dál k získání souboru a jeho umístění na FTP server. FetchFile — získává soubor podle cesty uvedené v nastavení a předává ho do dalšího procesu.

A dále čtverec PutSftp - vkládá soubor do úložiště. Konfigurační parametry vidíme níže.

Je třeba si všimnout, že každý čtverec je samostatný proces, který musí být spuštěn. Prozkoumali jsme nejjednodušší příklad, který nevyžaduje žádnou složitou přizpůsobení. Dále se podíváme na proces, který je trochu složitější, kde si pár věcí napíšeme v groovy.

Složitější příklad

Služba přenosu dat k zákazníkovi se ukázala jako o něco složitější díky procesu modifikace zprávy SOAP. Celkový proces je znázorněn na obrázku níže.

Zde je myšlenka také docela jednoduchá: obdrželi jsme požadavek od zákazníka, že potřebuje data, poslali jsme odpověď, že jsme obdrželi zprávu, spustili jsme proces získání odpovědního souboru, dále jsme jej upravili podle určité logiky a poté jsme předali soubor zákazníkovi ve formě zprávy SOAP na server.

Myslím, že není třeba znovu popisovat čtverce, které jsme viděli výše - přejdeme rovnou k novým. Pokud potřebujete upravit jakýkoli soubor a běžné čtverce typu ReplaceText nejsou vhodné, budete muset napsat svůj skript. To lze provést pomocí čtverce ExecuteGroogyScript. Jeho nastavení jsou uvedena níže.

Existují dva způsoby, jak načíst skript do tohoto bloku. První je pomocí načtení souboru se skriptem. Druhý je vložením skriptu do scriptBody. Dle mých informací blok executeScript podporuje několik programovacích jazyků, jedním z nich je groovy. Zklamu vývojáře v javě — ve javě nelze psát skripty v těchto blocích. Pro ty, kdo si to opravdu přejí, je třeba vytvořit svůj vlastní vlastní blok a vložit ho do systému NIFI. Celá tato operace je doprovázena poměrně dlouhými tanečky s bubnem, kterými se v rámci tohoto článku nezabýváme. Zvolila jsem jazyk groovy. Níže je uveden testovací skript, který jednoduše inkrementálně aktualizuje id v SOAP zprávě. Je důležité poznamenat, že si berete soubor z flowFile, aktualizujete ho a nesmíte zapomenout, že ho musíte vrátit zpět. Také je třeba poznamenat, že nejsou připojeny všechny knihovny. Může se stát, že budete muset importovat jednu z knihoven. Dalším nevýhodou je, že skript v tomto bloku je docela těžko ladit. Existuje způsob, jak se připojit k JVM NIFI a začít proces ladění. Osobně jsem spouštěla aplikaci lokálně a imitovala jsem získání souboru ze session. Laděním jsem se také zabývala lokálně. Chyby, které se při načítání skriptu objevují, se dají snadno hledat na Googlu a NIFI je zapisuje 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("Error during processing of validate.groovy", e)
    session.transfer(flowFile, REL_FAILURE)
}

Tímto vlastně končí přizpůsobení bloku. Aktualizovaný soubor je pak předán bloku, který se zabývá odesíláním souboru na server. Níže jsou uvedeny nastavení tohoto bloku.

Popisujeme metodu, kterou bude zasíláno SOAP zpráva. Uvádíme, kam. Dále je třeba uvést, že se jedná o SOAP.

Přidáváme několik vlastností, jako je hostitel a akce (soapAction). Ukládáme a kontrolujeme. Podrobněji o tom, jak posílat SOAP požadavky, se můžete podívat. tady

Prozkoumali jsme několik možností využití procesů NIFI. Jak spolu interagují a jaký mají skutečný přínos. Uvedené příklady jsou testovací a mírně se liší od toho, co se doopravdy používá v produkci. Doufám, že tento článek bude pro vývojáře trochu užitečný. Děkuji za pozornost. Pokud máte nějaké dotazy – pište. Pokusím se odpovědět.

Zdroj: habr.com