Apache NIFI - Ringkesan Singkat Fitur ing Praktek

Pambuka

Kedaden ing papan kerja saiki aku kudu kenal karo teknologi iki. Aku bakal miwiti karo latar mburi sethitik. Ing rapat sabanjure, tim kita dikandhani yen kita kudu nggawe integrasi karo sistem dikenal. Kanthi integrasi, tegese sistem sing kondhang iki bakal ngirim panjaluk liwat HTTP menyang titik pungkasan tartamtu, lan kita, aneh, bakal ngirim tanggapan maneh ing bentuk pesen SOAP. Kabeh katon prasaja lan ora pati penting. Saka iki, sampeyan kudu ...

Tujuan

Nggawe 3 layanan. Sing pertama yaiku Layanan Update Database. Layanan iki, nalika data anyar teka saka sistem pihak katelu, nganyari data ing database lan ngasilake file ing format CSV kanggo nransfer menyang sistem sabanjuré. Titik pungkasan saka layanan kapindho diarani - Layanan Transportasi FTP, sing nampa file sing ditransfer, validasi, lan dilebokake ing panyimpenan file liwat FTP. Layanan katelu, layanan transfer data konsumen, bisa digunakake kanthi ora sinkron karo loro pisanan. Nampa panjalukan saka sistem eksternal pihak katelu kanggo nampa file sing dibahas ing ndhuwur, njupuk file respon siap, ngowahi (nganyari id, katrangan, kolom linkToFile) lan ngirim respon ing wangun pesen SOAP. Sing, gambar sakabèhé minangka nderek: loro layanan pisanan miwiti karya mung nalika data kanggo nganyari wis teka. Layanan katelu dianggo terus amarga ana akeh konsumen informasi, bab 1000 panjalukan kanggo data saben menit. Layanan kasedhiya terus-terusan lan kedadeyane ana ing lingkungan sing beda-beda, kayata tes, demo, pra-produksi lan prod. Ing ngisor iki diagram babagan cara kerja layanan kasebut. Ayo kula njlentrehake langsung yen sawetara rincian wis disederhanakake supaya ora kerumitan sing ora perlu.

Teknik Deepening

Nalika ngrancang solusi kanggo masalah kasebut, mula kita mutusake nggawe aplikasi ing Jawa nggunakake kerangka Spring, Nginx balancer, database Postgres lan teknis liyane lan ora teknis. Wiwit wektu kanggo ngembangake solusi teknis ngidini kita nimbang pendekatan liyane kanggo ngrampungake masalah iki, pandangan kita tiba ing teknologi Apache NIFI, sing modis ing kalangan tartamtu. Aku bakal langsung ngomong manawa teknologi iki ngidini kita ngerteni 3 layanan kasebut. Artikel iki bakal njlèntrèhaké pangembangan layanan transportasi file lan layanan transfer data menyang konsumen, nanging yen artikel migunani, aku bakal nulis babagan layanan kanggo nganyari data ing database.

Apa iku?

NIFI minangka arsitektur sing disebarake kanggo loading lan pangolahan data paralel kanthi cepet, akeh plugin kanggo sumber lan transformasi, versi konfigurasi lan liya-liyane. A bonus becik iku gampang banget kanggo nggunakake. Proses sepele kayata getFile, sendHttpRequest lan liya-liyane bisa diwakili minangka kotak. Saben kothak nggambarake proses, interaksi sing bisa dideleng ing gambar ing ngisor iki. Dokumentasi sing luwih rinci babagan interaksi persiyapan proses wis ditulis kene , kanggo sing nganggo basa Rusia - kene. Dokumentasi kanthi sampurna nggambarake carane mbongkar lan mbukak NIFI, uga cara nggawe proses, uga dikenal minangka kotak.
Gagasan kanggo nulis artikel lair sawise panelusuran dawa lan struktur informasi sing ditampa menyang soko sadar, uga kepinginan kanggo nggawe urip luwih gampang kanggo pangembang mangsa.

Conto:

Conto carane kothak sesambungan karo saben liyane dianggep. Skema umum cukup prasaja: Kita nampa panjalukan HTTP (Ing teori, kanthi file ing awak panyuwunan. Kanggo nduduhake kemampuan NIFI, ing conto iki panyuwunan miwiti proses nampa file saka panyimpenan file lokal. ), banjur kita ngirim maneh respon yen panyuwunan wis ditampa, kanthi podo karo proses nampa file saka FH lan banjur proses mindhah liwat FTP menyang FH. Sampeyan kudu njlentrehake manawa proses sesambungan karo siji liyane liwat sing diarani flowFile. Iki minangka entitas dhasar ing NIFI sing nyimpen atribut lan konten. Isi minangka data sing diwakili dening file stream. Tegese, kira-kira, yen sampeyan nampa file saka siji kothak lan ditransfer menyang liyane, isi kasebut bakal dadi file sampeyan.

Nalika sampeyan bisa ndeleng, gambar iki nuduhake proses umum. HandleHttpRequest - nampa panjalukan, ReplaceText - ngasilake awak respon, HandleHttpResponse - ngirim respon. FetchFile - nampa file saka panyimpenan file, pindhah menyang kothak PutSftp - sijine file iki ing FTP, ing alamat sing ditemtokake. Saiki luwih akeh babagan proses iki.

Ing kasus iki, panjalukan minangka wiwitan kabeh. Ayo katon ing paramèter konfigurasi.

Kabeh ing kene cukup sepele kajaba StandardHttpContextMap - iki minangka layanan sing ngidini sampeyan ngirim lan nampa panjaluk. Kanthi luwih rinci lan malah nganggo conto, sampeyan bisa ndeleng - kene

Sabanjure, ayo goleki parameter konfigurasi ReplaceText saka alun-alun. Iku worth mbayar manungsa waé kanggo ReplacementValue - iki apa bakal bali menyang pangguna ing wangun respon. Ing setelan sampeyan bisa nyetel tingkat logging, sampeyan bisa ndeleng log {ngendi sampeyan unpacked nifi}/nifi-1.9.2/logs, ana uga gagal / paramèter sukses - adhedhasar paramèter iki sampeyan bisa ngatur proses minangka kabèh. . Yaiku, ing kasus pangolahan teks sing sukses, proses ngirim respon menyang pangguna bakal diarani, lan ing kasus liyane, kita mung bakal mlebu proses sing ora kasil.

Ora ana sing menarik banget ing sifat HandleHttpResponse kajaba status nalika respon kasil digawe.

We wis diurutake metu panjalukan lan respon - ayo pindhah menyang nampa file lan manggonke ing server FTP. FetchFile - nampa file ing path sing ditemtokake ing setelan lan diterusake menyang proses sabanjure.

Banjur kothak PutSftp - nyelehake file ing panyimpenan file. Kita bisa ndeleng paramèter konfigurasi ing ngisor iki.

Perlu dicathet yen saben kothak minangka proses kapisah sing kudu diluncurake. Kita ndeleng conto paling gampang sing ora mbutuhake kustomisasi rumit. Sabanjure, kita bakal katon ing proses sethitik liyane rumit, ngendi kita bakal nulis sethitik ing grooves.

Tuladha sing luwih kompleks

Layanan transfer data menyang konsumen dadi rada rumit amarga proses ngowahi pesen SOAP. Proses umum ditampilake ing gambar ing ngisor iki.

Ing kene, ide kasebut uga ora rumit banget: kita nampa panjaluk saka konsumen yen dheweke butuh data, ngirim tanggapan yen dheweke wis nampa pesen, miwiti proses nampa file respon, banjur diowahi kanthi logika tartamtu, banjur ditransfer file menyang konsumen ing wangun pesen SOAP menyang server.

Aku ora perlu njlèntrèhaké maneh kothak sing kita deleng ing ndhuwur - ayo langsung menyang sing anyar. Yen sampeyan kudu ngowahi file apa wae lan kothak jinis ReplaceText biasa ora cocog, sampeyan kudu nulis skrip sampeyan dhewe. Iki bisa ditindakake kanthi nggunakake kothak ExecuteGroogyScript. Setelan kasebut ditampilake ing ngisor iki.

Ana rong pilihan kanggo mbukak script menyang kothak iki. Kapisan yaiku kanthi ngundhuh file kanthi skrip. Kapindho yaiku kanthi nglebokake skrip menyang scriptBody. Sa adoh aku ngerti, kothak executeScript ndhukung sawetara basa - salah sijine yaiku groovy. Aku bakal nguciwani pangembang java - sampeyan ora bisa nulis skrip ing java ing kothak kasebut. Kanggo sing pengin tenan, sampeyan kudu nggawe kothak adat dhewe lan nambah menyang sistem NIFI. Kabeh operasi iki diiringi tarian sing cukup dawa karo rebana, sing ora bakal ditindakake ing artikel iki. Aku milih basa groovy. Ing ngisor iki minangka skrip test sing mung nambahake id ing pesen SOAP. Penting kanggo dicathet. Sampeyan njupuk file saka flowFile lan nganyari, aja lali yen sampeyan kudu sijine maneh, nganyari. Sampeyan uga kudu nyatet sing ora kabeh perpustakaan klebu. Bisa uga sampeyan isih kudu ngimpor salah sawijining lib. Kelemahane liyane yaiku naskah ing alun-alun iki cukup angel kanggo debug. Ana cara kanggo nyambung menyang NIFI JVM lan miwiti proses debugging. Secara pribadi, aku ngluncurake aplikasi lokal lan simulasi nampa file saka sesi kasebut. Aku uga nindakake debugging lokal. Kesalahan sing katon nalika ngemot skrip cukup gampang kanggo Google lan ditulis dening NIFI dhewe menyang 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)
}

Bener, ing kene kustomisasi kothak rampung. Sabanjure, file sing dianyari ditransfer menyang alun-alun, sing tanggung jawab kanggo ngirim file menyang server. Ing ngisor iki setelan kanggo kothak iki.

Kita nerangake cara ngirim pesen SOAP. Kita nulis ing ngendi. Sabanjure sampeyan kudu nuduhake yen iki SOAP.

Tambah sawetara properti kayata host lan tumindak (soapAction). Kita nyimpen lan mriksa. Sampeyan bisa ndeleng rincian liyane babagan carane ngirim panjalukan SOAP kene

Kita nyawang sawetara opsi kanggo nggunakake proses NIFI. Kepiye carane dheweke sesambungan lan apa manfaate sing nyata? Conto sing dianggep minangka tes lan rada beda karo apa sing kedadeyan ing pertempuran. Muga-muga artikel iki bakal migunani kanggo pangembang. Matur nuwun kawigatosanipun. Yen sampeyan duwe pitakon, tulisen. Aku bakal nyoba njawab.

Source: www.habr.com