Apache NIFI - Isang Maikling Pangkalahatang-ideya ng Mga Tampok sa Practice

Pagpapakilala

Ito ay nangyari na sa aking kasalukuyang lugar ng trabaho kailangan kong makilala ang teknolohiyang ito. Magsisimula ako sa isang maliit na background. Sa susunod na pagpupulong, sinabi sa aming koponan na kailangan naming lumikha ng pagsasama sa kilalang sistema. Sa pamamagitan ng pagsasama, sinadya na ang kilalang sistemang ito ay magpapadala sa amin ng mga kahilingan sa pamamagitan ng HTTP sa isang partikular na endpoint, at kami, kakaiba, ay magpapadala ng mga tugon sa anyo ng isang SOAP na mensahe. Ang lahat ay tila simple at walang kuwenta. Mula dito ay sumusunod na kailangan mo...

Gawain

Lumikha ng 3 serbisyo. Ang una sa kanila ay ang Database Update Service. Ang serbisyong ito, kapag dumating ang bagong data mula sa isang third-party na system, ina-update ang data sa database at bubuo ng file sa CSV format para ilipat ito sa susunod na system. Ang endpoint ng pangalawang serbisyo ay tinatawag - ang FTP Transport Service, na tumatanggap ng inilipat na file, nagpapatunay nito, at inilalagay ito sa imbakan ng file sa pamamagitan ng FTP. Ang pangatlong serbisyo, ang serbisyo sa paglilipat ng data ng consumer, ay gumagana nang asynchronous sa unang dalawa. Nakatanggap ito ng kahilingan mula sa isang third-party na panlabas na system na matanggap ang file na tinalakay sa itaas, kunin ang handa na tugon na file, binabago ito (ina-update ang id, paglalarawan, linkToFile na mga field) at ipinapadala ang tugon sa anyo ng isang SOAP na mensahe. Iyon ay, ang pangkalahatang larawan ay ang mga sumusunod: ang unang dalawang serbisyo ay magsisimula lamang sa kanilang trabaho kapag dumating na ang data para sa pag-update. Ang ikatlong serbisyo ay patuloy na gumagana dahil maraming mga mamimili ng impormasyon, tungkol sa 1000 mga kahilingan para sa data bawat minuto. Ang mga serbisyo ay palaging magagamit at ang kanilang mga instance ay matatagpuan sa iba't ibang mga kapaligiran, tulad ng pagsubok, demo, pre-production at prod. Nasa ibaba ang isang diagram kung paano gumagana ang mga serbisyong ito. Hayaan akong linawin kaagad na ang ilang mga detalye ay pinasimple upang maiwasan ang hindi kinakailangang kumplikado.

Teknikal na Pagpapalalim

Kapag nagpaplano ng solusyon sa problema, nagpasya muna kaming gumawa ng mga application sa Java gamit ang Spring framework, Nginx balancer, Postgres database at iba pang teknikal at hindi masyadong teknikal na mga bagay. Dahil ang oras upang bumuo ng isang teknikal na solusyon ay nagpapahintulot sa amin na isaalang-alang ang iba pang mga diskarte sa paglutas ng problemang ito, ang aming tingin ay nahulog sa Apache NIFI na teknolohiya, na kung saan ay sunod sa moda sa ilang mga lupon. Sasabihin ko kaagad na pinahintulutan kami ng teknolohiyang ito na mapansin ang 3 serbisyong ito. Ilalarawan ng artikulong ito ang pagbuo ng isang serbisyo sa transportasyon ng file at isang serbisyo sa paglilipat ng data sa consumer, ngunit kung kapaki-pakinabang ang artikulo, isusulat ko ang tungkol sa serbisyo para sa pag-update ng data sa database.

Ano ito

Ang NIFI ay isang distributed architecture para sa mabilis na parallel loading at processing ng data, isang malaking bilang ng mga plugin para sa mga source at transformation, versioning ng mga configuration at marami pang iba. Ang isang magandang bonus ay na ito ay napakadaling gamitin. Ang mga trivial na proseso tulad ng getFile, sendHttpRequest at iba pa ay maaaring katawanin bilang mga parisukat. Ang bawat parisukat ay kumakatawan sa isang proseso, ang pakikipag-ugnayan nito ay makikita sa figure sa ibaba. Naisulat ang mas detalyadong dokumentasyon sa mga pakikipag-ugnayan sa pag-setup ng proseso dito , para sa mga nagsasalita ng Russian - dito. Ang dokumentasyon ay perpektong naglalarawan kung paano i-unpack at patakbuhin ang NIFI, pati na rin kung paano lumikha ng mga proseso, na kilala rin bilang mga parisukat
Ang ideya na magsulat ng isang artikulo ay ipinanganak pagkatapos ng mahabang paghahanap at pag-istruktura ng impormasyong natanggap sa isang bagay na may kamalayan, pati na rin ang pagnanais na gawing mas madali ang buhay para sa mga susunod na developer..

Halimbawa

Isang halimbawa kung paano nakikipag-ugnayan ang mga parisukat sa isa't isa ay isinasaalang-alang. Ang pangkalahatang pamamaraan ay medyo simple: Nakatanggap kami ng isang kahilingan sa HTTP (Sa teorya, na may isang file sa katawan ng kahilingan. Upang ipakita ang mga kakayahan ng NIFI, sa halimbawang ito ang kahilingan ay nagsisimula sa proseso ng pagtanggap ng isang file mula sa lokal na imbakan ng file ), pagkatapos ay nagpapadala kami ng tugon na ang kahilingan ay natanggap, kasabay ng proseso ng pagtanggap ng isang file mula sa FH at pagkatapos ay ang proseso ng paglipat nito sa pamamagitan ng FTP sa FH. Ito ay nagkakahalaga ng paglilinaw na ang mga proseso ay nakikipag-ugnayan sa isa't isa sa pamamagitan ng tinatawag na flowFile. Ito ang batayang entity sa NIFI na nag-iimbak ng mga katangian at nilalaman. Ang nilalaman ay ang data na kinakatawan ng stream file. Iyon ay, halos nagsasalita, kung nakatanggap ka ng isang file mula sa isang parisukat at ilipat ito sa isa pa, ang nilalaman ay iyong magiging file.

Tulad ng nakikita mo, ipinapakita ng larawang ito ang pangkalahatang proseso. HandleHttpRequest - tumatanggap ng mga kahilingan, ReplaceText - bumubuo ng katawan ng tugon, HandleHttpResponse - nagpapadala ng tugon. FetchFile - tumatanggap ng file mula sa isang imbakan ng file, inilipat ito sa parisukat na PutSftp - inilalagay ang file na ito sa FTP, sa tinukoy na address. Ngayon higit pa tungkol sa prosesong ito.

Sa kasong ito, ang kahilingan ang simula ng lahat. Tingnan natin ang mga parameter ng pagsasaayos nito.

Ang lahat dito ay medyo walang halaga maliban sa StandardHttpContextMap - ito ay isang uri ng serbisyo na nagpapahintulot sa iyo na magpadala at tumanggap ng mga kahilingan. Sa mas detalyado at kahit na may mga halimbawa, makikita mo - dito

Susunod, tingnan natin ang mga parameter ng pagsasaayos ng ReplaceText ng parisukat. Ito ay nagkakahalaga ng pagbibigay pansin sa ReplacementValue - ito ang ibabalik sa gumagamit sa anyo ng isang tugon. Sa mga setting maaari mong ayusin ang antas ng pag-log, makikita mo ang mga log {where you unpacked nifi}/nifi-1.9.2/logs, mayroon ding failure/success parameters - batay sa mga parameter na ito maaari mong i-regulate ang proseso sa kabuuan . Iyon ay, sa kaso ng matagumpay na pagpoproseso ng teksto, ang proseso ng pagpapadala ng tugon sa gumagamit ay tatawagin, at sa ibang kaso ay i-log lang namin ang hindi matagumpay na proseso.

Walang partikular na interesante sa mga katangian ng HandleHttpResponse maliban sa status kapag matagumpay na nalikha ang isang tugon.

Inayos namin ang kahilingan at tugon - magpatuloy tayo sa pagtanggap ng file at ilagay ito sa FTP server. FetchFile - tumatanggap ng file sa path na tinukoy sa mga setting at ipapasa ito sa susunod na proseso.

At pagkatapos ay ang PutSftp square - inilalagay ang file sa imbakan ng file. Makikita natin ang mga parameter ng pagsasaayos sa ibaba.

Ito ay nagkakahalaga ng pagbibigay pansin sa katotohanan na ang bawat parisukat ay isang hiwalay na proseso na dapat ilunsad. Tiningnan namin ang pinakasimpleng halimbawa na hindi nangangailangan ng anumang kumplikadong pagpapasadya. Susunod, titingnan natin ang proseso nang medyo mas kumplikado, kung saan magsusulat tayo ng kaunti sa mga grooves.

Mas kumplikadong halimbawa

Ang serbisyo ng paglilipat ng data sa mamimili ay naging medyo mas kumplikado dahil sa proseso ng pagbabago ng mensahe ng SOAP. Ang pangkalahatang proseso ay ipinapakita sa figure sa ibaba.

Narito ang ideya ay hindi rin partikular na kumplikado: nakatanggap kami ng kahilingan mula sa mamimili na kailangan niya ng data, nagpadala ng tugon na nakatanggap siya ng mensahe, sinimulan ang proseso ng pagtanggap ng file ng tugon, pagkatapos ay na-edit ito gamit ang isang tiyak na lohika, at pagkatapos inilipat ang file sa consumer sa anyo ng SOAP message sa server.

Sa palagay ko hindi na kailangang ilarawan muli ang mga parisukat na nakita natin sa itaas - dumiretso tayo sa mga bago. Kung kailangan mong i-edit ang anumang file at hindi angkop ang ordinaryong mga parisukat na uri ng ReplaceText, kakailanganin mong magsulat ng iyong sariling script. Magagawa ito gamit ang ExecuteGroogyScript square. Ang mga setting nito ay ipinakita sa ibaba.

Mayroong dalawang mga pagpipilian para sa pag-load ng script sa parisukat na ito. Ang una ay sa pamamagitan ng pag-download ng file na may script. Ang pangalawa ay sa pamamagitan ng pagpasok ng isang script sa scriptBody. Sa pagkakaalam ko, ang executeScript square ay sumusuporta sa ilang mga wika - isa sa mga ito ay groovy. Biguin ko ang mga developer ng java - hindi ka maaaring magsulat ng mga script sa java sa mga parisukat. Para sa mga talagang gusto, kailangan mong lumikha ng iyong sariling pasadyang parisukat at idagdag ito sa sistema ng NIFI. Ang buong operasyon na ito ay sinamahan ng medyo mahabang sayaw na may tamburin, na hindi namin haharapin sa artikulong ito. Pinili ko ang groovy na wika. Nasa ibaba ang isang pansubok na script na unti-unting nag-a-update ng id sa isang SOAP na mensahe. Mahalagang tandaan. Kinuha mo ang file mula sa flowFile at i-update ito, huwag kalimutan na kailangan mong ibalik ito doon, na-update. Dapat ding tandaan na hindi lahat ng mga aklatan ay kasama. Maaaring mangyari na kailangan mo pa ring mag-import ng isa sa mga libs. Ang isa pang downside ay ang script sa parisukat na ito ay medyo mahirap i-debug. Mayroong isang paraan upang kumonekta sa NIFI JVM at simulan ang proseso ng pag-debug. Sa personal, naglunsad ako ng isang lokal na aplikasyon at na-simulate ang pagtanggap ng isang file mula sa session. Nag-debug din ako nang lokal. Ang mga error na lumilitaw kapag naglo-load ng script ay medyo madali sa Google at isinulat mismo ng NIFI sa log.

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)
}

Sa totoo lang, dito nagtatapos ang pagpapasadya ng parisukat. Susunod, ang na-update na file ay inilipat sa parisukat, na responsable para sa pagpapadala ng file sa server. Nasa ibaba ang mga setting para sa parisukat na ito.

Inilalarawan namin ang paraan kung saan ipapadala ang isang SOAP na mensahe. Nagsusulat kami kung saan. Susunod na kailangan mong ipahiwatig na ito ay SOAP.

Magdagdag ng ilang property gaya ng host at action (soapAction). I-save at suriin namin. Makakakita ka ng higit pang mga detalye kung paano magpadala ng mga kahilingan sa SOAP dito

Tumingin kami sa ilang mga opsyon para sa paggamit ng mga proseso ng NIFI. Paano sila nakikipag-ugnayan at ano ang kanilang tunay na pakinabang? Ang mga halimbawang isinasaalang-alang ay mga pagsubok at bahagyang naiiba sa kung ano ang aktwal na nangyayari sa labanan. Umaasa ako na ang artikulong ito ay medyo kapaki-pakinabang para sa mga developer. Salamat sa iyong atensyon. Kung mayroon kang anumang mga katanungan, sumulat. Susubukan kong sagutin.

Pinagmulan: www.habr.com