🥇Apache NIFI - Yon ti apèsi sou kapasite yo nan pratik

Entwodiksyon

Se konsa, nan plas travay mwen ye kounye a, mwen te oblije fè konesans ak teknoloji sa a. Mwen pral kòmanse ak yon ti background. Nan pwochen reyinyon an, yo te di ekip nou an ke nou bezwen kreye entegrasyon ak sistèm li te ye. Pa entegrasyon li te vle di ke sistèm byen li te ye sa a ta voye demann nou atravè HTTP nan yon pwen final espesifik, epi nou, etranj ase, ta voye tounen repons nan fòm lan nan yon mesaj SOAP. Tout bagay sanble senp ak trivial. Apati sa a li swiv ke ou bezwen ...

Objektif Travay la

Kreye 3 sèvis. Premye nan yo se Sèvis Mizajou Baz done. Sèvis sa a, lè nouvo done rive nan yon sistèm twazyèm pati, mete ajou done yo nan baz done a epi jenere yon dosye nan fòma CSV pou transfere li nan pwochen sistèm nan. Yo rele pwen final la nan dezyèm sèvis la - Sèvis Transpò FTP, ki resevwa fichye transfere a, valide li, epi mete l nan depo dosye atravè FTP. Twazyèm sèvis la, sèvis transfè done konsomatè, travay asynchrone ak de premye yo. Li resevwa yon demann ki soti nan yon twazyèm pati sistèm ekstèn pou resevwa dosye a diskite pi wo a, pran dosye a repons pare, modifye li (mete ajou id, deskripsyon, jaden linkToFile) epi voye repons lan nan fòm lan nan yon mesaj SOAP. Sa vle di, foto an jeneral se jan sa a: de premye sèvis yo kòmanse travay yo sèlman lè done yo pou mete ajou te rive. Twazyèm sèvis la ap travay toujou paske gen anpil konsomatè enfòmasyon, apeprè 1000 demann pou done pou chak minit. Sèvis yo disponib toutan e ka yo sitiye nan diferan anviwònman, tankou tès, Demo, pre-pwodiksyon ak prod. Anba a se yon dyagram sou fason sèvis sa yo fonksyone. Kite m 'klarifye touswit ke kèk detay yo te senplifye pou evite konpleksite nesesè.

Apwofondi teknik

Lè w ap planifye yon solisyon a pwoblèm nan, nou premye deside fè aplikasyon nan Java lè l sèvi avèk fondasyon an Spring, Nginx balans, baz done Postgres ak lòt bagay teknik ak pa tèlman teknik. Depi lè pou devlope yon solisyon teknik pèmèt nou konsidere lòt apwòch pou rezoud pwoblèm sa a, gade nou te tonbe sou teknoloji Apache NIFI, ki alamòd nan sèten sèk. Mwen pral di touswit ke teknoloji sa a pèmèt nou remake 3 sèvis sa yo. Atik sa a pral dekri devlopman yon sèvis transpò dosye ak yon sèvis transfè done bay konsomatè a, men si atik la itil, mwen pral ekri sou sèvis la pou mete ajou done nan baz done a.

Ki sa li ye?

NIFI se yon achitekti distribiye pou chaje rapid paralèl ak pwosesis done, yon gwo kantite grefon pou sous ak transfòmasyon, vèsyon konfigirasyon ak plis ankò. Yon bonis bèl se ke li trè fasil yo sèvi ak. Pwosesis trivial tankou getFile, sendHttpRequest ak lòt moun ka reprezante kòm kare. Chak kare reprezante yon pwosesis, entèraksyon ki ka wè nan figi ki anba a. Plis detay dokiman sou entèraksyon pwosesis konfigirasyon yo te ekri isit la , pou moun ki pale Ris - isit la. Dokimantasyon an dekri parfe kijan pou depale epi kouri NIFI, ansanm ak fason pou kreye pwosesis, ke yo rele tou kare.
Lide a ekri yon atik te fèt apre yon rechèch long ak estriktire enfòmasyon yo resevwa nan yon bagay konsyan, osi byen ke dezi a fè lavi yon ti kras pi fasil pou devlopè nan lavni.

Egzanp

Yo konsidere yon egzanp sou fason kare yo kominike youn ak lòt. Konplo jeneral la se byen senp: Nou resevwa yon demann HTTP (An teyori, ak yon fichye nan kò a nan demann lan. Pou demontre kapasite yo nan NIFI, nan egzanp sa a demann lan kòmanse pwosesis pou resevwa yon dosye nan depo dosye lokal la. ), Lè sa a, nou voye tounen yon repons ke demann lan te resevwa, an paralèl pwosesis pou resevwa yon fichye soti nan FH ak Lè sa a, pwosesis la nan deplase li atravè FTP nan FH. Li vo klarifye ke pwosesis yo kominike youn ak lòt atravè sa yo rele flowFile la. Sa a se antite de baz nan NIFI ki estoke atribi ak kontni. Kontni se done ki reprezante pa fichye kouran an. Sa vle di, apeprè pale, si ou resevwa yon dosye ki soti nan yon kare epi transfere li nan yon lòt, kontni an pral fichye ou a.

Kòm ou ka wè, foto sa a montre pwosesis jeneral la. HandleHttpRequest - aksepte demann, ReplaceText - jenere yon kò repons, HandleHttpResponse - voye yon repons. FetchFile - resevwa yon dosye ki soti nan yon depo dosye, transfere li nan kare PutSftp - mete fichye sa a sou FTP, nan adrès ki espesifye. Koulye a, plis sou pwosesis sa a.

Nan ka sa a, demann se kòmansman tout bagay. Ann gade nan paramèt konfigirasyon li yo.

Tout bagay isit la trè trivial eksepte StandardHttpContextMap - sa a se yon kalite sèvis ki pèmèt ou voye ak resevwa demann. Nan plis detay e menm ak egzanp, ou ka wè - isit la

Apre sa, ann gade nan paramèt konfigirasyon ReplaceText kare a. Li vo peye atansyon sou ReplacementValue - sa a se sa ki pral retounen bay itilizatè a nan fòm lan nan yon repons. Nan anviwònman ou ka ajiste nivo a nan antre, ou ka wè mòso bwa yo {kote ou depake nifi}/nifi-1.9.2/mòso bwa, gen tou paramèt echèk/siksè - ki baze sou paramèt sa yo ou ka kontwole pwosesis la kòm yon antye. . Sa vle di, nan ka a nan pwosesis tèks siksè, pwosesis la nan voye yon repons bay itilizatè a pral rele, ak nan yon lòt ka nou pral tou senpleman konekte pwosesis la san siksè.

Pa gen anyen patikilyèman enteresan nan pwopriyete HandleHttpResponse eksepte estati a lè yo kreye yon repons avèk siksè.

Nou te regle demann lan ak repons - ann ale nan resevwa fichye a epi mete l sou sèvè FTP la. FetchFile - resevwa yon fichye nan chemen ki espesifye nan anviwònman yo epi li pase li nan pwochen pwosesis la.

Lè sa a, kare PutSftp la - mete dosye a nan depo dosye a. Nou ka wè paramèt konfigirasyon yo anba a.

Li vo peye atansyon sou lefèt ke chak kare se yon pwosesis separe ki dwe lanse. Nou te gade egzanp ki pi senp ki pa mande pou okenn personnalisation konplèks. Apre sa, nou pral gade nan pwosesis la yon ti kras pi konplike, kote nou pral ekri yon ti kras sou genyen siyon yo.

Egzanp pi konplèks

Sèvis transfè done bay konsomatè a te vin yon ti kras pi konplike akòz pwosesis modifye mesaj SOAP la. Pwosesis jeneral la montre nan figi ki anba a.

Isit la lide a tou pa patikilyèman konplike: nou te resevwa yon demann nan men konsomatè a ke li te bezwen done, voye yon repons ke li te resevwa yon mesaj, te kòmanse pwosesis pou resevwa fichye repons lan, Lè sa a, edited li ak yon lojik sèten, ak Lè sa a transfere fichye a bay konsomatè a nan fòm yon mesaj SOAP sou sèvè a.

Mwen panse ke pa gen okenn nesesite dekri ankò kare sa yo ke nou te wè pi wo a - ann ale tou dwat nan nouvo yo. Si ou bezwen edite nenpòt ki fichye ak òdinè kare ReplaceText kalite yo pa apwopriye, w ap oblije ekri pwòp script ou. Sa a ka fè lè l sèvi avèk kare ExecuteGroogyScript la. Anviwònman li yo prezante anba a.

Gen de opsyon pou chaje script la nan kare sa a. Premye a se pa telechaje yon fichye ak yon script. Dezyèm lan se lè w mete yon script nan scriptBody. Dapre mwen konnen, kare executeScript la sipòte plizyè lang - youn nan yo se groovy. Mwen pral desevwa devlopè java - ou pa ka ekri scripts nan java nan kare sa yo. Pou moun ki reyèlman vle, ou bezwen kreye pwòp kare koutim ou epi ajoute li nan sistèm NIFI la. Tout operasyon sa a akonpaye pa yon dans byen long ak yon tanbourin, ki nou pa pral fè fas ak nan atik sa a. Mwen te chwazi lang groovy la. Anba a se yon script tès ki tou senpleman mete ajou id la nan yon mesaj SOAP. Li enpòtan sonje. Ou pran fichye a soti nan flowFile epi mete ajou li, pa bliye ke ou bezwen mete l tounen la, mete ajou. Li se tou vo sonje ke se pa tout bibliyotèk yo enkli. Li ka rive ke ou toujou gen enpòte youn nan lib yo. Yon lòt dezavantaj se ke script la nan kare sa a se byen difisil debogaj. Gen yon fason pou konekte ak NIFI JVM epi kòmanse pwosesis debogaj la. Pèsonèlman, mwen te lanse yon aplikasyon lokal ak simulation resevwa yon dosye nan sesyon an. Mwen menm mwen te fè debogaj lokalman. Erè ki parèt lè w ap chaje yon script se byen fasil pou Google epi yo ekri pa NIFI tèt li nan boutèy la.

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

Aktyèlman, sa a se kote personnalisation nan kare a fini. Apre sa, mete ajou dosye a transfere nan kare a, ki responsab pou voye fichye a nan sèvè a. Anba a se anviwònman yo pou kare sa a.

Nou dekri metòd pou yon mesaj SOAP pral transmèt. Nou ekri kote. Apre sa, ou bezwen endike ke sa a se SAVON.

Ajoute plizyè pwopriyete tankou lame ak aksyon (soapAction). Nou sove epi tcheke. Ou ka wè plis detay sou fason pou voye demann SOAP isit la

Nou te gade plizyè opsyon pou itilize pwosesis NIFI. Ki jan yo kominike ak ki benefis reyèl yo? Egzanp yo konsidere yo se tès yo epi yo yon ti kras diferan de sa ki aktyèlman k ap pase nan konba. Mwen espere atik sa a pral yon ti kras itil pou devlopè. Mèsi pou atansyon ou. Si w gen nenpòt kesyon, ekri. Mwen pral eseye reponn.

Sous: www.habr.com

Apache NIFI - Yon Brèf Apèsi sou Opòtinite nan Pratik