Apache NIFI - Gambaran Keseluruhan Ringkas Ciri dalam Amalan

Pengenalan

Kebetulan di tempat kerja saya sekarang saya terpaksa berkenalan dengan teknologi ini. Saya akan mulakan dengan sedikit latar belakang. Pada mesyuarat seterusnya, pasukan kami diberitahu bahawa kami perlu mewujudkan integrasi dengan sistem yang diketahui. Dengan penyepaduan ini bermakna bahawa sistem yang terkenal ini akan menghantar permintaan kepada kami melalui HTTP ke titik akhir tertentu, dan kami, anehnya, akan menghantar semula respons dalam bentuk mesej SOAP. Semuanya nampak mudah dan remeh. Dari sini, anda perlu...

Petugas

Buat 3 perkhidmatan. Yang pertama ialah Perkhidmatan Kemas Kini Pangkalan Data. Perkhidmatan ini, apabila data baharu tiba daripada sistem pihak ketiga, mengemas kini data dalam pangkalan data dan menjana fail dalam format CSV untuk memindahkannya ke sistem seterusnya. Titik akhir perkhidmatan kedua dipanggil - Perkhidmatan Pengangkutan FTP, yang menerima fail yang dipindahkan, mengesahkannya, dan meletakkannya dalam storan fail melalui FTP. Perkhidmatan ketiga, perkhidmatan pemindahan data pengguna, berfungsi secara tidak segerak dengan dua yang pertama. Ia menerima permintaan daripada sistem luaran pihak ketiga untuk menerima fail yang dibincangkan di atas, mengambil fail respons sedia, mengubah suainya (mengemas kini id, penerangan, medan linkToFile) dan menghantar respons dalam bentuk mesej SOAP. Iaitu, gambaran keseluruhan adalah seperti berikut: dua perkhidmatan pertama memulakan kerja mereka hanya apabila data untuk kemas kini telah tiba. Perkhidmatan ketiga berfungsi secara berterusan kerana terdapat ramai pengguna maklumat, kira-kira 1000 permintaan untuk data seminit. Perkhidmatan tersedia secara berterusan dan kejadiannya terletak dalam persekitaran yang berbeza, seperti ujian, tunjuk cara, pra-pengeluaran dan produk. Di bawah ialah gambar rajah cara perkhidmatan ini berfungsi. Biar saya jelaskan dengan segera bahawa beberapa butiran telah dipermudahkan untuk mengelakkan kerumitan yang tidak perlu.

Pendalaman Teknikal

Apabila merancang penyelesaian kepada masalah tersebut, kami mula-mula memutuskan untuk membuat aplikasi dalam Java menggunakan rangka kerja Spring, pengimbang Nginx, pangkalan data Postgres dan perkara teknikal dan bukan teknikal yang lain. Memandangkan masa untuk membangunkan penyelesaian teknikal membolehkan kami mempertimbangkan pendekatan lain untuk menyelesaikan masalah ini, pandangan kami jatuh pada teknologi Apache NIFI, yang bergaya dalam kalangan tertentu. Saya akan nyatakan dengan segera bahawa teknologi ini membolehkan kami melihat 3 perkhidmatan ini. Artikel ini akan menerangkan pembangunan perkhidmatan pengangkutan fail dan perkhidmatan pemindahan data kepada pengguna, tetapi jika artikel itu berguna, saya akan menulis tentang perkhidmatan untuk mengemas kini data dalam pangkalan data.

Apa ini

NIFI ialah seni bina teragih untuk pemuatan dan pemprosesan data selari pantas, sejumlah besar pemalam untuk sumber dan transformasi, versi konfigurasi dan banyak lagi. Bonus yang bagus ialah ia sangat mudah digunakan. Proses remeh seperti getFile, sendHttpRequest dan lain-lain boleh diwakili sebagai petak. Setiap segi empat sama mewakili proses, interaksi yang boleh dilihat dalam rajah di bawah. Dokumentasi yang lebih terperinci tentang interaksi persediaan proses telah ditulis di sini , bagi mereka yang berbahasa Rusia - di sini. Dokumentasi menerangkan dengan sempurna cara membongkar dan menjalankan NIFI, serta cara mencipta proses, juga dikenali sebagai petak
Idea untuk menulis artikel lahir selepas pencarian panjang dan penstrukturan maklumat yang diterima menjadi sesuatu yang sedar, serta keinginan untuk menjadikan hidup lebih mudah untuk pemaju masa depan..

Contoh

Contoh cara petak berinteraksi antara satu sama lain dipertimbangkan. Skim umum agak mudah: Kami menerima permintaan HTTP (Secara teori, dengan fail dalam badan permintaan. Untuk menunjukkan keupayaan NIFI, dalam contoh ini permintaan memulakan proses menerima fail daripada storan fail tempatan ), kemudian kami menghantar semula respons bahawa permintaan telah diterima, selari dengan proses menerima fail daripada FH dan kemudian proses mengalihkannya melalui FTP ke FH. Perlu dijelaskan bahawa proses berinteraksi antara satu sama lain melalui apa yang dipanggil flowFile. Ini ialah entiti asas dalam NIFI yang menyimpan atribut dan kandungan. Kandungan ialah data yang diwakili oleh fail strim. Iaitu, secara kasarnya, jika anda menerima fail dari satu petak dan memindahkannya ke petak lain, kandungan itu akan menjadi fail anda.

Seperti yang anda lihat, gambar ini menunjukkan proses umum. HandleHttpRequest - menerima permintaan, ReplaceText - menjana badan tindak balas, HandleHttpResponse - menghantar respons. FetchFile - menerima fail daripada storan fail, memindahkannya ke segi empat sama PutSftp - meletakkan fail ini pada FTP, pada alamat yang ditentukan. Sekarang lebih lanjut mengenai proses ini.

Dalam kes ini, permintaan adalah permulaan segala-galanya. Mari lihat parameter konfigurasinya.

Segala-galanya di sini agak remeh kecuali StandardHttpContextMap - ini adalah sejenis perkhidmatan yang membolehkan anda menghantar dan menerima permintaan. Dengan lebih terperinci dan juga dengan contoh, anda boleh lihat - di sini

Seterusnya, mari lihat pada parameter konfigurasi ReplaceText bagi segi empat sama. Perlu diberi perhatian kepada ReplacementValue - inilah yang akan dikembalikan kepada pengguna dalam bentuk respons. Dalam tetapan anda boleh melaraskan tahap pengelogan, anda boleh melihat log {where you unpacked nifi}/nifi-1.9.2/logs, terdapat juga parameter kegagalan/kejayaan - berdasarkan parameter ini anda boleh mengawal proses secara keseluruhan . Iaitu, dalam kes pemprosesan teks yang berjaya, proses menghantar respons kepada pengguna akan dipanggil, dan dalam kes lain kami hanya akan log proses yang tidak berjaya.

Tiada apa-apa yang menarik terutamanya dalam sifat HandleHttpResponse kecuali status apabila respons berjaya dibuat.

Kami telah menyelesaikan permintaan dan respons - mari kita teruskan untuk menerima fail dan meletakkannya pada pelayan FTP. FetchFile - menerima fail pada laluan yang ditentukan dalam tetapan dan menghantarnya ke proses seterusnya.

Dan kemudian petak PutSftp - meletakkan fail dalam storan fail. Kita boleh melihat parameter konfigurasi di bawah.

Perlu diberi perhatian kepada fakta bahawa setiap persegi adalah proses berasingan yang mesti dilancarkan. Kami melihat contoh paling mudah yang tidak memerlukan sebarang penyesuaian yang kompleks. Seterusnya, kita akan melihat prosesnya sedikit lebih rumit, di mana kita akan menulis sedikit pada alur.

Contoh yang lebih kompleks

Perkhidmatan pemindahan data kepada pengguna ternyata menjadi lebih rumit kerana proses mengubah suai mesej SOAP. Proses umum ditunjukkan dalam rajah di bawah.

Di sini idea itu juga tidak terlalu rumit: kami menerima permintaan daripada pengguna bahawa dia memerlukan data, menghantar respons bahawa dia telah menerima mesej, memulakan proses menerima fail respons, kemudian mengeditnya dengan logik tertentu, dan kemudian memindahkan fail kepada pengguna dalam bentuk mesej SOAP ke pelayan.

Saya rasa tidak perlu untuk menerangkan lagi petak yang kita lihat di atas - mari teruskan ke petak baharu. Jika anda perlu mengedit mana-mana fail dan petak biasa jenis ReplaceText tidak sesuai, anda perlu menulis skrip anda sendiri. Ini boleh dilakukan menggunakan petak ExecuteGroogyScript. Tetapannya dibentangkan di bawah.

Terdapat dua pilihan untuk memuatkan skrip ke dalam petak ini. Yang pertama ialah dengan memuat turun fail dengan skrip. Yang kedua ialah dengan memasukkan skrip ke dalam scriptBody. Setakat yang saya tahu, petak executeScript menyokong beberapa bahasa - salah satunya adalah groovy. Saya akan mengecewakan pembangun java - anda tidak boleh menulis skrip dalam java dalam petak sedemikian. Bagi mereka yang benar-benar mahu, anda perlu membuat petak tersuai anda sendiri dan menambahnya pada sistem NIFI. Seluruh operasi ini disertai dengan tarian yang agak panjang dengan rebana, yang tidak akan kami bincangkan dalam artikel ini. Saya memilih bahasa groovy. Di bawah ialah skrip ujian yang hanya mengemas kini id secara berperingkat dalam mesej SOAP. Adalah penting untuk diperhatikan. Anda mengambil fail dari flowFile dan mengemas kininya, jangan lupa bahawa anda perlu meletakkannya semula di sana, dikemas kini. Perlu diingatkan juga bahawa tidak semua perpustakaan disertakan. Ia mungkin berlaku bahawa anda masih perlu mengimport salah satu lib. Kelemahan lain ialah skrip dalam petak ini agak sukar untuk dinyahpepijat. Terdapat cara untuk menyambung ke NIFI JVM dan memulakan proses penyahpepijatan. Secara peribadi, saya melancarkan aplikasi tempatan dan mensimulasikan penerimaan fail daripada sesi tersebut. Saya juga melakukan penyahpepijatan secara tempatan. Ralat yang muncul semasa memuatkan skrip agak mudah untuk Google dan ditulis oleh NIFI sendiri ke 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)
}

Sebenarnya, di sinilah penyesuaian petak itu berakhir. Seterusnya, fail yang dikemas kini dipindahkan ke petak, yang bertanggungjawab untuk menghantar fail ke pelayan. Di bawah ialah tetapan untuk petak ini.

Kami menerangkan kaedah yang mana mesej SOAP akan dihantar. Kami menulis di mana. Seterusnya anda perlu menunjukkan bahawa ini adalah SOAP.

Tambahkan beberapa sifat seperti hos dan tindakan (soapAction). Kami simpan dan semak. Anda boleh melihat butiran lanjut tentang cara menghantar permintaan SOAP di sini

Kami melihat beberapa pilihan untuk menggunakan proses NIFI. Bagaimanakah mereka berinteraksi dan apakah faedah sebenar mereka? Contoh yang dipertimbangkan adalah ujian dan sedikit berbeza daripada apa yang sebenarnya berlaku dalam pertempuran. Saya harap artikel ini akan berguna sedikit untuk pembangun. Terima kasih kerana memberi perhatian. Jika anda mempunyai sebarang soalan, tulis. Saya akan cuba menjawab.

Sumber: www.habr.com