Apache NIFI - Amaliyotdagi xususiyatlarning qisqacha ko'rinishi

kirish

Shunday bo'ldiki, hozirgi ish joyimda men ushbu texnologiya bilan tanishishga majbur bo'ldim. Men bir oz fon bilan boshlayman. Keyingi uchrashuvda jamoamiz bilan integratsiyani yaratishimiz kerakligi aytildi ma'lum tizim. Integratsiya deganda, bu taniqli tizim HTTP orqali bizga so'rovlarni ma'lum bir so'nggi nuqtaga yuborishi va biz, g'alati, SOAP xabari ko'rinishida javoblarni jo'natishimiz nazarda tutilgan edi. Hamma narsa oddiy va ahamiyatsiz ko'rinadi. Shundan kelib chiqadiki, sizga kerak ...

Maqsad

3 ta xizmat yarating. Ulardan birinchisi ma'lumotlar bazasini yangilash xizmatidir. Ushbu xizmat, uchinchi tomon tizimidan yangi ma'lumotlar kelganda, ma'lumotlar bazasidagi ma'lumotlarni yangilaydi va keyingi tizimga o'tkazish uchun CSV formatida faylni yaratadi. Ikkinchi xizmatning so'nggi nuqtasi deyiladi - FTP Transport xizmati, u uzatilgan faylni qabul qiladi, uni tasdiqlaydi va FTP orqali fayl xotirasiga qo'yadi. Uchinchi xizmat, iste'molchi ma'lumotlarini uzatish xizmati, birinchi ikkitasi bilan asenkron ishlaydi. U yuqorida muhokama qilingan faylni qabul qilish uchun uchinchi tomon tashqi tizimidan so'rov oladi, tayyor javob faylini oladi, uni o'zgartiradi (id, tavsif, linkToFile maydonlarini yangilaydi) va javobni SOAP xabari ko'rinishida yuboradi. Ya'ni, umumiy rasm quyidagicha: birinchi ikkita xizmat o'z ishini faqat yangilash uchun ma'lumotlar kelgandan keyin boshlaydi. Uchinchi xizmat doimiy ravishda ishlaydi, chunki ma'lumotlarning ko'plab iste'molchilari mavjud, har daqiqada ma'lumotlar uchun 1000 ga yaqin so'rovlar. Xizmatlar doimiy ravishda mavjud va ularning namunalari test, demo, ishlab chiqarishdan oldingi va ishlab chiqarish kabi turli muhitlarda joylashgan. Quyida ushbu xizmatlarning qanday ishlashi diagrammasi keltirilgan. Darhol tushuntirib beramanki, keraksiz murakkablikdan qochish uchun ba'zi tafsilotlar soddalashtirilgan.

Texnik chuqurlashtirish

Muammoni hal qilishni rejalashtirayotganda, biz Java-da Spring Framework, Nginx balanslagichi, Postgres ma'lumotlar bazasi va boshqa texnik va unchalik texnik bo'lmagan narsalardan foydalangan holda ilovalar yaratishga qaror qildik. Texnik yechimni ishlab chiqish vaqti bizga ushbu muammoni hal qilishning boshqa yondashuvlarini ko'rib chiqishga imkon berganligi sababli, bizning nigohimiz ma'lum doiralarda moda bo'lgan Apache NIFI texnologiyasiga tushdi. Darhol aytamanki, ushbu texnologiya bizga ushbu 3 ta xizmatni ko'rishga imkon berdi. Ushbu maqolada fayllarni tashish xizmati va iste'molchiga ma'lumotlarni uzatish xizmatining rivojlanishi tasvirlangan, ammo agar maqola foydali bo'lsa, men ma'lumotlar bazasidagi ma'lumotlarni yangilash xizmati haqida yozaman.

Bu nima

NIFI - bu ma'lumotlarni tez parallel yuklash va qayta ishlash, manbalar va transformatsiyalar uchun ko'p sonli plaginlar, konfiguratsiyalarni versiyalash va boshqalar uchun taqsimlangan arxitektura. Yaxshi bonus - bu foydalanish juda oson. GetFile, sendHttpRequest va boshqalar kabi arzimas jarayonlar kvadrat shaklida ifodalanishi mumkin. Har bir kvadrat jarayonni ifodalaydi, uning o'zaro ta'sirini quyidagi rasmda ko'rish mumkin. Jarayonni sozlashning o'zaro ta'siri bo'yicha batafsilroq hujjatlar yozilgan shu yerda , rus tilida gapiradiganlar uchun - shu yerda. Hujjatlar NIFI-ni qanday ochish va ishga tushirishni, shuningdek, kvadratlar deb ham ataladigan jarayonlarni qanday yaratishni mukammal tasvirlaydi.
Maqola yozish g'oyasi uzoq izlanish va olingan ma'lumotni ongli narsaga aylantirish, shuningdek, kelajakdagi ishlab chiquvchilar uchun hayotni biroz osonlashtirish istagidan keyin tug'ilgan.

misol

Kvadratlarning bir-biri bilan o'zaro ta'siriga misol keltiriladi. Umumiy sxema juda oddiy: Biz HTTP so'rovini olamiz (Nazariy jihatdan so'rovning asosiy qismidagi fayl bilan. NIFI imkoniyatlarini ko'rsatish uchun ushbu misolda so'rov mahalliy fayl xotirasidan faylni qabul qilish jarayonini boshlaydi. ), keyin biz so'rov olinganligi to'g'risida javob qaytaramiz, parallel ravishda FH dan faylni qabul qilish jarayoni va keyin uni FTP orqali FH ga ko'chirish jarayoni. Jarayonlar flowFile deb ataladigan fayl orqali bir-biri bilan o'zaro ta'sir qilishini aniqlashtirish kerak. Bu atributlar va tarkibni saqlaydigan NIFIdagi asosiy ob'ektdir. Kontent - bu oqim fayli bilan ifodalanadigan ma'lumotlar. Ya'ni, taxminan aytganda, agar siz bir kvadratdan faylni qabul qilsangiz va uni boshqasiga o'tkazsangiz, tarkib sizning faylingiz bo'ladi.

Ko'rib turganingizdek, bu rasmda umumiy jarayon ko'rsatilgan. HandleHttpRequest - so'rovlarni qabul qiladi, ReplaceText - javob tanasini yaratadi, HandleHttpResponse - javob yuboradi. FetchFile - faylni saqlash joyidan faylni oladi, uni PutSftp kvadratiga o'tkazadi - bu faylni ko'rsatilgan manzilda FTP-ga joylashtiradi. Endi bu jarayon haqida ko'proq.

Bunday holda, so'rov hamma narsaning boshlanishidir. Keling, uning konfiguratsiya parametrlarini ko'rib chiqaylik.

Bu erda hamma narsa juda ahamiyatsiz, StandardHttpContextMap bundan mustasno - bu so'rovlarni yuborish va qabul qilish imkonini beruvchi xizmat turi. Batafsilroq va hatto misollar bilan ko'rishingiz mumkin - shu yerda

Keyin kvadratning ReplaceText konfiguratsiya parametrlarini ko'rib chiqamiz. ReplacementValue-ga e'tibor qaratish lozim - bu foydalanuvchiga javob shaklida qaytariladi. Sozlamalarda siz ro'yxatga olish darajasini sozlashingiz mumkin, jurnallarni ko'rishingiz mumkin {nifi-1.9.2/logsni ochgan joyingiz}/nifi-XNUMX/logs, shuningdek, muvaffaqiyatsizlik/muvaffaqiyat parametrlari mavjud - bu parametrlarga asoslanib, jarayonni bir butun sifatida tartibga solishingiz mumkin. . Ya'ni, matnni muvaffaqiyatli qayta ishlash holatida foydalanuvchiga javob yuborish jarayoni chaqiriladi va boshqa holatda biz shunchaki muvaffaqiyatsiz jarayonni qayd qilamiz.

HandleHttpResponse xususiyatlarida javob muvaffaqiyatli yaratilgan holatdan tashqari, ayniqsa qiziq narsa yo'q.

Biz so'rov va javobni saralab oldik - keling, faylni qabul qilish va uni FTP serveriga joylashtirishga o'tamiz. FetchFile - faylni sozlamalarda ko'rsatilgan yo'lda qabul qiladi va uni keyingi jarayonga o'tkazadi.

Va keyin PutSftp kvadrati - faylni fayl xotirasiga joylashtiradi. Quyida biz konfiguratsiya parametrlarini ko'rishimiz mumkin.

Har bir kvadrat ishga tushirilishi kerak bo'lgan alohida jarayon ekanligiga e'tibor qaratish lozim. Biz murakkab sozlashni talab qilmaydigan eng oddiy misolni ko'rib chiqdik. Keyinchalik, biz jarayonni biroz murakkabroq ko'rib chiqamiz, bu erda biz oluklarga bir oz yozamiz.

Keyinchalik murakkab misol

Iste'molchiga ma'lumotlarni uzatish xizmati SOAP xabarini o'zgartirish jarayoni tufayli biroz murakkabroq bo'lib chiqdi. Umumiy jarayon quyidagi rasmda ko'rsatilgan.

Bu erda g'oya ham unchalik murakkab emas: biz iste'molchidan unga ma'lumot kerakligi haqida so'rov oldik, u xabar olgani haqida javob yubordik, javob faylini qabul qilish jarayonini boshladik, keyin uni ma'lum bir mantiq bilan tahrir qildik va keyin faylni iste'molchiga SOAP xabari ko'rinishida serverga uzatdi.

O'ylaymanki, biz yuqorida ko'rgan kvadratlarni qayta ta'riflashning hojati yo'q - keling, to'g'ridan-to'g'ri yangilariga o'taylik. Agar biron bir faylni tahrirlashingiz kerak bo'lsa va oddiy ReplaceText tipidagi kvadratchalar mos kelmasa, siz o'zingizning skriptingizni yozishingiz kerak bo'ladi. Buni ExecuteGroogyScript kvadrati yordamida amalga oshirish mumkin. Uning sozlamalari quyida keltirilgan.

Ushbu kvadratga skriptni yuklashning ikkita varianti mavjud. Birinchisi, faylni skript bilan yuklab olishdir. Ikkinchisi skriptni scriptBody-ga kiritish orqali. Men bilishimcha, executeScript kvadrati bir nechta tillarni qo'llab-quvvatlaydi - ulardan biri ajoyib. Men java ishlab chiquvchilarni hafsalasi pir bo'laman - bunday kvadratlarda java-da skript yoza olmaysiz. Haqiqatan ham xohlaydiganlar uchun siz o'zingizning shaxsiy kvadratingizni yaratishingiz va uni NIFI tizimiga qo'shishingiz kerak. Bu butun operatsiya daf bilan uzoq raqs bilan birga keladi, biz ushbu maqolada ko'rib chiqmaymiz. Men ajoyib tilni tanladim. Quyida SOAP xabaridagi identifikatorni bosqichma-bosqich yangilaydigan sinov skripti mavjud. Shuni ta'kidlash kerak. Siz faylni flowFile'dan olasiz va uni yangilaysiz, uni yangilangan joyga qo'yish kerakligini unutmang. Shuni ham ta'kidlash kerakki, barcha kutubxonalar kiritilgan emas. Siz hali ham liblardan birini import qilishingiz kerak bo'lishi mumkin. Yana bir salbiy tomoni shundaki, bu kvadratdagi skriptni disk raskadrovka qilish juda qiyin. NIFI JVM ga ulanish va disk raskadrovka jarayonini boshlash usuli mavjud. Shaxsan men mahalliy dasturni ishga tushirdim va sessiyadan faylni olishni simulyatsiya qildim. Men ham mahalliy disk raskadrovka qildim. Skriptni yuklashda paydo bo'ladigan xatolar Google uchun juda oson va NIFI tomonidan jurnalga yoziladi.

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

Aslida, bu erda kvadratni sozlash tugaydi. Keyinchalik, yangilangan fayl faylni serverga yuborish uchun javobgar bo'lgan kvadratga o'tkaziladi. Quyida ushbu kvadrat uchun sozlamalar mavjud.

Biz SOAP xabarini uzatish usulini tasvirlaymiz. Biz qayerda yozamiz. Keyin bu SOAP ekanligini ko'rsatishingiz kerak.

Xost va harakat (soapAction) kabi bir nechta xususiyatlarni qo'shing. Saqlaymiz va tekshiramiz. SOAP so'rovlarini qanday yuborish haqida batafsil ma'lumotni ko'rishingiz mumkin shu yerda

NIFI jarayonlaridan foydalanishning bir nechta variantlarini ko'rib chiqdik. Ular qanday munosabatda bo'lishadi va ularning haqiqiy foydasi nimada? Ko'rib chiqilgan misollar sinovdir va jangda sodir bo'layotgan voqealardan biroz farq qiladi. Umid qilamanki, ushbu maqola ishlab chiquvchilar uchun ozgina foydali bo'ladi. E'tiboringiz uchun rahmat. Savollaringiz bo'lsa yozing. Men javob berishga harakat qilaman.

Manba: www.habr.com